9a800e3b6b
DNS A records now return the WireGuard server IP (10.0.0.1) instead of Docker bridge VIPs so cross-cell peers resolve service names correctly regardless of their bridge subnet. DNAT rules (wg0:53→cell-dns:53 and wg0:80→cell-caddy:80) are applied at startup. Caddy routes by Host header, eliminating the Docker bridge subnet conflict. Firewall cell rules allow DNS and service (Caddy) traffic from linked cell subnets. Split-tunnel AllowedIPs now dynamically includes connected-cell VPN subnets and drops the 172.20.0.0/16 range. Peers with route_via set now receive full-tunnel config (0.0.0.0/0) so all their traffic exits via the remote cell. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
624 lines
26 KiB
Python
624 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Unit tests for /api/peer/dashboard and /api/peer/services.
|
|
|
|
These tests verify the exact JSON field names and structure returned by
|
|
both endpoints so UI/API mismatches surface here before reaching users.
|
|
|
|
Coverage:
|
|
- peer_dashboard returns name/transfer_rx/transfer_tx (not peer_name/rx_bytes/tx_bytes)
|
|
- peer_dashboard includes service_urls dict keyed by service name
|
|
- peer_services uses files (not webdav) as the file storage key
|
|
- peer_services email block uses nested smtp/imap objects with host/port
|
|
- peer_services email.address is the full email address (not username)
|
|
- peer_services caldav URL uses calendar.{domain}, not radicale.{domain}:5232
|
|
- peer_services wireguard block includes a config text field with DNS = <coredns-ip>
|
|
- peer_services wireguard DNS is not 10.0.0.1 (WireGuard VPN IP, not CoreDNS)
|
|
- Unauthenticated requests return 401; admin sessions return 403 (peer-only zone)
|
|
- 404 when session has peer_name but peer not in registry
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
|
|
from app import app
|
|
from auth_manager import AuthManager
|
|
|
|
|
|
# ─────────────────────────── helpers ──────────────────────────────────────────
|
|
|
|
def _make_auth(tmp_path):
|
|
data_dir = str(tmp_path / 'data')
|
|
cfg_dir = str(tmp_path / 'cfg')
|
|
os.makedirs(data_dir, exist_ok=True)
|
|
os.makedirs(cfg_dir, exist_ok=True)
|
|
mgr = AuthManager(data_dir=data_dir, config_dir=cfg_dir)
|
|
mgr.create_user('admin', 'AdminPass123!', 'admin')
|
|
mgr.create_user('alice', 'AlicePass123!', 'peer')
|
|
return mgr
|
|
|
|
|
|
def _login(client, username, password):
|
|
return client.post('/api/auth/login',
|
|
data=json.dumps({'username': username, 'password': password}),
|
|
content_type='application/json')
|
|
|
|
|
|
FAKE_PEER = {
|
|
'peer': 'alice',
|
|
'ip': '10.0.0.5',
|
|
'public_key': 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=',
|
|
'private_key': 'BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB=',
|
|
'allowed_ips': '10.0.0.5/32',
|
|
'internet_access': True,
|
|
'service_access': ['calendar', 'files', 'mail', 'webdav'],
|
|
'active': True,
|
|
'config_needs_reinstall': False,
|
|
}
|
|
|
|
FAKE_WG_STATS = {
|
|
'online': True,
|
|
'transfer_rx': 1048576, # 1 MiB
|
|
'transfer_tx': 524288, # 512 KiB
|
|
'last_handshake': '2026-04-26T18:00:00',
|
|
}
|
|
|
|
DOMAIN = 'dev'
|
|
|
|
|
|
_REGISTRY_SENTINEL = object()
|
|
|
|
def _mock_registry(peer=_REGISTRY_SENTINEL):
|
|
reg = MagicMock()
|
|
reg.get_peer.return_value = FAKE_PEER if peer is _REGISTRY_SENTINEL else peer
|
|
return reg
|
|
|
|
|
|
def _mock_wg(dns='172.20.0.3'):
|
|
wg = MagicMock()
|
|
wg.get_peer_status.return_value = FAKE_WG_STATS
|
|
wg.get_keys.return_value = {'public_key': 'SERVERPUBKEY=='}
|
|
wg.get_server_config.return_value = {'endpoint': '1.2.3.4:51820'}
|
|
wg.FULL_TUNNEL_IPS = '0.0.0.0/0, ::/0'
|
|
wg.get_split_tunnel_ips.return_value = '10.0.0.0/24, 172.20.0.0/16'
|
|
wg.get_peer_config.return_value = (
|
|
'[Interface]\n'
|
|
f'PrivateKey = BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB=\n'
|
|
f'Address = 10.0.0.5/32\n'
|
|
f'DNS = {dns}\n'
|
|
'\n'
|
|
'[Peer]\n'
|
|
'PublicKey = SERVERPUBKEY==\n'
|
|
'AllowedIPs = 0.0.0.0/0, ::/0\n'
|
|
'Endpoint = 1.2.3.4:51820\n'
|
|
'PersistentKeepalive = 25\n'
|
|
)
|
|
return wg
|
|
|
|
|
|
def _mock_config(domain=DOMAIN):
|
|
cfg = MagicMock()
|
|
cfg.configs = {
|
|
'_identity': {'domain': domain, 'cell_name': 'pic0', 'ip_range': '172.20.0.0/16'}
|
|
}
|
|
return cfg
|
|
|
|
|
|
# ─────────────────────────── fixtures ─────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def auth_mgr(tmp_path):
|
|
return _make_auth(tmp_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def peer_client(auth_mgr):
|
|
app.config['TESTING'] = True
|
|
app.config['SECRET_KEY'] = 'test-secret'
|
|
with patch('app.auth_manager', auth_mgr):
|
|
try:
|
|
import auth_routes
|
|
with patch.object(auth_routes, 'auth_manager', auth_mgr, create=True):
|
|
with app.test_client() as c:
|
|
r = _login(c, 'alice', 'AlicePass123!')
|
|
assert r.status_code == 200, f'peer login failed: {r.data}'
|
|
yield c
|
|
except (ImportError, AttributeError):
|
|
with app.test_client() as c:
|
|
r = _login(c, 'alice', 'AlicePass123!')
|
|
assert r.status_code == 200, f'peer login failed: {r.data}'
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_client(auth_mgr):
|
|
app.config['TESTING'] = True
|
|
app.config['SECRET_KEY'] = 'test-secret'
|
|
with patch('app.auth_manager', auth_mgr):
|
|
try:
|
|
import auth_routes
|
|
with patch.object(auth_routes, 'auth_manager', auth_mgr, create=True):
|
|
with app.test_client() as c:
|
|
r = _login(c, 'admin', 'AdminPass123!')
|
|
assert r.status_code == 200, f'admin login failed: {r.data}'
|
|
yield c
|
|
except (ImportError, AttributeError):
|
|
with app.test_client() as c:
|
|
r = _login(c, 'admin', 'AdminPass123!')
|
|
assert r.status_code == 200, f'admin login failed: {r.data}'
|
|
yield c
|
|
|
|
|
|
# ─────────────────── peer_dashboard field names ────────────────────────────────
|
|
|
|
class TestPeerDashboardFieldNames:
|
|
"""
|
|
peer_dashboard() must return the field names PeerDashboard.jsx reads.
|
|
A mismatch causes silent zeros/blanks in the UI without any error.
|
|
"""
|
|
|
|
def _get(self, peer_client):
|
|
wg = _mock_wg()
|
|
reg = _mock_registry()
|
|
cfg = _mock_config()
|
|
with patch('app.peer_registry', reg), \
|
|
patch('app.wireguard_manager', wg), \
|
|
patch('app.config_manager', cfg), \
|
|
patch('app._resolve_peer_dns', return_value='172.20.0.3'):
|
|
return peer_client.get('/api/peer/dashboard')
|
|
|
|
def test_returns_200(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.status_code == 200, r.data
|
|
|
|
def test_has_name_not_peer_name(self, peer_client):
|
|
"""PeerDashboard.jsx reads peer.name — must NOT be peer_name."""
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
assert 'name' in data, f"'name' missing from dashboard; keys: {list(data)}"
|
|
assert 'peer_name' not in data, \
|
|
"'peer_name' still present — UI reads 'name', not 'peer_name'"
|
|
|
|
def test_name_value_is_peer_username(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['name'] == 'alice'
|
|
|
|
def test_has_transfer_rx_not_rx_bytes(self, peer_client):
|
|
"""PeerDashboard.jsx reads peer.transfer_rx — must NOT be rx_bytes."""
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
assert 'transfer_rx' in data, f"'transfer_rx' missing; keys: {list(data)}"
|
|
assert 'rx_bytes' not in data, \
|
|
"'rx_bytes' still present — UI reads 'transfer_rx'"
|
|
|
|
def test_has_transfer_tx_not_tx_bytes(self, peer_client):
|
|
"""PeerDashboard.jsx reads peer.transfer_tx — must NOT be tx_bytes."""
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
assert 'transfer_tx' in data, f"'transfer_tx' missing; keys: {list(data)}"
|
|
assert 'tx_bytes' not in data, \
|
|
"'tx_bytes' still present — UI reads 'transfer_tx'"
|
|
|
|
def test_transfer_rx_value_from_wg_stats(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['transfer_rx'] == 1048576
|
|
|
|
def test_transfer_tx_value_from_wg_stats(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['transfer_tx'] == 524288
|
|
|
|
def test_has_service_urls(self, peer_client):
|
|
"""Dashboard must include service_urls so UI can render direct service links."""
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
assert 'service_urls' in data, f"'service_urls' missing; keys: {list(data)}"
|
|
assert isinstance(data['service_urls'], dict)
|
|
|
|
def test_service_urls_keyed_by_service(self, peer_client):
|
|
r = self._get(peer_client)
|
|
urls = r.get_json()['service_urls']
|
|
for svc in ('calendar', 'files', 'mail', 'webdav'):
|
|
assert svc in urls, f"service_urls missing '{svc}'; got: {list(urls)}"
|
|
|
|
def test_service_urls_use_configured_domain(self, peer_client):
|
|
r = self._get(peer_client)
|
|
urls = r.get_json()['service_urls']
|
|
assert urls['calendar'] == 'http://calendar.dev'
|
|
assert urls['files'] == 'http://files.dev'
|
|
assert urls['mail'] == 'http://mail.dev'
|
|
assert urls['webdav'] == 'http://webdav.dev'
|
|
|
|
def test_online_and_last_handshake_present(self, peer_client):
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
assert 'online' in data
|
|
assert 'last_handshake' in data
|
|
|
|
def test_peer_not_in_registry_returns_404(self, peer_client):
|
|
reg = _mock_registry(peer=None)
|
|
cfg = _mock_config()
|
|
with patch('app.peer_registry', reg), patch('app.config_manager', cfg):
|
|
r = peer_client.get('/api/peer/dashboard')
|
|
assert r.status_code == 404
|
|
|
|
|
|
# ─────────────────── peer_services structure ──────────────────────────────────
|
|
|
|
class TestPeerServicesStructure:
|
|
"""
|
|
peer_services() must return the exact structure MyServices.jsx reads.
|
|
All field-name mismatches cause silent blanks in the UI.
|
|
"""
|
|
|
|
def _get(self, peer_client, dns='172.20.0.3'):
|
|
wg = _mock_wg(dns)
|
|
reg = _mock_registry()
|
|
cfg = _mock_config()
|
|
with patch('app.peer_registry', reg), \
|
|
patch('app.wireguard_manager', wg), \
|
|
patch('app.config_manager', cfg), \
|
|
patch('app._resolve_peer_dns', return_value=dns):
|
|
return peer_client.get('/api/peer/services')
|
|
|
|
def test_returns_200(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.status_code == 200, r.data
|
|
|
|
# -- top-level keys --------------------------------------------------------
|
|
|
|
def test_has_username_at_top_level(self, peer_client):
|
|
"""MyServices.jsx uses data?.username for the WireGuard config download filename."""
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
assert 'username' in data, f"'username' missing from top level; keys: {list(data)}"
|
|
assert data['username'] == 'alice'
|
|
|
|
def test_has_files_not_webdav(self, peer_client):
|
|
"""MyServices.jsx reads data?.files — key must be 'files', not 'webdav'."""
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
assert 'files' in data, f"'files' key missing; keys: {list(data)}"
|
|
assert 'webdav' not in data, \
|
|
"'webdav' key still present — MyServices.jsx reads 'files'"
|
|
|
|
def test_has_wireguard_email_caldav_files(self, peer_client):
|
|
r = self._get(peer_client)
|
|
data = r.get_json()
|
|
for section in ('wireguard', 'email', 'caldav', 'files'):
|
|
assert section in data, f"'{section}' section missing; keys: {list(data)}"
|
|
|
|
# -- email section ---------------------------------------------------------
|
|
|
|
def test_email_address_not_username(self, peer_client):
|
|
"""MyServices.jsx reads email.address — must NOT be email.username."""
|
|
r = self._get(peer_client)
|
|
email = r.get_json()['email']
|
|
assert 'address' in email, f"'address' missing from email; keys: {list(email)}"
|
|
assert 'username' not in email, \
|
|
"'username' still in email section — MyServices.jsx reads 'address'"
|
|
|
|
def test_email_address_is_full_address(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['email']['address'] == 'alice@dev'
|
|
|
|
def test_email_has_nested_smtp(self, peer_client):
|
|
"""MyServices.jsx reads email.smtp.host and email.smtp.port as nested objects."""
|
|
r = self._get(peer_client)
|
|
email = r.get_json()['email']
|
|
assert 'smtp' in email, f"'smtp' missing from email; keys: {list(email)}"
|
|
smtp = email['smtp']
|
|
assert 'host' in smtp, f"'host' missing from email.smtp; keys: {list(smtp)}"
|
|
assert 'port' in smtp, f"'port' missing from email.smtp; keys: {list(smtp)}"
|
|
|
|
def test_email_has_nested_imap(self, peer_client):
|
|
"""MyServices.jsx reads email.imap.host and email.imap.port as nested objects."""
|
|
r = self._get(peer_client)
|
|
email = r.get_json()['email']
|
|
assert 'imap' in email, f"'imap' missing from email; keys: {list(email)}"
|
|
imap = email['imap']
|
|
assert 'host' in imap, f"'host' missing from email.imap; keys: {list(imap)}"
|
|
assert 'port' in imap, f"'port' missing from email.imap; keys: {list(imap)}"
|
|
|
|
def test_email_no_flat_host_fields(self, peer_client):
|
|
"""Flat imap_host/smtp_host fields must not be present."""
|
|
r = self._get(peer_client)
|
|
email = r.get_json()['email']
|
|
assert 'imap_host' not in email, \
|
|
"'imap_host' still flat — MyServices.jsx reads email.imap.host"
|
|
assert 'smtp_host' not in email, \
|
|
"'smtp_host' still flat — MyServices.jsx reads email.smtp.host"
|
|
|
|
def test_email_smtp_host_is_mail_domain(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['email']['smtp']['host'] == 'mail.dev'
|
|
|
|
def test_email_imap_host_is_mail_domain(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['email']['imap']['host'] == 'mail.dev'
|
|
|
|
def test_email_smtp_port(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['email']['smtp']['port'] == 587
|
|
|
|
def test_email_imap_port(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['email']['imap']['port'] == 993
|
|
|
|
# -- caldav section --------------------------------------------------------
|
|
|
|
def test_caldav_url_uses_calendar_subdomain(self, peer_client):
|
|
"""CalDAV URL must be http://calendar.{domain}, not radicale.{domain}:5232.
|
|
radicale.dev has no DNS record; calendar.dev is the Caddy-proxied entry."""
|
|
r = self._get(peer_client)
|
|
url = r.get_json()['caldav']['url']
|
|
assert 'radicale' not in url, \
|
|
f"CalDAV URL contains 'radicale' — no DNS record exists for radicale.dev; got: {url}"
|
|
assert ':5232' not in url, \
|
|
f"CalDAV URL exposes internal port 5232 — should use Caddy-proxied URL; got: {url}"
|
|
assert url == f'http://calendar.dev', f"CalDAV URL wrong: {url}"
|
|
|
|
def test_caldav_username_is_peer_name(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['caldav']['username'] == 'alice'
|
|
|
|
# -- files section ---------------------------------------------------------
|
|
|
|
def test_files_url_uses_files_subdomain(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['files']['url'] == 'http://files.dev'
|
|
|
|
def test_files_username_is_peer_name(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['files']['username'] == 'alice'
|
|
|
|
# -- wireguard section -----------------------------------------------------
|
|
|
|
def test_wireguard_dns_is_not_vpn_gateway(self, peer_client):
|
|
"""DNS must be the CoreDNS container IP, not the WireGuard VPN gateway 10.0.0.1.
|
|
10.0.0.1 is the WireGuard server-side VPN IP which does NOT serve DNS."""
|
|
r = self._get(peer_client)
|
|
dns = r.get_json()['wireguard'].get('dns', '')
|
|
assert dns != '10.0.0.1', \
|
|
"wireguard.dns is 10.0.0.1 (WireGuard VPN gateway) — should be CoreDNS IP"
|
|
|
|
def test_wireguard_dns_is_coredns_ip(self, peer_client):
|
|
"""DNS must be 172.20.0.3 (the CoreDNS container on the Docker bridge)."""
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['wireguard']['dns'] == '172.20.0.3'
|
|
|
|
def test_wireguard_has_config_field(self, peer_client):
|
|
"""wg.config field allows peer to download/copy their WireGuard config."""
|
|
r = self._get(peer_client)
|
|
wg = r.get_json()['wireguard']
|
|
assert 'config' in wg, f"'config' missing from wireguard section; keys: {list(wg)}"
|
|
|
|
def test_wireguard_config_has_dns_line(self, peer_client):
|
|
"""The config text must contain a DNS = line pointing to CoreDNS."""
|
|
r = self._get(peer_client)
|
|
config = r.get_json()['wireguard'].get('config', '')
|
|
assert 'DNS = 172.20.0.3' in config, \
|
|
f"Config missing 'DNS = 172.20.0.3'; config:\n{config}"
|
|
|
|
def test_wireguard_config_has_interface_section(self, peer_client):
|
|
r = self._get(peer_client)
|
|
config = r.get_json()['wireguard'].get('config', '')
|
|
assert '[Interface]' in config and '[Peer]' in config, \
|
|
f"Config missing [Interface] or [Peer] section; config:\n{config}"
|
|
|
|
def test_wireguard_config_has_full_tunnel_allowed_ips(self, peer_client):
|
|
"""Full-tunnel peers must have AllowedIPs = 0.0.0.0/0 so all traffic goes via VPN."""
|
|
r = self._get(peer_client)
|
|
config = r.get_json()['wireguard'].get('config', '')
|
|
assert '0.0.0.0/0' in config, \
|
|
f"Config missing 0.0.0.0/0 AllowedIPs for full-tunnel peer; config:\n{config}"
|
|
|
|
def test_wireguard_has_ip_field(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert 'ip' in r.get_json()['wireguard']
|
|
|
|
def test_wireguard_ip_is_peer_vpn_ip(self, peer_client):
|
|
r = self._get(peer_client)
|
|
assert r.get_json()['wireguard']['ip'] == '10.0.0.5'
|
|
|
|
|
|
# ─────────────────── auth / access control ────────────────────────────────────
|
|
|
|
class TestPeerEndpointAccessControl:
|
|
"""Peer-only routes must block unauthenticated and admin sessions."""
|
|
|
|
def test_unauthenticated_dashboard_returns_401(self, auth_mgr):
|
|
app.config['TESTING'] = True
|
|
app.config['SECRET_KEY'] = 'test-secret'
|
|
with patch('app.auth_manager', auth_mgr):
|
|
with app.test_client() as c:
|
|
r = c.get('/api/peer/dashboard')
|
|
assert r.status_code == 401
|
|
|
|
def test_unauthenticated_services_returns_401(self, auth_mgr):
|
|
app.config['TESTING'] = True
|
|
app.config['SECRET_KEY'] = 'test-secret'
|
|
with patch('app.auth_manager', auth_mgr):
|
|
with app.test_client() as c:
|
|
r = c.get('/api/peer/services')
|
|
assert r.status_code == 401
|
|
|
|
def test_admin_dashboard_returns_403(self, admin_client):
|
|
r = admin_client.get('/api/peer/dashboard')
|
|
assert r.status_code == 403, \
|
|
f"Admin accessing peer-only /api/peer/dashboard should get 403, got {r.status_code}"
|
|
|
|
def test_admin_services_returns_403(self, admin_client):
|
|
r = admin_client.get('/api/peer/services')
|
|
assert r.status_code == 403, \
|
|
f"Admin accessing peer-only /api/peer/services should get 403, got {r.status_code}"
|
|
|
|
|
|
# ─────────────────── DNS zone records ─────────────────────────────────────────
|
|
|
|
class TestDNSZoneRecords:
|
|
"""
|
|
Verify that network_manager._build_dns_records() generates the correct IPs.
|
|
|
|
All service names now resolve to the WG server IP (10.0.0.1) rather than
|
|
Docker VIPs. ensure_service_dnat() routes wg0:80 → Caddy; Caddy routes to
|
|
the correct backend by Host header. This allows cross-cell peers to reach
|
|
services without Docker bridge subnet conflicts.
|
|
"""
|
|
|
|
_WG_SERVER_IP = '10.0.0.1'
|
|
|
|
def setUp(self):
|
|
pass
|
|
|
|
def _records(self, ip_range='172.20.0.0/16', cell_name='pic0'):
|
|
import network_manager as nm
|
|
mgr = nm.NetworkManager.__new__(nm.NetworkManager)
|
|
return mgr._build_dns_records(cell_name, ip_range)
|
|
|
|
def test_api_resolves_to_wg_server_ip(self):
|
|
records = self._records()
|
|
api_rec = next((r for r in records if r['name'] == 'api'), None)
|
|
assert api_rec is not None, "No DNS record for 'api'"
|
|
assert api_rec['value'] == self._WG_SERVER_IP, (
|
|
f"api.dev should resolve to WG server IP ({self._WG_SERVER_IP}); "
|
|
f"got {api_rec['value']}"
|
|
)
|
|
|
|
def test_webui_resolves_to_wg_server_ip(self):
|
|
records = self._records()
|
|
rec = next((r for r in records if r['name'] == 'webui'), None)
|
|
assert rec is not None, "No DNS record for 'webui'"
|
|
assert rec['value'] == self._WG_SERVER_IP, (
|
|
f"webui.dev should resolve to WG server IP ({self._WG_SERVER_IP}); "
|
|
f"got {rec['value']}"
|
|
)
|
|
|
|
def test_calendar_resolves_to_wg_server_ip(self):
|
|
records = self._records()
|
|
rec = next((r for r in records if r['name'] == 'calendar'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"calendar.dev should resolve to WG server IP; got {rec}"
|
|
|
|
def test_files_resolves_to_wg_server_ip(self):
|
|
records = self._records()
|
|
rec = next((r for r in records if r['name'] == 'files'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"files.dev should resolve to WG server IP; got {rec}"
|
|
|
|
def test_mail_resolves_to_wg_server_ip(self):
|
|
records = self._records()
|
|
rec = next((r for r in records if r['name'] == 'mail'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"mail.dev should resolve to WG server IP; got {rec}"
|
|
|
|
def test_webmail_resolves_to_wg_server_ip(self):
|
|
records = self._records()
|
|
rec = next((r for r in records if r['name'] == 'webmail'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"webmail.dev should resolve to WG server IP; got {rec}"
|
|
|
|
def test_webdav_resolves_to_wg_server_ip(self):
|
|
records = self._records()
|
|
rec = next((r for r in records if r['name'] == 'webdav'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"webdav.dev should resolve to WG server IP; got {rec}"
|
|
|
|
def test_cell_name_resolves_to_wg_server_ip(self):
|
|
records = self._records(cell_name='mypic')
|
|
rec = next((r for r in records if r['name'] == 'mypic'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"mypic.dev should resolve to WG server IP; got {rec}"
|
|
|
|
def test_all_records_are_type_a(self):
|
|
records = self._records()
|
|
for rec in records:
|
|
assert rec.get('type') == 'A', f"Record {rec} is not type A"
|
|
|
|
|
|
class TestDNSZoneRecordsWithPytest:
|
|
"""Same as above but using pytest-style (no setUp/tearDown)."""
|
|
|
|
_WG_SERVER_IP = '10.0.0.1'
|
|
|
|
@pytest.fixture
|
|
def records(self):
|
|
import network_manager as nm
|
|
mgr = nm.NetworkManager.__new__(nm.NetworkManager)
|
|
return mgr._build_dns_records('pic0', '172.20.0.0/16')
|
|
|
|
def test_api_resolves_to_wg_server_ip(self, records):
|
|
rec = next((r for r in records if r['name'] == 'api'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"api.dev should point to WG server IP ({self._WG_SERVER_IP}); got {rec}"
|
|
|
|
def test_webui_resolves_to_wg_server_ip(self, records):
|
|
rec = next((r for r in records if r['name'] == 'webui'), None)
|
|
assert rec and rec['value'] == self._WG_SERVER_IP, \
|
|
f"webui.dev should point to WG server IP ({self._WG_SERVER_IP}); got {rec}"
|
|
|
|
|
|
# ─────────────────── Caddyfile generation ─────────────────────────────────────
|
|
|
|
class TestCaddyfileGeneration:
|
|
"""
|
|
write_caddyfile() must produce a Caddyfile that Caddy can use to route
|
|
all service domains including webui.dev.
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def caddyfile(self, tmp_path):
|
|
from ip_utils import write_caddyfile
|
|
path = str(tmp_path / 'Caddyfile')
|
|
write_caddyfile('172.20.0.0/16', 'pic0', 'dev', path)
|
|
with open(path) as f:
|
|
return f.read()
|
|
|
|
def test_main_domain_block_present(self, caddyfile):
|
|
assert 'http://pic0.dev' in caddyfile
|
|
|
|
def test_api_block_present(self, caddyfile):
|
|
assert 'http://api.dev' in caddyfile
|
|
|
|
def test_webui_block_present(self, caddyfile):
|
|
assert 'http://webui.dev' in caddyfile, \
|
|
"Missing webui.dev Caddy block — webui is unreachable by domain name"
|
|
|
|
def test_calendar_block_present(self, caddyfile):
|
|
assert 'http://calendar.dev' in caddyfile
|
|
|
|
def test_files_block_present(self, caddyfile):
|
|
assert 'http://files.dev' in caddyfile
|
|
|
|
def test_mail_block_present(self, caddyfile):
|
|
assert 'http://mail.dev' in caddyfile
|
|
|
|
def test_webdav_block_present(self, caddyfile):
|
|
assert 'http://webdav.dev' in caddyfile
|
|
|
|
def test_caddy_vips_present(self, caddyfile):
|
|
assert '172.20.0.21' in caddyfile, "calendar VIP missing from Caddyfile"
|
|
assert '172.20.0.22' in caddyfile, "files VIP missing from Caddyfile"
|
|
assert '172.20.0.23' in caddyfile, "mail VIP missing from Caddyfile"
|
|
assert '172.20.0.24' in caddyfile, "webdav VIP missing from Caddyfile"
|
|
|
|
def test_no_radicale_subdomain_in_caddyfile(self, caddyfile):
|
|
"""radicale.dev has no DNS record; CalDAV should go via calendar.dev."""
|
|
assert 'radicale.dev' not in caddyfile, \
|
|
"radicale.dev should not appear in Caddyfile — no DNS record for it"
|
|
|
|
def test_auto_https_off(self, caddyfile):
|
|
assert 'auto_https off' in caddyfile
|
|
|
|
def test_reverse_proxy_targets_use_container_names(self, caddyfile):
|
|
"""Container-internal routing must use service names not IPs."""
|
|
assert 'cell-api:3000' in caddyfile
|
|
assert 'cell-radicale:5232' in caddyfile
|
|
assert 'cell-webui:80' in caddyfile
|