roof
9677755b4f
- Fix make test: was pointing to non-existent api/tests/, now runs unit tests correctly with --ignore=e2e --ignore=integration - Remove dead phase test targets (test-phase1..4, test-all-phases) that all referenced cd api && pytest tests/ (non-existent path) - Add .test_admin_pass file: reset_admin_password.py now writes a persistent test password file alongside .admin_initial_password; the API never deletes it (unlike .admin_initial_password which is consumed on first startup) - Update both integration/conftest.py and e2e/helpers/admin_password.py to read .test_admin_pass before .admin_initial_password — so tests work after make restart without needing PIC_ADMIN_PASS env var - Add AI collaboration rules to CLAUDE.md (auto-loaded every session) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
115 lines
3.6 KiB
Python
115 lines
3.6 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)
|
|
PIC_ADMIN_USER Admin username (default: admin)
|
|
PIC_ADMIN_PASS Admin password (default: read from data/api/.admin_initial_password or env)
|
|
"""
|
|
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')
|
|
ADMIN_USER = os.environ.get('PIC_ADMIN_USER', 'admin')
|
|
ADMIN_PASS = os.environ.get('PIC_ADMIN_PASS', '')
|
|
|
|
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
|
|
)
|
|
|
|
TEST_PEER_PASSWORD = 'IntegrationTest123!'
|
|
|
|
|
|
_DATA_DIR = os.path.normpath(os.path.join(os.path.dirname(__file__), '..', '..', 'data', 'api'))
|
|
|
|
|
|
def _resolve_admin_pass() -> str:
|
|
if ADMIN_PASS:
|
|
return ADMIN_PASS
|
|
for fname in ('.test_admin_pass', '.admin_initial_password'):
|
|
candidate = os.path.join(_DATA_DIR, fname)
|
|
if os.path.exists(candidate):
|
|
return open(candidate).read().strip()
|
|
raise RuntimeError(
|
|
"Admin password unknown. Set PIC_ADMIN_PASS env var or run: "
|
|
"make reset-test-admin-pass PIC_TEST_ADMIN_PASS=<password>"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def api():
|
|
"""Authenticated requests.Session logged in as admin."""
|
|
s = requests.Session()
|
|
s.headers['Content-Type'] = 'application/json'
|
|
password = _resolve_admin_pass()
|
|
r = s.post(f"{API_BASE}/api/auth/login", json={'username': ADMIN_USER, 'password': password})
|
|
if r.status_code != 200:
|
|
raise RuntimeError(f"Integration test login failed: {r.status_code} {r.text}")
|
|
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(session: requests.Session = None) -> dict:
|
|
"""
|
|
Read virtual IPs from the config API using an authenticated session.
|
|
Falls back to a new unauthenticated request only if no session provided (legacy).
|
|
"""
|
|
s = session or requests.Session()
|
|
cfg = s.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', ''),
|
|
}
|