feat(e2e): add mode configs (demo/cloud/local) with Directus JWT auth
This commit is contained in:
parent
39d8c2f006
commit
20c776260f
4 changed files with 165 additions and 0 deletions
76
tests/e2e/modes/cloud.py
Normal file
76
tests/e2e/modes/cloud.py
Normal file
|
|
@ -0,0 +1,76 @@
|
|||
"""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",
|
||||
auth_setup=_cloud_auth_setup,
|
||||
expected_failures=[],
|
||||
results_dir=Path("tests/e2e/results/cloud"),
|
||||
settings_tabs=_BASE_SETTINGS_TABS,
|
||||
)
|
||||
25
tests/e2e/modes/demo.py
Normal file
25
tests/e2e/modes/demo.py
Normal file
|
|
@ -0,0 +1,25 @@
|
|||
"""Demo mode config — port 8504, DEMO_MODE=true, LLM/scraping neutered."""
|
||||
from pathlib import Path
|
||||
from tests.e2e.models import ModeConfig
|
||||
|
||||
_BASE_SETTINGS_TABS = [
|
||||
"👤 My Profile", "📝 Resume Profile", "🔎 Search",
|
||||
"⚙️ System", "🎯 Fine-Tune", "🔑 License", "💾 Data",
|
||||
]
|
||||
|
||||
DEMO = ModeConfig(
|
||||
name="demo",
|
||||
base_url="http://localhost:8504",
|
||||
auth_setup=lambda ctx: None,
|
||||
expected_failures=[
|
||||
"Fetch*",
|
||||
"Generate Cover Letter*",
|
||||
"Generate*",
|
||||
"Analyze Screenshot*",
|
||||
"Push to Calendar*",
|
||||
"Sync Email*",
|
||||
"Start Email Sync*",
|
||||
],
|
||||
results_dir=Path("tests/e2e/results/demo"),
|
||||
settings_tabs=_BASE_SETTINGS_TABS,
|
||||
)
|
||||
17
tests/e2e/modes/local.py
Normal file
17
tests/e2e/modes/local.py
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
"""Local mode config — port 8502, full features, no auth."""
|
||||
from pathlib import Path
|
||||
from tests.e2e.models import ModeConfig
|
||||
|
||||
_BASE_SETTINGS_TABS = [
|
||||
"👤 My Profile", "📝 Resume Profile", "🔎 Search",
|
||||
"⚙️ System", "🎯 Fine-Tune", "🔑 License", "💾 Data",
|
||||
]
|
||||
|
||||
LOCAL = ModeConfig(
|
||||
name="local",
|
||||
base_url="http://localhost:8502",
|
||||
auth_setup=lambda ctx: None,
|
||||
expected_failures=[],
|
||||
results_dir=Path("tests/e2e/results/local"),
|
||||
settings_tabs=_BASE_SETTINGS_TABS,
|
||||
)
|
||||
|
|
@ -4,6 +4,7 @@ import pytest
|
|||
from unittest.mock import patch, MagicMock
|
||||
import time
|
||||
from tests.e2e.models import ErrorRecord, ModeConfig, diff_errors
|
||||
import tests.e2e.modes.cloud as cloud_mod # imported early so load_dotenv runs before any monkeypatch
|
||||
|
||||
|
||||
def test_error_record_equality():
|
||||
|
|
@ -62,3 +63,49 @@ def test_mode_config_no_expected_failures():
|
|||
settings_tabs=[],
|
||||
)
|
||||
assert not config.matches_expected_failure("Fetch New Jobs")
|
||||
|
||||
|
||||
def test_get_jwt_strategy_b_fallback(monkeypatch):
|
||||
"""Falls back to persistent JWT when no email env var set."""
|
||||
monkeypatch.delenv("E2E_DIRECTUS_EMAIL", raising=False)
|
||||
monkeypatch.setenv("E2E_DIRECTUS_JWT", "persistent.jwt.token")
|
||||
cloud_mod._token_cache.update({"token": None, "expires_at": 0.0})
|
||||
assert cloud_mod._get_jwt() == "persistent.jwt.token"
|
||||
|
||||
|
||||
def test_get_jwt_strategy_b_raises_if_no_token(monkeypatch):
|
||||
"""Raises if neither email nor JWT env var is set."""
|
||||
monkeypatch.delenv("E2E_DIRECTUS_EMAIL", raising=False)
|
||||
monkeypatch.delenv("E2E_DIRECTUS_JWT", raising=False)
|
||||
cloud_mod._token_cache.update({"token": None, "expires_at": 0.0})
|
||||
with pytest.raises(RuntimeError, match="Cloud mode requires"):
|
||||
cloud_mod._get_jwt()
|
||||
|
||||
|
||||
def test_get_jwt_strategy_a_login(monkeypatch):
|
||||
"""Strategy A: calls Directus /auth/login and caches token."""
|
||||
monkeypatch.setenv("E2E_DIRECTUS_EMAIL", "e2e@circuitforge.tech")
|
||||
monkeypatch.setenv("E2E_DIRECTUS_PASSWORD", "testpass")
|
||||
monkeypatch.setenv("E2E_DIRECTUS_URL", "http://fake-directus:8055")
|
||||
cloud_mod._token_cache.update({"token": None, "expires_at": 0.0})
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"data": {"access_token": "fresh.jwt", "expires": 900_000}}
|
||||
mock_resp.raise_for_status = lambda: None
|
||||
|
||||
with patch("tests.e2e.modes.cloud.requests.post", return_value=mock_resp) as mock_post:
|
||||
token = cloud_mod._get_jwt()
|
||||
|
||||
assert token == "fresh.jwt"
|
||||
mock_post.assert_called_once()
|
||||
assert cloud_mod._token_cache["token"] == "fresh.jwt"
|
||||
|
||||
|
||||
def test_get_jwt_uses_cache(monkeypatch):
|
||||
"""Returns cached token if not yet expired."""
|
||||
monkeypatch.setenv("E2E_DIRECTUS_EMAIL", "e2e@circuitforge.tech")
|
||||
cloud_mod._token_cache.update({"token": "cached.jwt", "expires_at": time.time() + 500})
|
||||
with patch("tests.e2e.modes.cloud.requests.post") as mock_post:
|
||||
token = cloud_mod._get_jwt()
|
||||
assert token == "cached.jwt"
|
||||
mock_post.assert_not_called()
|
||||
|
|
|
|||
Loading…
Reference in a new issue