roof
768571f2b7
Port conflict validation: - api/port_registry.py: detect_conflicts() checks all service sections for shared port values - api/app.py: returns HTTP 409 on port conflict after existing range validation - webui/src/pages/Settings.jsx: JS-side detectPortConflicts() with useMemo shows inline conflict errors and blocks Save before the request is made; catch blocks surface server error messages (including 409) instead of generic fallbacks Config autosave on Apply: - webui/src/contexts/DraftConfigContext.jsx: new context; Settings registers flush callbacks per section; App calls flushAll() before applyPending() when any section is dirty - webui/src/App.jsx: wraps tree with DraftConfigProvider, handleApply shows 'saving' banner state and awaits flushAll() - webui/src/pages/Settings.jsx: registers identity + per-service flushers; propagates dirty state into context via setDirty; uses refs to avoid stale closures Extended integration test coverage (114 new tests): - tests/integration/test_config_api.py: GET/PUT config, export, import, backup lifecycle - tests/integration/test_network_services.py: DNS records + DHCP reservations CRUD - tests/integration/test_containers.py: list, restart, logs, stats; recovery polling - tests/integration/test_negative_scenarios.py: error-path coverage for all endpoints - tests/test_port_conflicts.py: 20 unit tests for port_registry.detect_conflicts() Pre-commit hook updated to skip tests/integration/ (live-stack tests require a running stack and must be run explicitly via `make test-integration`). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
257 lines
8.8 KiB
Python
257 lines
8.8 KiB
Python
"""
|
|
Read-only integration tests: health, config, containers, WireGuard, network services.
|
|
|
|
Run with: pytest tests/integration/test_live_api.py -v
|
|
Or: PIC_HOST=192.168.31.51 pytest tests/integration/test_live_api.py -v
|
|
"""
|
|
import pytest
|
|
import sys, os
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
from conftest import API_BASE
|
|
|
|
# Shorthand helpers — always hits the live API
|
|
import requests as _req
|
|
|
|
def get(path, **kw):
|
|
return _req.get(f"{API_BASE}{path}", **kw)
|
|
|
|
def post(path, **kw):
|
|
return _req.post(f"{API_BASE}{path}", **kw)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Health & status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHealth:
|
|
def test_health_returns_200(self):
|
|
r = get('/health')
|
|
assert r.status_code == 200
|
|
|
|
def test_health_body(self):
|
|
r = get('/health')
|
|
data = r.json()
|
|
assert data.get('status') == 'healthy'
|
|
assert 'timestamp' in data
|
|
|
|
def test_api_status_returns_200(self):
|
|
r = get('/api/status')
|
|
assert r.status_code == 200
|
|
|
|
def test_api_status_body(self):
|
|
r = get('/api/status')
|
|
data = r.json()
|
|
assert 'cell_name' in data or 'status' in data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestConfig:
|
|
def test_get_config(self):
|
|
r = get('/api/config')
|
|
assert r.status_code == 200
|
|
|
|
def test_config_has_required_fields(self):
|
|
data = get('/api/config').json()
|
|
for field in ('cell_name', 'domain', 'ip_range'):
|
|
assert field in data, f"config missing field: {field}"
|
|
|
|
def test_config_ip_range_is_cidr(self):
|
|
import ipaddress
|
|
ip_range = get('/api/config').json()['ip_range']
|
|
ipaddress.ip_network(ip_range, strict=False) # raises if invalid
|
|
|
|
def test_pending_endpoint_reachable(self):
|
|
r = get('/api/config/pending')
|
|
assert r.status_code == 200
|
|
|
|
def test_backups_endpoint_reachable(self):
|
|
r = get('/api/config/backups')
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json(), list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Containers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
EXPECTED_CONTAINERS = [
|
|
'cell-caddy', 'cell-dns', 'cell-dhcp', 'cell-ntp',
|
|
'cell-mail', 'cell-radicale', 'cell-webdav', 'cell-wireguard',
|
|
'cell-api', 'cell-webui', 'cell-rainloop', 'cell-filegator',
|
|
]
|
|
|
|
def _containers_accessible():
|
|
try:
|
|
return get('/api/containers').status_code != 403
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
class TestContainers:
|
|
@pytest.mark.skipif(not _containers_accessible(), reason="Container endpoint returns 403 — run `make update`")
|
|
def test_containers_endpoint_reachable(self):
|
|
r = get('/api/containers')
|
|
assert r.status_code == 200
|
|
|
|
@pytest.mark.skipif(not _containers_accessible(), reason="Container endpoint returns 403 — run `make update`")
|
|
def test_containers_returns_list(self):
|
|
data = get('/api/containers').json()
|
|
assert isinstance(data, list)
|
|
assert len(data) > 0
|
|
|
|
@pytest.mark.skipif(not _containers_accessible(), reason="Container endpoint returns 403 — run `make update`")
|
|
def test_all_expected_containers_present(self):
|
|
data = get('/api/containers').json()
|
|
running = {c['name'] for c in data}
|
|
missing = set(EXPECTED_CONTAINERS) - running
|
|
assert not missing, f"Containers not found: {missing}"
|
|
|
|
@pytest.mark.skipif(not _containers_accessible(), reason="Container endpoint returns 403 — run `make update`")
|
|
def test_all_expected_containers_running(self):
|
|
data = get('/api/containers').json()
|
|
by_name = {c['name']: c for c in data}
|
|
not_running = [
|
|
name for name in EXPECTED_CONTAINERS
|
|
if by_name.get(name, {}).get('status') != 'running'
|
|
]
|
|
assert not not_running, f"Containers not running: {not_running}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WireGuard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestWireGuard:
|
|
def test_wireguard_status_up(self):
|
|
r = get('/api/wireguard/status')
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data.get('running') is True, f"WireGuard not running: {data}"
|
|
|
|
def test_wireguard_interface_name(self):
|
|
data = get('/api/wireguard/status').json()
|
|
assert data.get('interface') == 'wg0'
|
|
|
|
def test_wireguard_keys_endpoint(self):
|
|
r = get('/api/wireguard/keys')
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert 'public_key' in data
|
|
|
|
def test_wireguard_wg_peers_endpoint(self):
|
|
r = get('/api/wireguard/peers')
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json(), list)
|
|
|
|
def test_wireguard_config_endpoint(self):
|
|
r = get('/api/wireguard/config')
|
|
assert r.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Network services: DNS, DHCP, NTP
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNetworkServices:
|
|
def test_dns_records_endpoint(self):
|
|
r = get('/api/dns/records')
|
|
assert r.status_code == 200
|
|
|
|
def test_dns_status_endpoint(self):
|
|
r = get('/api/dns/status')
|
|
assert r.status_code == 200
|
|
|
|
def test_dhcp_leases_endpoint(self):
|
|
r = get('/api/dhcp/leases')
|
|
assert r.status_code == 200
|
|
|
|
def test_ntp_status_endpoint(self):
|
|
r = get('/api/ntp/status')
|
|
assert r.status_code == 200
|
|
|
|
def test_network_info_endpoint(self):
|
|
r = get('/api/network/info')
|
|
assert r.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Services bus / all-service status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestServicesStatus:
|
|
def test_all_services_status_reachable(self):
|
|
r = get('/api/services/status')
|
|
assert r.status_code == 200
|
|
|
|
def test_services_status_has_expected_keys(self):
|
|
data = get('/api/services/status').json()
|
|
for svc in ('network', 'wireguard', 'email', 'calendar', 'files'):
|
|
assert svc in data, f"Missing service in status: {svc}"
|
|
|
|
def test_services_connectivity_reachable(self):
|
|
r = get('/api/services/connectivity')
|
|
assert r.status_code == 200
|
|
|
|
def test_health_history_reachable(self):
|
|
r = get('/api/health/history')
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json(), list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Peers read-only
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPeersReadOnly:
|
|
def test_peers_list_endpoint(self):
|
|
r = get('/api/peers')
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json(), list)
|
|
|
|
def test_peers_have_required_fields(self):
|
|
peers = get('/api/peers').json()
|
|
for peer in peers:
|
|
for field in ('peer', 'ip', 'public_key', 'service_access'):
|
|
assert field in peer, f"Peer missing field '{field}': {peer}"
|
|
|
|
def test_peer_service_access_values_are_valid(self):
|
|
valid = {'calendar', 'files', 'mail', 'webdav'}
|
|
peers = get('/api/peers').json()
|
|
for peer in peers:
|
|
for svc in peer.get('service_access', []):
|
|
assert svc in valid, f"Unknown service '{svc}' in peer {peer['peer']}"
|
|
|
|
def test_wg_peer_statuses_endpoint(self):
|
|
r = get('/api/wireguard/peers/statuses')
|
|
assert r.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Input validation (no state changes)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestValidation:
|
|
def test_add_peer_missing_name_returns_400(self):
|
|
r = post('/api/peers', json={'public_key': 'dummykey=='})
|
|
assert r.status_code == 400
|
|
|
|
def test_add_peer_missing_key_returns_400(self):
|
|
r = post('/api/peers', json={'name': 'no-key-peer'})
|
|
assert r.status_code == 400
|
|
|
|
def test_add_peer_invalid_service_access_returns_400(self):
|
|
r = post('/api/peers', json={
|
|
'name': 'bad-svc-peer',
|
|
'public_key': 'dummykey==',
|
|
'service_access': ['invalid_service'],
|
|
})
|
|
assert r.status_code == 400
|
|
assert 'service_access' in r.json().get('error', '')
|
|
|
|
def test_generate_keys_missing_name_returns_400(self):
|
|
r = post('/api/wireguard/keys/peer', json={})
|
|
assert r.status_code == 400
|