Files
vacuum-wall/tests/test_api.py
T
mteehan d1ab717c0f refactor: unify project structure, improve security, and enhance deployment
- Fix WireGuard private key leak in API responses and config updates
- Update systemd service to serve from repo root with adjusted sandbox
- Add CLI flags, idempotency, and dev mode to install.sh
- Extract common utilities to lib/common.py and webui/api/common.py
- Migrate frontend to htmx for simpler, more maintainable UI
- Update docs to reflect current architecture and deployment model
- Vendor htmx dependencies per project requirements
2026-05-25 00:53:32 +00:00

615 lines
21 KiB
Python

"""
API integration tests — all blueprints tested via a single Flask app fixture.
"""
from unittest.mock import patch
import pytest
from webui.api.certs import bp as certs_bp
from webui.api.dhcp import bp as dhcp_bp
from webui.api.firewall import bp
from webui.api.proxy import bp as proxy_bp
from webui.api.wireguard import bp as wg_bp
@pytest.fixture
def client():
from flask import Flask
app = Flask(__name__)
app.register_blueprint(bp, url_prefix="/api/firewall")
app.register_blueprint(dhcp_bp, url_prefix="/api/dhcp")
app.register_blueprint(proxy_bp, url_prefix="/api/proxy")
app.register_blueprint(certs_bp, url_prefix="/api/certs")
app.register_blueprint(wg_bp, url_prefix="/api/wireguard")
return app.test_client()
# ============================================================================
# Firewall
# ============================================================================
class TestFirewallListZones:
@patch("webui.api.firewall.get_active_zones")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_available, mock_active, client):
mock_active.return_value = {"public": ["eth0"]}
mock_available.return_value = ["public", "internal"]
resp = client.get("/api/firewall/zones")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
assert "public" in data["data"]["active"]
@patch("webui.api.firewall.get_active_zones")
def test_runtime_error(self, mock_active, client):
mock_active.side_effect = RuntimeError("no sudo")
resp = client.get("/api/firewall/zones")
assert resp.status_code == 500
data = resp.get_json()
assert data["ok"] is False
class TestFirewallZoneDetails:
@patch("webui.api.firewall.get_zone_info")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_available, mock_info, client):
mock_available.return_value = ["public", "internal"]
mock_info.return_value = {"name": "public", "services": ["ssh"]}
resp = client.get("/api/firewall/zones/public")
assert resp.status_code == 200
data = resp.get_json()
assert data["data"]["name"] == "public"
class TestFirewallCreateZone:
@patch("webui.api.firewall.create_zone")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_zones, mock_create, client):
mock_zones.return_value = ["public", "internal"]
mock_create.return_value = None
resp = client.post(
"/api/firewall/zones",
json={"name": "dmz", "target": "default"},
)
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
def test_missing_name(self, client):
resp = client.post(
"/api/firewall/zones",
json={"target": "default"},
)
assert resp.status_code == 400
data = resp.get_json()
assert data["ok"] is False
class TestFirewallDeleteZone:
@patch("webui.api.firewall.delete_zone")
@patch("webui.api.firewall.get_available_zones")
def test_success(self, mock_zones, mock_delete, client):
mock_zones.return_value = ["public", "dmz"]
mock_delete.return_value = None
resp = client.delete("/api/firewall/zones/dmz")
assert resp.status_code == 200
@patch("webui.api.firewall.get_available_zones")
def test_not_found(self, mock_zones, client):
mock_zones.return_value = ["public"]
resp = client.delete("/api/firewall/zones/dmz")
assert resp.status_code == 404
class TestFirewallRichRules:
@patch("webui.api.firewall.add_rich_rule")
def test_add(self, mock_add, client):
mock_add.return_value = {"id": "abc123", "rule": "rule accept"}
resp = client.post(
"/api/firewall/rich-rules",
json={
"zone": "public",
"rule": 'rule family="ipv4" port protocol="tcp" port="443" accept',
},
)
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
assert data["data"]["id"] == "abc123"
def test_missing_fields(self, client):
resp = client.post("/api/firewall/rich-rules", json={})
assert resp.status_code == 400
@patch("webui.api.firewall.get_rich_rules")
@patch("webui.api.firewall.get_config")
def test_list(self, mock_cfg, mock_list, client):
mock_list.return_value = ["rule1"]
mock_cfg.return_value = {
"zones": {"public": {"rich_rules": [{"id": "a1", "rule": "rule1"}]}}
}
resp = client.get("/api/firewall/rich-rules/public")
assert resp.status_code == 200
data = resp.get_json()
assert isinstance(data["data"], list)
@patch("webui.api.firewall.remove_rich_rule_by_id")
def test_remove_by_id(self, mock_remove, client):
mock_remove.return_value = None
resp = client.delete("/api/firewall/rich-rules/public/abc123")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@patch("webui.api.firewall.remove_rich_rule_by_id")
def test_remove_not_found(self, mock_remove, client):
mock_remove.side_effect = ValueError("not found")
resp = client.delete("/api/firewall/rich-rules/public/abc123")
assert resp.status_code == 404
class TestFirewallServices:
@patch("webui.api.firewall.get_services")
def test_list(self, mock_services, client):
mock_services.return_value = ["ssh", "http", "dns"]
resp = client.get("/api/firewall/services")
assert resp.status_code == 200
assert resp.get_json()["data"] == ["ssh", "http", "dns"]
class TestFirewallInterfaces:
@patch("webui.api.firewall.get_interfaces")
def test_list(self, mock_ifaces, client):
mock_ifaces.return_value = ["eth0", "eth1"]
resp = client.get("/api/firewall/interfaces")
assert resp.status_code == 200
assert resp.get_json()["data"] == ["eth0", "eth1"]
class TestFirewallMasquerade:
@patch("webui.api.firewall.set_masquerade")
def test_enable(self, mock_set, client):
mock_set.return_value = None
resp = client.post(
"/api/firewall/masquerade",
json={"zone": "internal", "enable": True},
)
assert resp.status_code == 200
def test_missing_fields(self, client):
resp = client.post("/api/firewall/masquerade", json={})
assert resp.status_code == 400
class TestFirewallForwardPort:
@patch("webui.api.firewall.add_forward_port")
def test_add(self, mock_add, client):
mock_add.return_value = {"id": "fp1", "port": 443, "proto": "tcp"}
resp = client.post(
"/api/firewall/forward-port",
json={"zone": "public", "port": 443, "proto": "tcp"},
)
assert resp.status_code == 200
data = resp.get_json()
assert data["data"]["id"] == "fp1"
def test_missing_fields(self, client):
resp = client.post(
"/api/firewall/forward-port",
json={"zone": "public"},
)
assert resp.status_code == 400
@patch("webui.api.firewall.remove_forward_port_by_id")
def test_remove_by_id(self, mock_remove, client):
mock_remove.return_value = None
resp = client.delete("/api/firewall/forward-port/public/443/tcp")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@patch("webui.api.firewall.remove_forward_port_by_id")
def test_remove_not_found(self, mock_remove, client):
mock_remove.side_effect = ValueError("not found")
resp = client.delete("/api/firewall/forward-port/public/999/tcp")
assert resp.status_code == 404
# ============================================================================
# DHCP
# ============================================================================
class TestDhcpConfig:
@patch("webui.api.dhcp.get_config")
def test_get(self, mock_get, client):
mock_get.return_value = {"dhcp": {}, "dns": {}}
resp = client.get("/api/dhcp/config")
assert resp.status_code == 200
assert resp.get_json()["ok"] is True
def test_post_invalid_body(self, client):
resp = client.post(
"/api/dhcp/config", data="not json", content_type="text/plain"
)
data = resp.get_json()
assert data is not None
class TestDhcpApply:
@patch("webui.api.dhcp.apply_config")
def test_apply(self, mock_apply, client):
mock_apply.return_value = None
resp = client.post("/api/dhcp/apply")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
class TestDhcpStatus:
@patch("webui.api.dhcp.dnsmasq_status")
def test_success(self, mock_status, client):
mock_status.return_value = {"service_active": True}
resp = client.get("/api/dhcp/status")
assert resp.status_code == 200
data = resp.get_json()
assert data["data"]["service_active"] is True
class TestDhcpRanges:
@patch("webui.api.dhcp.set_dhcp_range")
def test_add_range(self, mock_set, client):
mock_set.return_value = None
resp = client.post(
"/api/dhcp/ranges",
json={
"interface": "eth0",
"start": "192.168.1.100",
"end": "192.168.1.200",
"lease_time": "2h",
},
)
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
def test_add_range_missing_fields(self, client):
resp = client.post("/api/dhcp/ranges", json={"start": "192.168.1.100"})
assert resp.status_code == 400
@patch("webui.api.dhcp.remove_dhcp_range")
def test_remove_range(self, mock_remove, client):
mock_remove.return_value = None
resp = client.delete(
"/api/dhcp/ranges",
json={
"interface": "eth0",
"start": "192.168.1.100",
"end": "192.168.1.200",
},
)
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
def test_remove_range_missing_fields(self, client):
resp = client.delete("/api/dhcp/ranges", json={})
assert resp.status_code == 400
class TestDhcpStaticLease:
@patch("webui.api.dhcp.add_static_lease")
def test_add(self, mock_add, client):
mock_add.return_value = None
resp = client.post(
"/api/dhcp/static-lease",
json={"mac": "AA:BB:CC", "ip": "10.0.0.5"},
)
assert resp.status_code == 200
def test_missing_mac(self, client):
resp = client.post("/api/dhcp/static-lease", json={"ip": "10.0.0.5"})
assert resp.status_code == 400
@patch("webui.api.dhcp.remove_static_lease")
@patch("webui.api.dhcp.get_config")
def test_remove(self, mock_get, mock_remove, client):
mock_get.return_value = {
"dhcp": {"static_leases": [{"mac": "AA:BB:CC", "ip": "10.0.0.5"}]}
}
mock_remove.return_value = None
resp = client.delete("/api/dhcp/static-lease/AA:BB:CC")
assert resp.status_code == 200
@patch("webui.api.dhcp.get_config")
def test_remove_not_found(self, mock_get, client):
mock_get.return_value = {"dhcp": {"static_leases": []}}
resp = client.delete("/api/dhcp/static-lease/AA:BB:CC")
assert resp.status_code == 404
class TestDhcpDnsRecord:
@patch("webui.api.dhcp.add_dns_record")
def test_add(self, mock_add, client):
mock_add.return_value = None
resp = client.post(
"/api/dhcp/dns-record",
json={"name": "host.local", "address": "10.0.0.10"},
)
assert resp.status_code == 200
def test_missing_fields(self, client):
resp = client.post("/api/dhcp/dns-record", json={})
assert resp.status_code == 400
@patch("webui.api.dhcp.remove_dns_record")
@patch("webui.api.dhcp.get_config")
def test_remove(self, mock_get, mock_remove, client):
mock_get.return_value = {
"dns": {"custom_records": [{"name": "host.local", "address": "10.0.0.10"}]}
}
mock_remove.return_value = None
resp = client.delete("/api/dhcp/dns-record/host.local")
assert resp.status_code == 200
@patch("webui.api.dhcp.get_config")
def test_remove_not_found(self, mock_get, client):
mock_get.return_value = {"dns": {"custom_records": []}}
resp = client.delete("/api/dhcp/dns-record/host.local")
assert resp.status_code == 404
# ============================================================================
# Proxy
# ============================================================================
class TestProxyDomains:
@patch("webui.api.proxy.get_domains")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/proxy/domains")
assert resp.status_code == 200
@patch("webui.api.proxy.add_domain")
def test_add(self, mock_add, client):
mock_add.return_value = None
resp = client.post(
"/api/proxy/domains",
json={"domain": "ex.com", "backend_host": "10.0.0.1", "backend_port": 80},
)
assert resp.status_code == 200
def test_add_missing_domain(self, client):
resp = client.post("/api/proxy/domains", json={})
assert resp.status_code == 400
class TestProxyApply:
@patch("webui.api.proxy.apply")
def test_apply(self, mock_apply, client):
mock_apply.return_value = None
resp = client.post("/api/proxy/apply")
assert resp.status_code == 200
class TestProxyTest:
@patch("webui.api.proxy.test_config")
def test_valid(self, mock_test, client):
mock_test.return_value = (True, "syntax ok")
resp = client.post("/api/proxy/test")
assert resp.status_code == 200
data = resp.get_json()
assert data["data"]["valid"] is True
@patch("webui.api.proxy.test_config")
def test_invalid(self, mock_test, client):
mock_test.return_value = (False, "error msg")
resp = client.post("/api/proxy/test")
assert resp.status_code == 400
data = resp.get_json()
assert data["ok"] is False
assert data["error"] == "error msg"
# ============================================================================
# Certs
# ============================================================================
class TestCertsList:
@patch("webui.api.certs.list_certs")
def test_list(self, mock_list, client):
mock_list.return_value = []
resp = client.get("/api/certs/list")
assert resp.status_code == 200
@patch("webui.api.certs.get_cert_info")
def test_details_not_found(self, mock_info, client):
mock_info.side_effect = ValueError("not found")
resp = client.get("/api/certs/example.com")
assert resp.status_code == 404
class TestCertsIssue:
def test_missing_domain(self, client):
resp = client.post("/api/certs/issue", json={})
assert resp.status_code == 400
class TestCertsEmail:
def test_missing_email(self, client):
resp = client.post("/api/certs/email", json={})
assert resp.status_code == 400
# ============================================================================
# WireGuard
# ============================================================================
class TestWireguardConfig:
@patch("webui.api.wireguard.get_config")
def test_get(self, mock_get, client):
mock_get.return_value = {
"interface": {"name": "wg0", "private_key": "secret"},
"peers": {},
}
resp = client.get("/api/wireguard/config")
data = resp.get_json()
assert data["ok"] is True
assert "private_key" not in data["data"]["interface"]
@patch("webui.api.wireguard.save_config")
def test_post(self, mock_save, client):
mock_save.return_value = None
resp = client.post("/api/wireguard/config", json={"peers": {}})
assert resp.status_code == 200
@patch("webui.api.wireguard.save_config")
@patch("webui.api.wireguard.get_config")
def test_post_strips_private_key(self, mock_get, mock_save, client):
mock_get.return_value = {
"interface": {"name": "wg0", "private_key": "existing"},
"peers": {},
}
mock_save.return_value = None
resp = client.post(
"/api/wireguard/config",
json={"interface": {"name": "wg0", "private_key": "secret"}, "peers": {}},
)
assert resp.status_code == 200
saved = mock_save.call_args[0][0]
assert saved["interface"]["private_key"] == "existing"
@patch("webui.api.wireguard.save_config")
@patch("webui.api.wireguard.get_config")
def test_patch_strips_private_key(self, mock_get, mock_save, client):
mock_get.return_value = {
"interface": {"name": "wg0", "private_key": "existing"},
"peers": {},
}
mock_save.return_value = None
resp = client.patch(
"/api/wireguard/config",
json={"interface": {"name": "wg1", "private_key": "injected"}},
)
assert resp.status_code == 200
saved = mock_save.call_args[0][0]
assert saved.get("interface", {}).get("private_key") == "existing"
class TestWireguardPeers:
@patch("webui.api.wireguard.get_peers")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/wireguard/peers")
assert resp.status_code == 200
@patch("webui.api.wireguard.add_peer")
def test_add(self, mock_add, client):
mock_add.return_value = {
"name": "client1",
"public_key": "pub",
}
resp = client.post(
"/api/wireguard/peers",
json={"name": "client1"},
)
data = resp.get_json()
assert data["ok"] is True
assert "private_key" not in data["data"]
def test_add_missing_name(self, client):
resp = client.post("/api/wireguard/peers", json={})
assert resp.status_code == 400
@patch("webui.api.wireguard.remove_peer")
@patch("webui.api.wireguard.get_config")
def test_remove_by_name(self, mock_get, mock_remove, client):
mock_get.return_value = {"peers": {"client1": {}}}
mock_remove.return_value = None
resp = client.delete("/api/wireguard/peers/client1")
assert resp.status_code == 200
@patch("webui.api.wireguard.get_config")
def test_remove_not_found(self, mock_get, client):
mock_get.return_value = {"peers": {}}
resp = client.delete("/api/wireguard/peers/unknown")
assert resp.status_code == 404
class TestWireguardInitialize:
@patch("webui.api.wireguard.initialize")
def test_initialize(self, mock_init, client):
mock_init.return_value = None
resp = client.post("/api/wireguard/initialize")
data = resp.get_json()
assert data["ok"] is True
class TestWireguardGenerateClient:
def test_missing_name(self, client):
resp = client.post("/api/wireguard/generate-client", json={})
assert resp.status_code == 400
class TestWireguardStatus:
@patch("webui.api.wireguard.status")
def test_get(self, mock_status, client):
mock_status.return_value = {"up": True, "interface": {}, "peers": []}
resp = client.get("/api/wireguard/status")
assert resp.status_code == 200
class TestWireguardUp:
@patch("webui.api.wireguard.apply")
def test_up_starts_tunnel(self, mock_apply, client):
mock_apply.return_value = None
resp = client.post("/api/wireguard/up")
assert resp.status_code == 200
@patch("webui.api.wireguard.apply")
def test_up_error(self, mock_apply, client):
mock_apply.side_effect = RuntimeError("interface down")
resp = client.post("/api/wireguard/up")
assert resp.status_code == 500
class TestWireguardDown:
@patch("webui.api.wireguard.down")
def test_down_stops_tunnel(self, mock_down, client):
mock_down.return_value = None
resp = client.post("/api/wireguard/down")
assert resp.status_code == 200
class TestWireguardApply:
@patch("webui.api.wireguard.apply")
def test_apply(self, mock_apply, client):
mock_apply.return_value = None
resp = client.post("/api/wireguard/apply")
assert resp.status_code == 200
# ============================================================================
# Helpers
# ============================================================================
class TestResponseHelpers:
@patch("webui.api.firewall.get_active_zones")
@patch("webui.api.firewall.get_available_zones")
def test_error_response_format(self, mock_a, mock_b, client):
mock_a.side_effect = RuntimeError("fail")
resp = client.get("/api/firewall/zones")
data = resp.get_json()
assert "error" in data
assert "ok" in data
assert data["ok"] is False