Fix post-deploy auth issues: best-effort service provisioning, integration test auth, test mock corrections
- api/app.py: email/calendar/files provisioning now best-effort (non-fatal); fixed email_manager.create_email_user call to include domain argument - tests/integration: added module-level auth sessions to all integration test files; added admin auth to api fixture and _resolve_admin_pass() helper; added TEST_PEER_PASSWORD constant; added password to peer creation calls - tests/test_peer_provisioning.py: renamed rollback test to reflect new best-effort semantics (email failure no longer causes rollback) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -2,10 +2,12 @@
|
||||
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_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
|
||||
@@ -17,6 +19,8 @@ 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}"
|
||||
@@ -28,11 +32,33 @@ TEST_PEERS = (
|
||||
'bad-svc-peer', # guard against validation-test leak
|
||||
)
|
||||
|
||||
TEST_PEER_PASSWORD = 'IntegrationTest123!'
|
||||
|
||||
|
||||
def _resolve_admin_pass() -> str:
|
||||
if ADMIN_PASS:
|
||||
return ADMIN_PASS
|
||||
# Try reading from the initial password file (present on first run before bootstrap)
|
||||
candidate = os.path.join(
|
||||
os.path.dirname(__file__), '..', '..', 'data', 'api', '.admin_initial_password'
|
||||
)
|
||||
candidate = os.path.normpath(candidate)
|
||||
if os.path.exists(candidate):
|
||||
return open(candidate).read().strip()
|
||||
raise RuntimeError(
|
||||
"Admin password unknown. Set PIC_ADMIN_PASS env var or run make setup first."
|
||||
)
|
||||
|
||||
|
||||
@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
|
||||
|
||||
|
||||
@@ -71,17 +97,13 @@ def peer_rules(peer_ip: str) -> list[str]:
|
||||
return [line for line in iptables_forward().splitlines() if comment in line]
|
||||
|
||||
|
||||
def get_live_service_vips() -> dict:
|
||||
def get_live_service_vips(session: requests.Session = None) -> 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.
|
||||
Read virtual IPs from the config API using an authenticated session.
|
||||
Falls back to a new unauthenticated request only if no session provided (legacy).
|
||||
"""
|
||||
cfg = requests.get(f"{API_BASE}/api/config").json()
|
||||
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', ''),
|
||||
|
||||
Reference in New Issue
Block a user