Files
pic/tests/integration/conftest.py
T
roof 3ce45a8911 fix: get_live_service_vips uses config API, require CIDR prefix for ip_range
- tests/integration/conftest.py: get_live_service_vips() now reads from the
  config API's service_ips field instead of docker exec.  The docker exec approach
  spawns a fresh Python process that imports firewall_manager with its hardcoded
  initial SERVICE_IPS, ignoring any update_service_ips() calls made at runtime.
  The config API always computes VIPs from the current ip_range, so it matches what
  the running app actually uses when writing iptables rules.

- api/app.py: reject ip_range values without a CIDR prefix (e.g. '10.0.0.1')
  with a 400.  Bare IPs are parsed as /32 by ipaddress.ip_network(strict=False),
  which shifts all VIP offsets and produces unusable Docker subnet configs.

- tests/integration/test_config_api.py: update bare-ip test to expect 400 now
  that the API enforces the prefix requirement.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-24 04:54:47 -04:00

92 lines
2.7 KiB
Python

"""
Shared fixtures for live integration tests.
Configure with environment variables:
PIC_HOST API host (default: localhost)
PIC_API_PORT API port (default: 3000)
PIC_WEBUI_PORT WebUI port (default: 80)
PIC_WG_CONTAINER WireGuard container name (default: cell-wireguard)
"""
import os
import json
import subprocess
import pytest
import requests
PIC_HOST = os.environ.get('PIC_HOST', 'localhost')
API_PORT = int(os.environ.get('PIC_API_PORT', '3000'))
WEBUI_PORT = int(os.environ.get('PIC_WEBUI_PORT', '80'))
WG_CONTAINER = os.environ.get('PIC_WG_CONTAINER', 'cell-wireguard')
API_BASE = f"http://{PIC_HOST}:{API_PORT}"
WEBUI_BASE = f"http://{PIC_HOST}:{WEBUI_PORT}"
TEST_PEERS = (
'integration-test-full',
'integration-test-restricted',
'integration-test-none',
'bad-svc-peer', # guard against validation-test leak
)
@pytest.fixture(scope='session')
def api():
s = requests.Session()
s.headers['Content-Type'] = 'application/json'
return s
@pytest.fixture(scope='session')
def api_base():
return API_BASE
@pytest.fixture(scope='session')
def webui_base():
return WEBUI_BASE
@pytest.fixture(scope='session', autouse=True)
def cleanup_test_peers(api):
"""Delete any leftover test peers before and after the entire session."""
for name in TEST_PEERS:
api.delete(f"{API_BASE}/api/peers/{name}")
yield
for name in TEST_PEERS:
api.delete(f"{API_BASE}/api/peers/{name}")
def iptables_forward() -> str:
"""Return iptables-save output from the WireGuard container."""
result = subprocess.run(
['docker', 'exec', WG_CONTAINER, 'iptables-save'],
capture_output=True, text=True, timeout=10,
)
return result.stdout
def peer_rules(peer_ip: str) -> list[str]:
"""Return FORWARD rule lines for a specific peer IP."""
comment = f'pic-peer-{peer_ip.replace(".", "-")}'
return [line for line in iptables_forward().splitlines() if comment in line]
def get_live_service_vips() -> dict:
"""
Read virtual IPs from the config API.
The config API computes service_ips from the current ip_range at request time,
so it always matches what the running firewall_manager will use when applying
peer rules. Using docker exec on the API container is NOT reliable because
it spawns a fresh Python process that imports firewall_manager with its initial
hardcoded SERVICE_IPS, ignoring any update_service_ips() calls made at runtime.
"""
cfg = requests.get(f"{API_BASE}/api/config").json()
sips = cfg.get('service_ips', {})
return {
'calendar': sips.get('vip_calendar', ''),
'files': sips.get('vip_files', ''),
'mail': sips.get('vip_mail', ''),
'webdav': sips.get('vip_webdav', ''),
}