mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8feb56faf6145553e84e2e582f52459a0146f5a8
vacuum-wall/tests/test_api.py
T
mteehan bc72db903c feat: add networkd subsystem and fix code review issues 
Phase 1-4: Networkd subsystem
- lib/network.py: systemd-networkd config renderer (.network INI files)
  with full schema support: [Match], [Link], [Network], [Address], [Route],
  [DHCPv4], [DHCPv6] sections. One Address/=DNS= line per value per spec.
  Route sections use #N suffix per systemd.syntax(7).
- lib/network.py: generate_network_files() with 50-<name>.network prefix
  and stale file cleanup
- lib/network.py: collect_upstream_dns() filters local/private DNS
- lib/network.py: infer_dhcp_ranges() and infer_zones() helpers
- daemon/handlers/network.py: routes for GET/POST /network/interfaces
  and full apply with DNS upstream sync to dnsmasq
- webui/api/network.py: Flask blueprint for /api/network/* endpoints
- webui/api: interfaces page updated with IP config inline editing
- lib/state.py: networkd collector using parse_networkctl_status()
- system/sudoers.d/vacuum-walld: networkctl + systemd-network rules
- system/systemd/vacuum-walld.service: ReadWritePaths for /etc/systemd/network
- install.sh: ACME email now optional, configured from WebUI
- lib/acme.py: get_email() falls back to declarative config

Phase 5: Code review fixes
- daemon/server.py: path params now win over JSON body and query params
  in request body merge (prevents config save name override)
- daemon/server.py: remove dead 'import re'
- daemon/handlers/network.py: replace Path.mkdir() with sudo mkdir
  for /etc/systemd/network (ProtectSystem=strict compatibility)
- system/sudoers.d/vacuum-walld: pin systemctl to specific commands
  (reload/is-active dnsmasq instead of wildcard)
- system/sudoers.d/vacuum-walld: restore !requiretty and section comment
- lib/network.py: remove unused _MANAGEMENT_PORTS constant
- webui/api/network.py: remove redundant body[\name\] = name in save_interface

Tests: 332 passing (110 new/updated), ruff clean
2026-06-01 03:15:50 +00:00

901 lines
28 KiB
Python
Raw Blame History

"""
API integration tests — all blueprints tested via a single Flask app fixture.
Mocks daemon.client in each blueprint's module namespace to avoid needing
a running daemon.
"""
from unittest.mock import patch as _patch
import pytest
def _fw(func, **kw):
"""Patch daemon.client.{func} in the firewall blueprint namespace."""
return _patch(f"webui.api.firewall.{func}", **kw)
def _dh(func, **kw):
"""Patch daemon.client.{func} in the dhcp blueprint namespace."""
return _patch(f"webui.api.dhcp.{func}", **kw)
def _px(func, **kw):
"""Patch daemon.client.{func} in the proxy blueprint namespace."""
return _patch(f"webui.api.proxy.{func}", **kw)
def _ce(func, **kw):
"""Patch daemon.client.{func} in the certs blueprint namespace."""
return _patch(f"webui.api.certs.{func}", **kw)
def _wg(func, **kw):
"""Patch daemon.client.{func} in the wireguard blueprint namespace."""
return _patch(f"webui.api.wireguard.{func}", **kw)
def _ne(func, **kw):
"""Patch daemon.client.{func} in the network blueprint namespace."""
return _patch(f"webui.api.network.{func}", **kw)
@pytest.fixture
def client():
from flask import Flask
from webui.api.certs import bp as certs_bp
from webui.api.dhcp import bp as dhcp_bp
from webui.api.firewall import bp as firewall_bp
from webui.api.network import bp as network_bp
from webui.api.proxy import bp as proxy_bp
from webui.api.wireguard import bp as wg_bp
app = Flask(__name__)
app.register_blueprint(firewall_bp, url_prefix="/api/firewall")
app.register_blueprint(network_bp, url_prefix="/api/network")
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:
@_fw("get")
def test_success(self, mock_get, client):
mock_get.return_value = {
"active": {"public": ["eth0"]},
"available": ["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"]
@_fw("get")
def test_runtime_error(self, mock_get, client):
mock_get.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:
@_fw("get")
def test_success(self, mock_get, client):
mock_get.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:
@_fw("post")
@_fw("get")
def test_success(self, mock_get, mock_post, client):
mock_get.return_value = {"active": {}, "available": ["public", "internal"]}
mock_post.return_value = {"zone": "dmz"}
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:
@_fw("delete")
def test_success(self, mock_delete, client):
mock_delete.return_value = {"zone": "dmz"}
resp = client.delete("/api/firewall/zones/dmz")
assert resp.status_code == 200
@_fw("delete")
def test_not_found(self, mock_delete, client):
from daemon.client import NotFound
mock_delete.side_effect = NotFound("Zone does not exist")
resp = client.delete("/api/firewall/zones/dmz")
assert resp.status_code == 404
class TestFirewallRichRules:
@_fw("post")
def test_add(self, mock_post, client):
mock_post.return_value = {
"id": "abc123",
"rule": "rule accept",
"zone": "public",
}
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
@_fw("get")
def test_list(self, mock_get, client):
mock_get.return_value = [{"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)
@_fw("delete")
def test_remove_by_id(self, mock_delete, client):
mock_delete.return_value = {"zone": "public", "id": "abc123"}
resp = client.delete("/api/firewall/rich-rules/public/abc123")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@_fw("delete")
def test_remove_not_found(self, mock_delete, client):
from daemon.client import NotFound
mock_delete.side_effect = NotFound("not found")
resp = client.delete("/api/firewall/rich-rules/public/abc123")
assert resp.status_code == 404
class TestFirewallServices:
@_fw("get")
def test_list(self, mock_get, client):
mock_get.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:
@_fw("get")
def test_list(self, mock_get, client):
mock_get.return_value = [
{
"name": "eth0",
"mac": "aa:bb:cc:dd:ee:00",
"state": "UP",
"mtu": 1500,
"ips": ["192.168.1.1/24"],
"ipv6": [],
"zone": "internal",
},
{
"name": "eth1",
"mac": "aa:bb:cc:dd:ee:01",
"state": "UP",
"mtu": 1500,
"ips": ["10.0.0.1/24"],
"ipv6": [],
"zone": "public",
},
]
resp = client.get("/api/firewall/interfaces")
assert resp.status_code == 200
data = resp.get_json()["data"]
assert len(data) == 2
assert data[0]["name"] == "eth0"
class TestFirewallMasquerade:
@_fw("post")
def test_enable(self, mock_post, client):
mock_post.return_value = {"zone": "internal", "masquerade": True}
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:
@_fw("post")
def test_add(self, mock_post, client):
mock_post.return_value = {
"id": "fp1",
"port": 443,
"proto": "tcp",
"zone": "public",
}
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
@_fw("delete")
def test_remove_by_id(self, mock_delete, client):
mock_delete.return_value = {"zone": "public", "port": 443, "proto": "tcp"}
resp = client.delete("/api/firewall/forward-port/public/443/tcp")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@_fw("delete")
def test_remove_not_found(self, mock_delete, client):
from daemon.client import NotFound
mock_delete.side_effect = NotFound("not found")
resp = client.delete("/api/firewall/forward-port/public/999/tcp")
assert resp.status_code == 404
def test_add_value_error(self, client):
resp = client.post(
"/api/firewall/forward-port",
json={"zone": "public", "port": "not_a_number", "proto": "tcp"},
)
assert resp.status_code == 400
class TestFirewallConfigApply:
@_fw("post")
def test_config_apply_success(self, mock_post, client):
mock_post.return_value = {
"applied_zones": ["public"],
"backup": "/tmp/rules.json",
}
resp = client.post("/api/firewall/config/apply")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
assert "public" in data["data"]["applied_zones"]
@_fw("post")
def test_config_apply_error(self, mock_post, client):
mock_post.side_effect = RuntimeError("apply failed")
resp = client.post("/api/firewall/config/apply")
assert resp.status_code == 500
data = resp.get_json()
assert data["ok"] is False
assert data["error"] == "apply failed"
# ============================================================================
# DHCP
# ============================================================================
class TestDhcpConfig:
@_dh("get")
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
@_dh("post")
def test_post_invalid_body(self, mock_post, 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:
@_dh("post")
def test_apply(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/dhcp/apply")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
class TestDhcpStatus:
@_dh("get")
def test_success(self, mock_get, client):
mock_get.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:
@_dh("post")
def test_add_range(self, mock_post, client):
mock_post.return_value = {
"interface": "eth0",
"start": "192.168.1.100",
"end": "192.168.1.200",
}
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
@_dh("delete")
def test_remove_range(self, mock_delete, client):
mock_delete.return_value = {
"interface": "eth0",
"start": "192.168.1.100",
"end": "192.168.1.200",
}
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:
@_dh("post")
def test_add(self, mock_post, client):
mock_post.return_value = {"mac": "AA:BB:CC", "ip": "10.0.0.5"}
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
@_dh("delete")
def test_remove(self, mock_delete, client):
mock_delete.return_value = {"mac": "AA:BB:CC"}
resp = client.delete("/api/dhcp/static-lease/AA:BB:CC")
assert resp.status_code == 200
class TestDhcpDnsRecord:
@_dh("post")
def test_add(self, mock_post, client):
mock_post.return_value = {"name": "host.local", "address": "10.0.0.10"}
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
@_dh("delete")
def test_remove(self, mock_delete, client):
mock_delete.return_value = {"name": "host.local"}
resp = client.delete("/api/dhcp/dns-record/host.local")
assert resp.status_code == 200
# ============================================================================
# Proxy
# ============================================================================
class TestProxyDomains:
@_px("get")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/proxy/domains")
assert resp.status_code == 200
@_px("post")
def test_add(self, mock_post, client):
mock_post.return_value = {"domain": "ex.com"}
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:
@_px("post")
def test_apply(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/proxy/apply")
assert resp.status_code == 200
class TestProxyTest:
@_px("post")
def test_valid(self, mock_post, client):
mock_post.return_value = {"valid": True, "output": "syntax ok"}
resp = client.post("/api/proxy/test")
assert resp.status_code == 200
data = resp.get_json()
assert data["data"]["valid"] is True
@_px("post")
def test_invalid(self, mock_post, client):
mock_post.return_value = {"valid": False, "output": "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:
@_ce("get")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/certs/list")
assert resp.status_code == 200
@_ce("get")
def test_details_not_found(self, mock_get, client):
from daemon.client import NotFound
mock_get.side_effect = NotFound("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/start", 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:
@_wg("get")
def test_get(self, mock_get, client):
mock_get.return_value = {
"interface": {"name": "wg0"},
"peers": {},
}
resp = client.get("/api/wireguard/config")
data = resp.get_json()
assert data["ok"] is True
assert "private_key" not in data["data"]["interface"]
@_wg("post")
def test_post(self, mock_post, client):
mock_post.return_value = {"config_saved": True}
resp = client.post("/api/wireguard/config", json={"peers": {}})
assert resp.status_code == 200
@_wg("patch")
def test_patch(self, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
resp = client.patch(
"/api/wireguard/config",
json={"interface": {"name": "wg0"}},
)
assert resp.status_code == 200
class TestWireguardPeers:
@_wg("get")
def test_list(self, mock_get, client):
mock_get.return_value = []
resp = client.get("/api/wireguard/peers")
assert resp.status_code == 200
@_wg("post")
def test_add(self, mock_post, client):
mock_post.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
@_wg("delete")
def test_remove_by_name(self, mock_delete, client):
mock_delete.return_value = {"name": "client1"}
resp = client.delete("/api/wireguard/peers/client1")
assert resp.status_code == 200
class TestWireguardInitialize:
@_wg("post")
def test_initialize(self, mock_post, client):
mock_post.return_value = {"initialized": True}
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:
@_wg("get")
def test_get(self, mock_get, client):
mock_get.return_value = {"up": True, "interface": {}, "peers": []}
resp = client.get("/api/wireguard/status")
assert resp.status_code == 200
class TestWireguardUp:
@_wg("post")
def test_up_starts_tunnel(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/wireguard/up")
assert resp.status_code == 200
@_wg("post")
def test_up_error(self, mock_post, client):
mock_post.side_effect = RuntimeError("interface down")
resp = client.post("/api/wireguard/up")
assert resp.status_code == 500
class TestWireguardDown:
@_wg("post")
def test_down_stops_tunnel(self, mock_post, client):
mock_post.return_value = {"down": True}
resp = client.post("/api/wireguard/down")
assert resp.status_code == 200
class TestWireguardApply:
@_wg("post")
def test_apply(self, mock_post, client):
mock_post.return_value = {"applied": True}
resp = client.post("/api/wireguard/apply")
assert resp.status_code == 200
# ============================================================================
# Helpers
# ============================================================================
class TestResponseHelpers:
@_fw("get")
def test_error_response_format(self, mock_get, client):
mock_get.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
# ============================================================================
# Firewall Config CRUD (daemon-backed)
# ============================================================================
class TestFirewallConfig:
@_fw("get")
def test_config_get(self, mock_get, client):
mock_get.return_value = {"zones": {"public": {"interfaces": ["eth0"]}}}
resp = client.get("/api/firewall/config")
assert resp.status_code == 200
@_fw("post")
@_fw("get")
def test_config_save(self, mock_get, mock_post, client):
mock_post.return_value = {"config_saved": True}
mock_get.return_value = {
"pending": [],
"needs_apply": False,
"unmanaged_zones": {},
}
resp = client.post(
"/api/firewall/config", json={"zones": {"public": {"interfaces": ["eth0"]}}}
)
assert resp.status_code == 200
def test_config_save_missing_zones(self, client):
resp = client.post("/api/firewall/config", json={})
assert resp.status_code == 400
@_fw("patch")
@_fw("get")
def test_config_patch(self, mock_get, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
mock_get.return_value = {
"pending": [],
"needs_apply": False,
"unmanaged_zones": {},
}
resp = client.patch("/api/firewall/config", json={"zones": {}})
assert resp.status_code == 200
@_fw("post")
def test_config_apply(self, mock_post, client):
mock_post.return_value = {
"applied_zones": ["public"],
"backup": "/tmp/rules.json",
}
resp = client.post("/api/firewall/config/apply")
assert resp.status_code == 200
@_fw("get")
def test_config_pending(self, mock_get, client):
mock_get.return_value = {"pending": [], "needs_apply": False}
resp = client.get("/api/firewall/config/pending")
assert resp.status_code == 200
# ============================================================================
# Proxy config
# ============================================================================
class TestProxyConfig:
@_px("get")
def test_get(self, mock_get, client):
mock_get.return_value = {"domains": {}, "ssl": {}}
resp = client.get("/api/proxy/config")
assert resp.status_code == 200
@_px("post")
def test_post(self, mock_post, client):
mock_post.return_value = {"config_saved": True}
resp = client.post("/api/proxy/config", json={"domains": {}})
assert resp.status_code == 200
@_px("patch")
def test_patch(self, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
resp = client.patch("/api/proxy/config", json={})
assert resp.status_code == 200
# ============================================================================
# DHCP config
# ============================================================================
class TestDhcpConfigCrud:
@_dh("patch")
def test_patch(self, mock_patch, client):
mock_patch.return_value = {"config_saved": True}
resp = client.patch("/api/dhcp/config", json={"dhcp": {}})
assert resp.status_code == 200
# ============================================================================
# Proxy management
# ============================================================================
class TestProxyManagement:
@_px("post")
def test_set_management(self, mock_post, client):
mock_post.return_value = {"domain": "vacuum-wall.local"}
resp = client.post(
"/api/proxy/management",
json={
"domain": "vacuum-wall.local",
"flask_host": "127.0.0.1",
"flask_port": 9090,
},
)
assert resp.status_code == 200
# ============================================================================
# Proxy domain update
# ============================================================================
class TestProxyDomainUpdate:
@_px("post")
def test_update(self, mock_post, client):
mock_post.return_value = {"domain": "ex.com"}
resp = client.put(
"/api/proxy/domains/ex.com",
json={"backend_host": "10.0.0.2"},
)
assert resp.status_code == 200
# ============================================================================
# Network
# ============================================================================
class TestNetworkListInterfaces:
@_ne("get")
def test_success(self, mock_get, client):
mock_get.return_value = [
{"name": "eth0", "config": {"addresses": ["10.0.0.1/24"]}}
]
resp = client.get("/api/network/interfaces")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@_ne("get")
def test_runtime_error(self, mock_get, client):
mock_get.side_effect = RuntimeError("networkctl not found")
resp = client.get("/api/network/interfaces")
assert resp.status_code == 500
assert resp.get_json()["ok"] is False
class TestNetworkGetInterface:
@_ne("get")
def test_success(self, mock_get, client):
mock_get.return_value = {
"name": "eth0",
"config": {"addresses": ["10.0.0.1/24"]},
}
resp = client.get("/api/network/interfaces/eth0")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@_ne("get")
def test_not_found(self, mock_get, client):
from daemon.client import NotFound
mock_get.side_effect = NotFound("interface not found")
resp = client.get("/api/network/interfaces/nonexist")
assert resp.status_code == 404
class TestNetworkSaveInterface:
@_ne("post")
def test_success(self, mock_post, client):
mock_post.return_value = {"name": "eth0", "applied": True}
resp = client.post(
"/api/network/interfaces/eth0",
json={
"addresses": ["10.0.0.1/24"],
"gateway": "10.0.0.254",
"dns": ["8.8.8.8"],
"routes": [],
},
)
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
@_ne("post")
def test_not_found(self, mock_post, client):
from daemon.client import NotFound
mock_post.side_effect = NotFound("interface not found")
resp = client.post("/api/network/interfaces/missing", json={})
assert resp.status_code == 404
class TestNetworkReloadInterface:
@_ne("post")
def test_success(self, mock_post, client):
mock_post.return_value = {"name": "eth0", "reloaded": True}
resp = client.post("/api/network/interfaces/eth0/reload")
assert resp.status_code == 200
assert resp.get_json()["ok"] is True
@_ne("post")
def test_runtime_error(self, mock_post, client):
mock_post.side_effect = RuntimeError("reload failed")
resp = client.post("/api/network/interfaces/eth0/reload")
assert resp.status_code == 500
class TestNetworkApplyAll:
@_ne("post")
def test_success(self, mock_post, client):
mock_post.return_value = {"applied": 2, "interfaces": ["eth0", "eth1"]}
resp = client.post("/api/network/apply")
assert resp.status_code == 200
data = resp.get_json()
assert data["ok"] is True
assert data["data"]["applied"] == 2
@_ne("post")
def test_runtime_error(self, mock_post, client):
mock_post.side_effect = RuntimeError("apply failed")
resp = client.post("/api/network/apply")
assert resp.status_code == 500
Reference in New Issue View Git Blame Copy Permalink