"""Tests for furtka.wireguard — tunnel-config write/remove + systemctl toggle. Same shape as test_https.py: stub the systemctl call so we assert the conf file is written/removed and that a reload failure rolls the conf file back to its pre-call state. """ import subprocess import pytest from furtka import wireguard WG = { "private_key": "cHJpdmF0ZS1rZXk=", "address": "10.8.0.2/24", "server_public_key": "c2VydmVyLXB1YmxpYy1rZXk=", "endpoint": "gateway.example.com:51820", "allowed_ips": "10.8.0.0/24", } def test_conf_content_shape(): content = wireguard._conf_content(WG) assert "[Interface]" in content assert "PrivateKey = cHJpdmF0ZS1rZXk=" in content assert "Address = 10.8.0.2/24" in content assert "[Peer]" in content assert "PublicKey = c2VydmVyLXB1YmxpYy1rZXk=" in content assert "Endpoint = gateway.example.com:51820" in content assert "AllowedIPs = 10.8.0.0/24" in content assert "PersistentKeepalive = 25" in content def test_enable_requires_wg_config(tmp_path): with pytest.raises(ValueError, match="wg config is required"): wireguard.set_tunnel_enabled(True, conf_path=tmp_path / "furtka-gw.conf") def test_enable_writes_conf_and_calls_reload(tmp_path): conf_path = tmp_path / "furtka-gw.conf" calls = [] def fake_reload(action): calls.append(action) result = wireguard.set_tunnel_enabled(True, wg=WG, conf_path=conf_path, reload=fake_reload) assert result is True assert calls == ["enable"] content = conf_path.read_text() assert "PrivateKey = cHJpdmF0ZS1rZXk=" in content assert oct(conf_path.stat().st_mode)[-3:] == "600" def test_disable_removes_conf_and_calls_reload(tmp_path): conf_path = tmp_path / "furtka-gw.conf" conf_path.write_text(wireguard._conf_content(WG)) calls = [] result = wireguard.set_tunnel_enabled( False, conf_path=conf_path, reload=lambda action: calls.append(action) ) assert result is False assert calls == ["disable"] assert not conf_path.exists() def test_disable_is_idempotent_when_already_off(tmp_path): conf_path = tmp_path / "furtka-gw.conf" result = wireguard.set_tunnel_enabled(False, conf_path=conf_path, reload=lambda action: None) assert result is False assert not conf_path.exists() def test_reload_failure_rolls_back_enable(tmp_path): conf_path = tmp_path / "furtka-gw.conf" def failing_reload(action): raise subprocess.CalledProcessError(1, ["systemctl"], stderr="bad config") with pytest.raises(wireguard.WireguardError, match="systemctl failed: bad config"): wireguard.set_tunnel_enabled(True, wg=WG, conf_path=conf_path, reload=failing_reload) # No conf existed before this call, so none should exist after rollback. assert not conf_path.exists() def test_reload_failure_rolls_back_disable(tmp_path): conf_path = tmp_path / "furtka-gw.conf" original = wireguard._conf_content(WG) conf_path.write_text(original) def failing_reload(action): raise subprocess.CalledProcessError(1, ["systemctl"], stderr="bad config") with pytest.raises(wireguard.WireguardError): wireguard.set_tunnel_enabled(False, conf_path=conf_path, reload=failing_reload) assert conf_path.read_text() == original def test_systemctl_missing_raises_and_rolls_back(tmp_path): conf_path = tmp_path / "furtka-gw.conf" def missing_systemctl(action): raise FileNotFoundError(2, "No such file", "systemctl") with pytest.raises(wireguard.WireguardError, match="systemctl not available"): wireguard.set_tunnel_enabled(True, wg=WG, conf_path=conf_path, reload=missing_systemctl) assert not conf_path.exists() def test_status_not_configured(tmp_path): s = wireguard.status(conf_path=tmp_path / "does-not-exist.conf") assert s == {"configured": False, "latest_handshake": None} def test_status_configured_reports_handshake(tmp_path): conf_path = tmp_path / "furtka-gw.conf" conf_path.write_text(wireguard._conf_content(WG)) s = wireguard.status(conf_path=conf_path, wg_show=lambda: "1234567890") assert s == {"configured": True, "latest_handshake": "1234567890"} def test_status_configured_but_wg_tools_missing(tmp_path): conf_path = tmp_path / "furtka-gw.conf" conf_path.write_text(wireguard._conf_content(WG)) def missing_wg(): raise FileNotFoundError(2, "No such file", "wg") s = wireguard.status(conf_path=conf_path, wg_show=missing_wg) assert s == {"configured": True, "latest_handshake": None}