peregrine/tests/e2e/modes/cloud.py
pyr0ball cb8afa6539 fix(e2e): cloud auth via cookie, local port, Playwright WebSocket gotcha
E2E harness fixes to get all three modes (demo/cloud/local) passing:

- conftest.py: use ctx.add_cookies() for cloud auth instead of
  ctx.route() or set_extra_http_headers(). Playwright's route() only
  intercepts HTTP; set_extra_http_headers() explicitly excludes
  WebSocket handshakes. Streamlit reads st.context.headers from the
  WebSocket upgrade, so cookies are the only vehicle that reaches it
  without a reverse proxy.

- cloud_session.py: fall back to Cookie header when X-CF-Session is
  absent — supports direct access (E2E tests, dev without Caddy).
  In production Caddy sets X-CF-Session; in tests the cf_session cookie
  is set on the browser context and arrives in the Cookie header.

- modes/cloud.py: add /peregrine base URL path (STREAMLIT_SERVER_BASE_URL_PATH=peregrine)

- modes/local.py: correct port from 8502 → 8501 and add /peregrine path

All three modes now pass smoke + interaction tests clean.
2026-03-17 20:01:42 -07:00

76 lines
2.3 KiB
Python

"""Cloud mode config — port 8505, CLOUD_MODE=true, Directus JWT auth."""
from __future__ import annotations
import os
import time
import logging
from pathlib import Path
from typing import Any
import requests
from dotenv import load_dotenv
from tests.e2e.models import ModeConfig
load_dotenv(".env.e2e")
log = logging.getLogger(__name__)
_BASE_SETTINGS_TABS = [
"👤 My Profile", "📝 Resume Profile", "🔎 Search",
"⚙️ System", "🎯 Fine-Tune", "🔑 License", "💾 Data", "🔒 Privacy",
]
_token_cache: dict[str, Any] = {"token": None, "expires_at": 0.0}
def _get_jwt() -> str:
"""
Acquire a Directus JWT for the e2e test user.
Strategy A: user/pass login (preferred).
Strategy B: persistent JWT from E2E_DIRECTUS_JWT env var.
Caches the token and refreshes 100s before expiry.
"""
if not os.environ.get("E2E_DIRECTUS_EMAIL"):
jwt = os.environ.get("E2E_DIRECTUS_JWT", "")
if not jwt:
raise RuntimeError(
"Cloud mode requires E2E_DIRECTUS_EMAIL+PASSWORD or E2E_DIRECTUS_JWT in .env.e2e"
)
return jwt
if _token_cache["token"] and time.time() < _token_cache["expires_at"] - 100:
return _token_cache["token"]
directus_url = os.environ.get("E2E_DIRECTUS_URL", "http://172.31.0.2:8055")
resp = requests.post(
f"{directus_url}/auth/login",
json={
"email": os.environ["E2E_DIRECTUS_EMAIL"],
"password": os.environ["E2E_DIRECTUS_PASSWORD"],
},
timeout=10,
)
resp.raise_for_status()
data = resp.json()["data"]
token = data["access_token"]
expires_in_ms = data.get("expires", 900_000)
_token_cache["token"] = token
_token_cache["expires_at"] = time.time() + (expires_in_ms / 1000)
log.info("Acquired Directus JWT (expires in %ds)", expires_in_ms // 1000)
return token
def _cloud_auth_setup(context: Any) -> None:
"""Placeholder — actual JWT injection done via context.route() in conftest."""
pass # Route-based injection set up in conftest.py mode_contexts fixture
CLOUD = ModeConfig(
name="cloud",
base_url="http://localhost:8505/peregrine",
auth_setup=_cloud_auth_setup,
expected_failures=[],
results_dir=Path("tests/e2e/results/cloud"),
settings_tabs=_BASE_SETTINGS_TABS,
)