furtka-gateway/tests/test_wgeasy.py

import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import pytest

from control_plane.wgeasy import WgEasyClient, WgEasyError

ADMIN_PASSWORD = "hunter2"

FAKE_CONF = """[Interface]
PrivateKey = client-private-key==
Address = 10.8.0.5/24

[Peer]
PublicKey = server-public-key==
AllowedIPs = 10.8.0.0/24
PersistentKeepalive = 25
Endpoint = gateway.example.com:51820
"""

CLIENT_RECORD = {
    "id": "11111111-1111-1111-1111-111111111111",
    "name": "box-1",
    "enabled": True,
    "address": "10.8.0.5",
    "publicKey": "client-public-key==",
    "createdAt": "2026-01-01T00:00:00.000Z",
}


class FakeWgEasyHandler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass

    def _json(self, status, payload):
        body = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _check_auth(self) -> bool:
        ok = self.headers.get("Authorization") == ADMIN_PASSWORD
        if not ok:
            self._json(401, {"error": "Incorrect Password"})
        return ok

    def do_POST(self):
        self.server.requests.append(("POST", self.path))
        if not self._check_auth():
            return
        if self.path == "/api/wireguard/client":
            self._json(200, {"success": True})
            return
        self._json(404, {"error": "not found"})

    def do_GET(self):
        self.server.requests.append(("GET", self.path))
        if not self._check_auth():
            return
        if self.path == "/api/wireguard/client":
            self._json(200, [CLIENT_RECORD])
            return
        if self.path == f"/api/wireguard/client/{CLIENT_RECORD['id']}/configuration":
            body = FAKE_CONF.encode()
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        self._json(404, {"error": "not found"})

    def do_DELETE(self):
        self.server.requests.append(("DELETE", self.path))
        if not self._check_auth():
            return
        if self.path == f"/api/wireguard/client/{CLIENT_RECORD['id']}":
            self._json(200, {"success": True})
            return
        self._json(404, {"error": "not found"})


@pytest.fixture
def fake_wgeasy():
    server = ThreadingHTTPServer(("127.0.0.1", 0), FakeWgEasyHandler)
    server.requests = []
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield server
    finally:
        server.shutdown()
        thread.join()


def _client(server, password=ADMIN_PASSWORD):
    port = server.server_address[1]
    return WgEasyClient(f"http://127.0.0.1:{port}", password)


def test_create_client_lists_and_parses_configuration(fake_wgeasy):
    client = _client(fake_wgeasy)

    peer = client.create_client("box-1")

    assert peer["id"] == CLIENT_RECORD["id"]
    assert peer["public_key"] == "client-public-key=="
    assert peer["private_key"] == "client-private-key=="
    assert peer["address"] == "10.8.0.5/24"
    assert peer["server_public_key"] == "server-public-key=="
    assert peer["endpoint"] == "gateway.example.com:51820"
    assert peer["allowed_ips"] == "10.8.0.0/24"

    paths_hit = [(m, p) for m, p in fake_wgeasy.requests]
    assert ("POST", "/api/wireguard/client") in paths_hit
    assert ("GET", "/api/wireguard/client") in paths_hit


def test_every_request_carries_the_admin_password_header(fake_wgeasy):
    client = _client(fake_wgeasy, password="wrong-password")
    with pytest.raises(WgEasyError):
        client.create_client("box-1")


def test_create_client_raises_if_not_found_after_creation(fake_wgeasy):
    client = _client(fake_wgeasy)
    with pytest.raises(WgEasyError):
        client.create_client("some-other-name-not-in-fake-list")


def test_delete_client_swallows_not_found(fake_wgeasy):
    client = _client(fake_wgeasy)
    client.delete_client("does-not-exist")  # 404 from fake server — must not raise


def test_delete_client_calls_endpoint(fake_wgeasy):
    client = _client(fake_wgeasy)
    client.delete_client(CLIENT_RECORD["id"])
    assert ("DELETE", f"/api/wireguard/client/{CLIENT_RECORD['id']}") in [
        (m, p) for m, p in fake_wgeasy.requests
    ]


def test_bad_endpoint_raises_wgeasy_error(fake_wgeasy):
    client = _client(fake_wgeasy)
    with pytest.raises(WgEasyError):
        client._request("GET", "/api/does-not-exist")