tests/test_app.py

@@ -104,6 +104,37 @@ def test_wireguard_status_command_uses_sudo_and_accepts_off_keepalive():
     assert '0 if fields[7] == "off" else int(fields[7])' in script
 
 
+def test_wireguard_peer_remove_is_scoped_and_audited(monkeypatch):
+    calls = []
+    monkeypatch.setattr(app, "_find_inventory_host", lambda host: {"user": "debian", "ip": "141.94.237.199"})
+    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=30: (calls.append((host, command, timeout)) or (0, json.dumps({"status": "removed", "removed_routes": ["10.7.1.0/24"]}), "")))
+
+    with TestClient(app.app) as client:
+        response = client.request(
+            "DELETE",
+            "/network/wireguard/guck-vps/peer",
+            headers={"Authorization": "Bearer test-token"},
+            json={"public_key": "A" * 43 + "=", "expected_allowed_ip": "10.7.1.0/24", "dry_run": False},
+        )
+
+    assert response.status_code == 200
+    assert response.json()["status"] == "removed"
+    assert calls[0][0] == "debian@141.94.237.199"
+    assert calls[0][2] == 45
+
+
+def test_wireguard_peer_remove_rejects_non_allowlisted_host(monkeypatch):
+    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("SSH must not run")))
+    with TestClient(app.app) as client:
+        response = client.request(
+            "DELETE",
+            "/network/wireguard/node7/peer",
+            headers={"Authorization": "Bearer test-token"},
+            json={"public_key": "A" * 43 + "=", "expected_allowed_ip": "10.7.1.0/24", "dry_run": True},
+        )
+    assert response.status_code == 403
+
+
 def test_tts_generate_returns_cloned_wav(monkeypatch):
     captured = {}