196 lines
6.7 KiB
Python
196 lines
6.7 KiB
Python
"""Deep tests for prisma.vpn — exercises all lifecycle functions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import prisma.vpn as vpn
|
|
from prisma.vpn import (
|
|
_cloud_init,
|
|
_delete_account_ssh_key,
|
|
_ensure_account_ssh_key,
|
|
_gen_ssh_key,
|
|
_tunnel_running,
|
|
down,
|
|
status,
|
|
up,
|
|
)
|
|
|
|
|
|
class TestCloudInit:
|
|
def test_generates_config(self):
|
|
result = _cloud_init("ssh-ed25519 AAAA test")
|
|
assert "#cloud-config" in result
|
|
assert "dante-server" in result
|
|
assert "ssh-ed25519" in result
|
|
|
|
|
|
class TestGenSshKey:
|
|
@patch("prisma.vpn.subprocess.run")
|
|
def test_generates_key(self, mock_run, tmp_path):
|
|
key_path = tmp_path / "id_ed25519"
|
|
|
|
# _gen_ssh_key unlinks existing then runs ssh-keygen, which creates the file
|
|
# Mock the subprocess but also create the pubkey file it would produce
|
|
def fake_keygen(*a, **kw):
|
|
key_path.write_text("private")
|
|
key_path.with_suffix(".pub").write_text("ssh-ed25519 AAAA test")
|
|
return MagicMock(returncode=0)
|
|
|
|
mock_run.side_effect = fake_keygen
|
|
result = _gen_ssh_key(key_path)
|
|
assert "ssh-ed25519" in result
|
|
|
|
|
|
class TestEnsureAccountSshKey:
|
|
def test_reuses_existing(self):
|
|
client = MagicMock()
|
|
client.ssh_keys.list.return_value = {
|
|
"ssh_keys": [{"id": 42, "public_key": "ssh-ed25519 AAAA test"}]
|
|
}
|
|
result = _ensure_account_ssh_key(client, "ssh-ed25519 AAAA test", "name")
|
|
assert result == 42
|
|
|
|
def test_creates_new(self):
|
|
client = MagicMock()
|
|
client.ssh_keys.list.return_value = {"ssh_keys": []}
|
|
client.ssh_keys.create.return_value = {"ssh_key": {"id": 99}}
|
|
result = _ensure_account_ssh_key(client, "ssh-ed25519 BBBB new", "name")
|
|
assert result == 99
|
|
|
|
|
|
class TestDeleteAccountSshKey:
|
|
def test_deletes(self):
|
|
client = MagicMock()
|
|
_delete_account_ssh_key(client, 42)
|
|
client.ssh_keys.delete.assert_called_once()
|
|
|
|
def test_handles_error(self):
|
|
client = MagicMock()
|
|
client.ssh_keys.delete.side_effect = Exception("gone")
|
|
_delete_account_ssh_key(client, 42) # should not raise
|
|
|
|
|
|
class TestTunnelRunning:
|
|
def test_no_pid_file(self):
|
|
with patch("prisma.vpn._TUNNEL_PID") as mock_path:
|
|
mock_path.is_file.return_value = False
|
|
assert _tunnel_running() is False
|
|
|
|
|
|
class TestUp:
|
|
@patch("prisma.vpn._do_client")
|
|
@patch("prisma.vpn._gen_ssh_key", return_value="ssh-ed25519 AAAA")
|
|
@patch("prisma.vpn._ensure_account_ssh_key", return_value=1)
|
|
@patch("prisma.vpn._DROPLET_JSON")
|
|
@patch("prisma.vpn._STATE_DIR")
|
|
def test_raises_if_tracked(
|
|
self, mock_state, mock_json, mock_key_id, mock_gen, mock_client
|
|
):
|
|
mock_json.exists.return_value = True
|
|
with pytest.raises(RuntimeError, match="already tracked"):
|
|
up()
|
|
|
|
|
|
class TestDown:
|
|
@patch("prisma.vpn.detach_zotero_proxy")
|
|
@patch("prisma.vpn._tunnel_stop")
|
|
@patch("prisma.vpn._DROPLET_JSON")
|
|
def test_noop_if_not_tracked(self, mock_json, mock_stop, mock_detach):
|
|
mock_json.exists.return_value = False
|
|
result = down()
|
|
assert result == {"status": "nothing-to-do"}
|
|
|
|
|
|
class TestStatus:
|
|
@patch("prisma.vpn._DROPLET_JSON")
|
|
def test_down_if_not_tracked(self, mock_json):
|
|
mock_json.exists.return_value = False
|
|
result = status()
|
|
assert result["status"] == "down"
|
|
|
|
|
|
class TestDownOrdering:
|
|
"""#664: the droplet (billable) is destroyed first; a local failure in
|
|
the Zotero detach must not skip the destroy, the key delete or the
|
|
state wipe."""
|
|
|
|
def _state(self, tmp_path):
|
|
dj = tmp_path / "droplet.json"
|
|
dj.write_text(
|
|
'{"id": 565110236, "ssh_key_id": 55643846, "public_ip": "1.2.3.4"}'
|
|
)
|
|
files = [
|
|
tmp_path / n for n in ("key", "key.pub", "known_hosts", "env", "tunnel.pid")
|
|
]
|
|
for f in files:
|
|
f.write_text("x")
|
|
return dj, files
|
|
|
|
@patch("prisma.vpn._do_client")
|
|
@patch("prisma.vpn._delete_account_ssh_key")
|
|
@patch("prisma.vpn._tunnel_stop")
|
|
@patch("prisma.vpn.detach_zotero_proxy", side_effect=PermissionError("prefs.js"))
|
|
def test_detach_failure_does_not_skip_destroy(
|
|
self, mc_detach, mc_stop, mc_delkey, mc_client, tmp_path
|
|
):
|
|
dj, files = self._state(tmp_path)
|
|
client = MagicMock()
|
|
mc_client.return_value = client
|
|
with (
|
|
patch.object(vpn, "_DROPLET_JSON", dj),
|
|
patch.object(vpn, "_SSH_KEY", files[0]),
|
|
patch.object(vpn, "_KNOWN_HOSTS", files[2]),
|
|
patch.object(vpn, "_ENV_FILE", files[3]),
|
|
patch.object(vpn, "_TUNNEL_PID", files[4]),
|
|
):
|
|
result = down()
|
|
assert result == {"status": "destroyed", "droplet_id": 565110236}
|
|
client.droplets.destroy.assert_called_once_with(droplet_id=565110236)
|
|
mc_delkey.assert_called_once_with(client, 55643846)
|
|
assert not dj.exists() and not any(f.exists() for f in files)
|
|
|
|
@patch("prisma.vpn._do_client")
|
|
@patch("prisma.vpn._delete_account_ssh_key")
|
|
@patch("prisma.vpn._tunnel_stop")
|
|
@patch("prisma.vpn.detach_zotero_proxy")
|
|
def test_destroy_happens_before_local_cleanup(
|
|
self, mc_detach, mc_stop, mc_delkey, mc_client, tmp_path
|
|
):
|
|
dj, files = self._state(tmp_path)
|
|
order: list[str] = []
|
|
client = MagicMock()
|
|
client.droplets.destroy.side_effect = lambda **k: order.append("destroy")
|
|
mc_client.return_value = client
|
|
mc_stop.side_effect = lambda: order.append("tunnel")
|
|
mc_detach.side_effect = lambda: order.append("detach")
|
|
with (
|
|
patch.object(vpn, "_DROPLET_JSON", dj),
|
|
patch.object(vpn, "_SSH_KEY", files[0]),
|
|
patch.object(vpn, "_KNOWN_HOSTS", files[2]),
|
|
patch.object(vpn, "_ENV_FILE", files[3]),
|
|
patch.object(vpn, "_TUNNEL_PID", files[4]),
|
|
):
|
|
down()
|
|
assert order == ["destroy", "tunnel", "detach"]
|
|
|
|
|
|
class TestRestorePrefsPermission:
|
|
@patch("prisma.vpn.subprocess.run")
|
|
def test_falls_back_to_sudo_cp(self, mc_run, tmp_path):
|
|
backup = tmp_path / "prefs.js.bak"
|
|
prefs = tmp_path / "prefs.js"
|
|
backup.write_text("bak")
|
|
prefs.write_text("cur")
|
|
with (
|
|
patch.object(vpn, "_ZOTERO_PREFS_BACKUP", backup),
|
|
patch.object(vpn, "_ZOTERO_PREFS", prefs),
|
|
patch("shutil.copy2", side_effect=PermissionError),
|
|
):
|
|
vpn._restore_zotero_prefs()
|
|
cmds = [c.args[0][:2] for c in mc_run.call_args_list]
|
|
assert ["sudo", "cp"] in cmds
|
|
assert not backup.exists()
|