furtka/tests/test_wireguard.py

"""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}