""" 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, _resolve_admin_pass # Shorthand helpers — always hits the live API import requests as _req _S = None @pytest.fixture(scope='module', autouse=True) def _auth_session(): global _S _S = requests.Session() _S.headers['Content-Type'] = 'application/json' r = _S.post(f"{API_BASE}/api/auth/login", json={'username': 'admin', 'password': _resolve_admin_pass()}) assert r.status_code == 200, f"Login failed: {{r.text}}" def get(path, **kw): return _S.get(f"{API_BASE}{path}", **kw) def post(path, **kw): return _S.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