Compare commits

..

11 commits

Author SHA1 Message Date
8d9c5782fd chore(release): v0.6.2
Some checks failed
CI / test (push) Has been cancelled
Bugfix release — demo mode error fixes + Playwright E2E test harness.
See CHANGELOG.md for full details.
2026-03-17 20:08:55 -07:00
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
5d14542142 feat(e2e): add smoke + interaction tests; fix two demo mode errors
- Add tests/e2e/test_smoke.py: page-load error check for all pages
- Add tests/e2e/test_interactions.py: click every interactable, diff
  errors, XFAIL expected demo failures, flag regressions as XPASS
- Fix conftest get_page_errors() to use text_content() instead of
  inner_text() so errors inside collapsed expanders are captured with
  their actual message text (inner_text respects CSS display:none)
- Fix tests/e2e/modes/demo.py base_url to include /peregrine path prefix
  (STREAMLIT_SERVER_BASE_URL_PATH=peregrine set in demo container)

App fixes surfaced by the harness:
- task_runner.py: add DEMO_MODE guard for discovery task — previously
  crashed with FileNotFoundError on search_profiles.yaml before any
  demo guard could fire; now returns friendly error immediately
- 6_Interview_Prep.py: stop auto-triggering LLM session on page load
  in demo mode; show "AI features disabled" info instead, preventing
  a silent st.error() inside the collapsed Practice Q&A expander

Both smoke and interaction tests now pass clean against demo mode.
2026-03-17 07:00:54 -07:00
a55e09d30b feat(e2e): add conftest with Streamlit helpers, browser fixtures, console filter 2026-03-16 23:14:24 -07:00
0cdd97f1c0 feat(e2e): add BasePage and 7 page objects
BasePage provides navigation, error capture, and interactable discovery
with fnmatch-based expected_failure matching. SettingsPage extends it
with tab-aware discovery. All conftest imports are deferred to method
bodies so the module loads without a live browser fixture.
2026-03-16 23:14:20 -07:00
20c776260f feat(e2e): add mode configs (demo/cloud/local) with Directus JWT auth 2026-03-16 23:07:34 -07:00
39d8c2f006 feat(e2e): add ErrorRecord, ModeConfig, diff_errors models with tests 2026-03-16 23:06:02 -07:00
5efc6d48eb chore(e2e): scaffold E2E harness directory and install deps
Add pytest-playwright and pytest-json-report to requirements.txt; create
tests/e2e/ skeleton (modes/, pages/, results/) with __init__.py files and
.gitkeep; add results subdirs to .gitignore.
2026-03-16 22:58:47 -07:00
fadbabbaf4 chore(e2e): add xvfb-run wrapper for headed debugging sessions
Adds `e2e` subcommand to manage.sh supporting headless (default) and
headed (E2E_HEADLESS=false via xvfb-run) Playwright test runs with
per-mode JSON report output under tests/e2e/results/<mode>/.
2026-03-16 22:57:21 -07:00
3baed0c9b2 feat(e2e): add E2E test harness implementation plan
Multi-mode Playwright/pytest plan covering demo/cloud/local.
Addresses reviewer feedback: test isolation, JWT route refresh,
2000ms settle window, stAlert detection, tab collision fix,
instance availability guard, background_tasks seeding.
2026-03-16 22:53:49 -07:00
ce5f7d09c5 chore(e2e): add .env.e2e.example and gitignore .env.e2e
Committed credential template for E2E harness setup.
Directus e2e test user provisioned: e2e@circuitforge.tech (user ID: e2c224f7-a2dd-481f-bb3e-e2a5674f8337).
2026-03-16 22:41:24 -07:00
29 changed files with 2520 additions and 2 deletions

16
.env.e2e.example Normal file
View file

@ -0,0 +1,16 @@
# Peregrine E2E test harness credentials
# Copy to .env.e2e and fill in real values — .env.e2e is gitignored
HEIMDALL_ADMIN_TOKEN=changeme
HEIMDALL_URL=http://localhost:8900
# Cloud auth — Strategy A (preferred): Directus user/pass → fresh JWT per run
E2E_DIRECTUS_EMAIL=e2e@circuitforge.tech
E2E_DIRECTUS_PASSWORD=changeme
E2E_DIRECTUS_URL=http://172.31.0.2:8055
# Cloud auth — Strategy B (fallback): persistent JWT (uncomment to use)
# E2E_DIRECTUS_JWT=changeme
E2E_HEADLESS=true
E2E_SLOW_MO=0

6
.gitignore vendored
View file

@ -48,3 +48,9 @@ demo/seed_demo.py
# Git worktrees
.worktrees/
.env.e2e
# E2E test result artifacts
tests/e2e/results/demo/
tests/e2e/results/cloud/
tests/e2e/results/local/

View file

@ -9,6 +9,38 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
---
## [0.6.2] — 2026-03-18
### Added
- **Playwright E2E test harness** — smoke + interaction test suite covering all
three Peregrine instances (demo / cloud / local). Navigates every page, checks
for DOM errors on load, clicks every interactable element, diffs errors
before/after each click, and XFAIL-marks expected demo-mode failures so
neutering-guard regressions are surfaced as XPASSes. Screenshots on failure.
- `tests/e2e/test_smoke.py` — page-load error detection
- `tests/e2e/test_interactions.py` — full click-through with XFAIL/XPASS bucketing
- `tests/e2e/conftest.py` — Streamlit-aware wait helpers, error scanner, fixtures
- `tests/e2e/models.py``ErrorRecord`, `ModeConfig`, `diff_errors`
- `tests/e2e/modes/` — per-mode configs (demo / cloud / local)
- `tests/e2e/pages/` — page objects for all 7 pages including Settings tabs
### Fixed
- **Demo: "Discovery failed" error on Home page load**`task_runner.py` now
checks `DEMO_MODE` before importing `discover.py`; returns a friendly error
immediately instead of crashing on missing `search_profiles.yaml` (#21)
- **Demo: silent `st.error()` in collapsed Practice Q&A expander** — Interview
Prep no longer auto-triggers the LLM on page render in demo mode; shows an
`st.info` placeholder instead, eliminating the hidden error element (#22)
- **Cloud: auth wall shown to E2E test browser**`cloud_session.py` now falls
back to the `Cookie` header when `X-CF-Session` is absent (direct access
without Caddy). Playwright's `set_extra_http_headers()` does not propagate to
WebSocket handshakes; cookies do. Test harness uses `ctx.add_cookies()`.
- **E2E error scanner returned empty text for collapsed expanders** — switched
from `inner_text()` (respects CSS `display:none`) to `text_content()` so
errors inside collapsed Streamlit expanders are captured with their full text.
---
## [0.6.1] — 2026-03-16
### Fixed

View file

@ -151,7 +151,12 @@ def resolve_session(app: str = "peregrine") -> None:
if st.session_state.get("user_id"):
return
cookie_header = st.context.headers.get("x-cf-session", "")
# Primary: Caddy injects X-CF-Session header in production.
# Fallback: direct access (E2E tests, dev without Caddy) reads the cookie header.
cookie_header = (
st.context.headers.get("x-cf-session", "")
or st.context.headers.get("cookie", "")
)
session_jwt = _extract_session_token(cookie_header)
if not session_jwt:
_render_auth_wall("Please sign in to access Peregrine.")

View file

@ -224,7 +224,11 @@ with col_prep:
st.markdown(msg["content"])
# Initial question if session is empty
if not st.session_state[qa_key]:
import os as _os
_is_demo = _os.environ.get("DEMO_MODE", "").lower() in ("1", "true", "yes")
if not st.session_state[qa_key] and _is_demo:
st.info("AI features are disabled in the public demo. Run your own instance to use Practice Q&A.")
elif not st.session_state[qa_key]:
with st.spinner("Setting up your mock interview…"):
try:
from scripts.llm_router import complete

File diff suppressed because it is too large Load diff

View file

@ -33,6 +33,8 @@ usage() {
echo -e " ${GREEN}update${NC} Pull latest images + rebuild app"
echo -e " ${GREEN}preflight${NC} Check ports + resources; write .env"
echo -e " ${GREEN}test${NC} Run test suite"
echo -e " ${GREEN}e2e [mode]${NC} Run E2E tests (mode: demo|cloud|local, default: demo)"
echo -e " Set E2E_HEADLESS=false to run headed via Xvfb"
echo -e " ${GREEN}prepare-training${NC} Extract cover letters → training JSONL"
echo -e " ${GREEN}finetune${NC} Run LoRA fine-tune (needs GPU profile)"
echo -e " ${GREEN}clean${NC} Remove containers, images, volumes (DESTRUCTIVE)"
@ -170,6 +172,24 @@ case "$CMD" in
fi
;;
e2e)
MODE="${2:-demo}"
RESULTS_DIR="tests/e2e/results/${MODE}"
mkdir -p "${RESULTS_DIR}"
HEADLESS="${E2E_HEADLESS:-true}"
if [ "$HEADLESS" = "false" ]; then
RUNNER="xvfb-run --auto-servernum --server-args='-screen 0 1280x900x24'"
else
RUNNER=""
fi
info "Running E2E tests (mode=${MODE}, headless=${HEADLESS})..."
$RUNNER conda run -n job-seeker pytest tests/e2e/ \
--mode="${MODE}" \
--json-report \
--json-report-file="${RESULTS_DIR}/report.json" \
-v "${@:3}"
;;
help|--help|-h)
usage
;;

View file

@ -13,6 +13,8 @@ streamlit-paste-button>=0.1.0
# ── Job scraping ──────────────────────────────────────────────────────────
python-jobspy>=1.1
playwright>=1.40
pytest-playwright>=0.4
pytest-json-report>=1.5
selenium
undetected-chromedriver
webdriver-manager

View file

@ -150,6 +150,13 @@ def _run_task(db_path: Path, task_id: int, task_type: str, job_id: int,
try:
if task_type == "discovery":
import os as _os
if _os.environ.get("DEMO_MODE", "").lower() in ("1", "true", "yes"):
update_task_status(
db_path, task_id, "failed",
error="Discovery is disabled in the public demo. Run your own instance to use this feature.",
)
return
from scripts.discover import run_discovery
new_count = run_discovery(db_path)
n = new_count or 0

0
tests/e2e/__init__.py Normal file
View file

180
tests/e2e/conftest.py Normal file
View file

@ -0,0 +1,180 @@
"""
Peregrine E2E test harness shared fixtures and Streamlit helpers.
Run with: pytest tests/e2e/ --mode=demo|cloud|local|all
"""
from __future__ import annotations
import os
import logging
from pathlib import Path
import pytest
from dotenv import load_dotenv
from playwright.sync_api import Page, BrowserContext
from tests.e2e.models import ErrorRecord, ModeConfig, diff_errors
from tests.e2e.modes.demo import DEMO
from tests.e2e.modes.cloud import CLOUD
from tests.e2e.modes.local import LOCAL
load_dotenv(".env.e2e")
log = logging.getLogger(__name__)
_ALL_MODES = {"demo": DEMO, "cloud": CLOUD, "local": LOCAL}
_CONSOLE_NOISE = [
"WebSocket connection",
"WebSocket is closed",
"_stcore/stream",
"favicon.ico",
]
def pytest_addoption(parser):
parser.addoption(
"--mode",
action="store",
default="demo",
choices=["demo", "cloud", "local", "all"],
help="Which Peregrine instance(s) to test against",
)
def pytest_configure(config):
config.addinivalue_line("markers", "e2e: mark test as E2E (requires running Peregrine instance)")
def pytest_collection_modifyitems(config, items):
"""Skip E2E tests if --mode not explicitly passed (belt-and-suspenders isolation)."""
# Only skip if we're collecting from tests/e2e/ without explicit --mode
e2e_items = [i for i in items if "tests/e2e/" in str(i.fspath)]
if e2e_items and not any("--mode" in arg for arg in config.invocation_params.args):
skip = pytest.mark.skip(reason="E2E tests require explicit --mode flag")
for item in e2e_items:
item.add_marker(skip)
@pytest.fixture(scope="session")
def active_modes(pytestconfig) -> list[ModeConfig]:
mode_arg = pytestconfig.getoption("--mode")
if mode_arg == "all":
return list(_ALL_MODES.values())
return [_ALL_MODES[mode_arg]]
@pytest.fixture(scope="session", autouse=True)
def assert_instances_reachable(active_modes):
"""Fail fast with a clear message if any target instance is not running."""
import socket
from urllib.parse import urlparse
for mode in active_modes:
parsed = urlparse(mode.base_url)
host, port = parsed.hostname, parsed.port or 80
try:
with socket.create_connection((host, port), timeout=3):
pass
except OSError:
pytest.exit(
f"[{mode.name}] Instance not reachable at {mode.base_url}"
"start the instance before running E2E tests.",
returncode=1,
)
@pytest.fixture(scope="session")
def mode_contexts(active_modes, playwright) -> dict[str, BrowserContext]:
"""One browser context per active mode, with auth injected via route handler."""
from tests.e2e.modes.cloud import _get_jwt
headless = os.environ.get("E2E_HEADLESS", "true").lower() != "false"
slow_mo = int(os.environ.get("E2E_SLOW_MO", "0"))
browser = playwright.chromium.launch(headless=headless, slow_mo=slow_mo)
contexts = {}
for mode in active_modes:
ctx = browser.new_context(viewport={"width": 1280, "height": 900})
if mode.name == "cloud":
# Cookies are sent on WebSocket upgrade requests; set_extra_http_headers
# and ctx.route() are both HTTP-only and miss st.context.headers.
# cloud_session.py falls back to the Cookie header when X-CF-Session
# is absent (direct access without Caddy).
jwt = _get_jwt()
ctx.add_cookies([{
"name": "cf_session",
"value": jwt,
"domain": "localhost",
"path": "/",
}])
else:
mode.auth_setup(ctx)
contexts[mode.name] = ctx
yield contexts
browser.close()
def wait_for_streamlit(page: Page, timeout: int = 10_000) -> None:
"""
Wait until Streamlit has finished rendering.
Uses 2000ms idle window NOT networkidle (Playwright's networkidle uses
500ms which is too short for Peregrine's 3s sidebar fragment poller).
"""
try:
page.wait_for_selector('[data-testid="stSpinner"]', state="hidden", timeout=timeout)
except Exception:
pass
try:
page.wait_for_function(
"() => !document.querySelector('[data-testid=\"stStatusWidget\"]')"
"?.textContent?.includes('running')",
timeout=5_000,
)
except Exception:
pass
page.wait_for_timeout(2_000)
def get_page_errors(page) -> list[ErrorRecord]:
"""Scan DOM for Streamlit error indicators."""
errors: list[ErrorRecord] = []
for el in page.query_selector_all('[data-testid="stException"]'):
# text_content() includes text from CSS-hidden elements (e.g. collapsed expanders)
msg = (el.text_content() or "").strip()[:500]
errors.append(ErrorRecord(
type="exception",
message=msg,
element_html=el.inner_html()[:1000],
))
for el in page.query_selector_all('[data-testid="stAlert"]'):
# Streamlit 1.35+: st.error() renders child [data-testid="stAlertContentError"]
# kind is a React prop — NOT a DOM attribute. Child detection is authoritative.
if el.query_selector('[data-testid="stAlertContentError"]'):
msg = (el.text_content() or "").strip()[:500]
errors.append(ErrorRecord(
type="alert",
message=msg,
element_html=el.inner_html()[:1000],
))
return errors
def get_console_errors(messages) -> list[str]:
"""Filter browser console messages to real errors, excluding Streamlit noise."""
result = []
for msg in messages:
if msg.type != "error":
continue
text = msg.text
if any(noise in text for noise in _CONSOLE_NOISE):
continue
result.append(text)
return result
def screenshot_on_fail(page: Page, mode_name: str, test_name: str) -> Path:
results_dir = Path(f"tests/e2e/results/{mode_name}/screenshots")
results_dir.mkdir(parents=True, exist_ok=True)
path = results_dir / f"{test_name}.png"
page.screenshot(path=str(path), full_page=True)
return path

41
tests/e2e/models.py Normal file
View file

@ -0,0 +1,41 @@
"""Shared data models for the Peregrine E2E test harness."""
from __future__ import annotations
import fnmatch
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Any
@dataclass(frozen=True)
class ErrorRecord:
type: str # "exception" | "alert"
message: str
element_html: str
def __eq__(self, other: object) -> bool:
if not isinstance(other, ErrorRecord):
return NotImplemented
return (self.type, self.message) == (other.type, other.message)
def __hash__(self) -> int:
return hash((self.type, self.message))
def diff_errors(before: list[ErrorRecord], after: list[ErrorRecord]) -> list[ErrorRecord]:
"""Return errors in `after` that were not present in `before`."""
before_set = set(before)
return [e for e in after if e not in before_set]
@dataclass
class ModeConfig:
name: str
base_url: str
auth_setup: Callable[[Any], None]
expected_failures: list[str] # fnmatch glob patterns against element labels
results_dir: Path | None
settings_tabs: list[str] # tabs expected per mode
def matches_expected_failure(self, label: str) -> bool:
"""Return True if label matches any expected_failure pattern (fnmatch)."""
return any(fnmatch.fnmatch(label, pattern) for pattern in self.expected_failures)

View file

76
tests/e2e/modes/cloud.py Normal file
View 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/peregrine",
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
View 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/peregrine",
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
View 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:8501/peregrine",
auth_setup=lambda ctx: None,
expected_failures=[],
results_dir=Path("tests/e2e/results/local"),
settings_tabs=_BASE_SETTINGS_TABS,
)

View file

View file

@ -0,0 +1,4 @@
from tests.e2e.pages.base_page import BasePage
class ApplyPage(BasePage):
nav_label = "Apply Workspace"

View file

@ -0,0 +1,79 @@
"""Base page object — navigation, error capture, interactable discovery."""
from __future__ import annotations
import logging
import warnings
import fnmatch
from dataclasses import dataclass
from playwright.sync_api import Page
from tests.e2e.models import ErrorRecord, ModeConfig
log = logging.getLogger(__name__)
INTERACTABLE_SELECTORS = [
'[data-testid="baseButton-primary"] button',
'[data-testid="baseButton-secondary"] button',
'[data-testid="stTab"] button[role="tab"]',
'[data-testid="stSelectbox"]',
'[data-testid="stCheckbox"] input',
]
@dataclass
class InteractableElement:
label: str
selector: str
index: int
class BasePage:
"""Base page object for all Peregrine pages."""
nav_label: str = ""
def __init__(self, page: Page, mode: ModeConfig, console_messages: list):
self.page = page
self.mode = mode
self._console_messages = console_messages
def navigate(self) -> None:
"""Navigate to this page by clicking its sidebar nav link."""
from tests.e2e.conftest import wait_for_streamlit
sidebar = self.page.locator('[data-testid="stSidebarNav"]')
sidebar.get_by_text(self.nav_label, exact=False).first.click()
wait_for_streamlit(self.page)
def get_errors(self) -> list[ErrorRecord]:
from tests.e2e.conftest import get_page_errors
return get_page_errors(self.page)
def get_console_errors(self) -> list[str]:
from tests.e2e.conftest import get_console_errors
return get_console_errors(self._console_messages)
def discover_interactables(self, skip_sidebar: bool = True) -> list[InteractableElement]:
"""Find all interactive elements on the current page, excluding sidebar."""
found: list[InteractableElement] = []
for selector in INTERACTABLE_SELECTORS:
elements = self.page.query_selector_all(selector)
for i, el in enumerate(elements):
if skip_sidebar and el.evaluate(
"el => el.closest('[data-testid=\"stSidebar\"]') !== null"
):
continue
label = (el.inner_text() or el.get_attribute("aria-label") or f"element-{i}").strip()
label = label[:80]
found.append(InteractableElement(label=label, selector=selector, index=i))
for pattern in self.mode.expected_failures:
matches = [e for e in found if fnmatch.fnmatch(e.label, pattern)]
if len(matches) > 1:
warnings.warn(
f"expected_failure pattern '{pattern}' matches {len(matches)} elements: "
+ ", ".join(f'"{m.label}"' for m in matches),
stacklevel=2,
)
return found

View file

@ -0,0 +1,4 @@
from tests.e2e.pages.base_page import BasePage
class HomePage(BasePage):
nav_label = "Home"

View file

@ -0,0 +1,4 @@
from tests.e2e.pages.base_page import BasePage
class InterviewPrepPage(BasePage):
nav_label = "Interview Prep"

View file

@ -0,0 +1,4 @@
from tests.e2e.pages.base_page import BasePage
class InterviewsPage(BasePage):
nav_label = "Interviews"

View file

@ -0,0 +1,4 @@
from tests.e2e.pages.base_page import BasePage
class JobReviewPage(BasePage):
nav_label = "Job Review"

View file

@ -0,0 +1,44 @@
"""Settings page — tab-aware page object."""
from __future__ import annotations
import logging
from tests.e2e.pages.base_page import BasePage, InteractableElement
log = logging.getLogger(__name__)
class SettingsPage(BasePage):
nav_label = "Settings"
def discover_interactables(self, skip_sidebar: bool = True) -> list[InteractableElement]:
"""
Settings has multiple tabs. Click each expected tab, collect interactables,
return the full combined list.
"""
from tests.e2e.conftest import wait_for_streamlit
all_elements: list[InteractableElement] = []
tab_labels = self.mode.settings_tabs
for tab_label in tab_labels:
# Match on full label text — handles emoji correctly.
# Do NOT use tab_label.split()[-1]: "My Profile" and "Resume Profile"
# both end in "Profile", causing a silent collision.
tab_btn = self.page.locator(
'[data-testid="stTab"] button[role="tab"]'
).filter(has_text=tab_label)
if tab_btn.count() == 0:
log.warning("Settings tab not found: %s", tab_label)
continue
tab_btn.first.click()
wait_for_streamlit(self.page)
tab_elements = super().discover_interactables(skip_sidebar=skip_sidebar)
# Exclude tab buttons (already handled by clicking)
tab_elements = [
e for e in tab_elements
if 'role="tab"' not in e.selector
]
all_elements.extend(tab_elements)
return all_elements

View file

@ -0,0 +1,4 @@
from tests.e2e.pages.base_page import BasePage
class SurveyPage(BasePage):
nav_label = "Survey Assistant"

View file

View file

@ -0,0 +1,126 @@
"""
Interaction pass discover every interactable element on each page, click it,
diff errors before/after. Demo mode XFAIL patterns are checked; unexpected passes
are flagged as regressions.
Run: pytest tests/e2e/test_interactions.py --mode=demo -v
"""
from __future__ import annotations
import pytest
from tests.e2e.conftest import (
wait_for_streamlit, get_page_errors, screenshot_on_fail,
)
from tests.e2e.models import ModeConfig, diff_errors
from tests.e2e.pages.home_page import HomePage
from tests.e2e.pages.job_review_page import JobReviewPage
from tests.e2e.pages.apply_page import ApplyPage
from tests.e2e.pages.interviews_page import InterviewsPage
from tests.e2e.pages.interview_prep_page import InterviewPrepPage
from tests.e2e.pages.survey_page import SurveyPage
from tests.e2e.pages.settings_page import SettingsPage
PAGE_CLASSES = [
HomePage, JobReviewPage, ApplyPage, InterviewsPage,
InterviewPrepPage, SurveyPage, SettingsPage,
]
@pytest.mark.e2e
def test_interactions_all_pages(active_modes, mode_contexts, playwright):
"""
For each active mode and page: click every discovered interactable,
diff errors, XFAIL expected demo failures, FAIL on unexpected errors.
XPASS (expected failure that didn't fail) is also reported.
"""
failures: list[str] = []
xfails: list[str] = []
xpasses: list[str] = []
for mode in active_modes:
ctx = mode_contexts[mode.name]
page = ctx.new_page()
console_msgs: list = []
page.on("console", lambda msg: console_msgs.append(msg))
page.goto(mode.base_url)
wait_for_streamlit(page)
for PageClass in PAGE_CLASSES:
pg = PageClass(page, mode, console_msgs)
pg.navigate()
elements = pg.discover_interactables()
for element in elements:
pg.navigate()
before = pg.get_errors()
try:
all_matches = page.query_selector_all(element.selector)
content_matches = [
el for el in all_matches
if not el.evaluate(
"el => el.closest('[data-testid=\"stSidebar\"]') !== null"
)
]
if element.index < len(content_matches):
content_matches[element.index].click()
else:
continue
except Exception as e:
failures.append(
f"[{mode.name}] {PageClass.nav_label} / '{element.label}'"
f"could not interact: {e}"
)
continue
wait_for_streamlit(page)
after = pg.get_errors()
new_errors = diff_errors(before, after)
is_expected = mode.matches_expected_failure(element.label)
if new_errors:
if is_expected:
xfails.append(
f"[{mode.name}] {PageClass.nav_label} / '{element.label}' "
f"(expected) — {new_errors[0].message[:120]}"
)
else:
shot = screenshot_on_fail(
page, mode.name,
f"interact_{PageClass.__name__}_{element.label[:30]}"
)
failures.append(
f"[{mode.name}] {PageClass.nav_label} / '{element.label}'"
f"unexpected error: {new_errors[0].message[:200]}\n screenshot: {shot}"
)
else:
if is_expected:
xpasses.append(
f"[{mode.name}] {PageClass.nav_label} / '{element.label}' "
f"— expected to fail but PASSED (neutering guard may be broken!)"
)
page.close()
report_lines = []
if xfails:
report_lines.append(f"XFAIL ({len(xfails)} expected failures, demo mode working correctly):")
report_lines.extend(f" {x}" for x in xfails)
if xpasses:
report_lines.append(f"\nXPASS — REGRESSION ({len(xpasses)} neutering guards broken!):")
report_lines.extend(f" {x}" for x in xpasses)
if failures:
report_lines.append(f"\nFAIL ({len(failures)} unexpected errors):")
report_lines.extend(f" {x}" for x in failures)
if report_lines:
print("\n\n=== E2E Interaction Report ===\n" + "\n".join(report_lines))
if xpasses or failures:
pytest.fail(
f"{len(failures)} unexpected error(s), {len(xpasses)} xpass regression(s). "
"See report above."
)

61
tests/e2e/test_smoke.py Normal file
View file

@ -0,0 +1,61 @@
"""
Smoke pass navigate each page, wait for Streamlit to settle, assert no errors on load.
Errors on page load are always real bugs (not mode-specific).
Run: pytest tests/e2e/test_smoke.py --mode=demo
"""
from __future__ import annotations
import pytest
from tests.e2e.conftest import wait_for_streamlit, get_page_errors, get_console_errors, screenshot_on_fail
from tests.e2e.models import ModeConfig
from tests.e2e.pages.home_page import HomePage
from tests.e2e.pages.job_review_page import JobReviewPage
from tests.e2e.pages.apply_page import ApplyPage
from tests.e2e.pages.interviews_page import InterviewsPage
from tests.e2e.pages.interview_prep_page import InterviewPrepPage
from tests.e2e.pages.survey_page import SurveyPage
from tests.e2e.pages.settings_page import SettingsPage
PAGE_CLASSES = [
HomePage, JobReviewPage, ApplyPage, InterviewsPage,
InterviewPrepPage, SurveyPage, SettingsPage,
]
@pytest.mark.e2e
def test_smoke_all_pages(active_modes, mode_contexts, playwright):
"""For each active mode: navigate to every page and assert no errors on load."""
failures: list[str] = []
for mode in active_modes:
ctx = mode_contexts[mode.name]
page = ctx.new_page()
console_msgs: list = []
page.on("console", lambda msg: console_msgs.append(msg))
page.goto(mode.base_url)
wait_for_streamlit(page)
for PageClass in PAGE_CLASSES:
pg = PageClass(page, mode, console_msgs)
pg.navigate()
console_msgs.clear()
dom_errors = pg.get_errors()
console_errors = pg.get_console_errors()
if dom_errors or console_errors:
shot_path = screenshot_on_fail(page, mode.name, f"smoke_{PageClass.__name__}")
detail = "\n".join(
[f" DOM: {e.message}" for e in dom_errors]
+ [f" Console: {e}" for e in console_errors]
)
failures.append(
f"[{mode.name}] {PageClass.nav_label} — errors on load:\n{detail}\n screenshot: {shot_path}"
)
page.close()
if failures:
pytest.fail("Smoke test failures:\n\n" + "\n\n".join(failures))

181
tests/test_e2e_helpers.py Normal file
View file

@ -0,0 +1,181 @@
"""Unit tests for E2E harness models and helper utilities."""
import fnmatch
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():
a = ErrorRecord(type="exception", message="boom", element_html="<div>boom</div>")
b = ErrorRecord(type="exception", message="boom", element_html="<div>boom</div>")
assert a == b
def test_error_record_inequality():
a = ErrorRecord(type="exception", message="boom", element_html="")
b = ErrorRecord(type="alert", message="boom", element_html="")
assert a != b
def test_diff_errors_returns_new_only():
before = [ErrorRecord("exception", "old error", "")]
after = [
ErrorRecord("exception", "old error", ""),
ErrorRecord("alert", "new error", ""),
]
result = diff_errors(before, after)
assert result == [ErrorRecord("alert", "new error", "")]
def test_diff_errors_empty_when_no_change():
errors = [ErrorRecord("exception", "x", "")]
assert diff_errors(errors, errors) == []
def test_diff_errors_empty_before():
after = [ErrorRecord("alert", "boom", "")]
assert diff_errors([], after) == after
def test_mode_config_expected_failure_match():
config = ModeConfig(
name="demo",
base_url="http://localhost:8504",
auth_setup=lambda ctx: None,
expected_failures=["Fetch*", "Generate Cover Letter"],
results_dir=None,
settings_tabs=["👤 My Profile"],
)
assert config.matches_expected_failure("Fetch New Jobs")
assert config.matches_expected_failure("Generate Cover Letter")
assert not config.matches_expected_failure("View Jobs")
def test_mode_config_no_expected_failures():
config = ModeConfig(
name="local",
base_url="http://localhost:8502",
auth_setup=lambda ctx: None,
expected_failures=[],
results_dir=None,
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()
def test_get_page_errors_finds_exceptions(monkeypatch):
"""get_page_errors returns ErrorRecord for stException elements."""
from tests.e2e.conftest import get_page_errors
mock_el = MagicMock()
mock_el.get_attribute.return_value = None
mock_el.inner_text.return_value = "RuntimeError: boom"
mock_el.inner_html.return_value = "<div>RuntimeError: boom</div>"
mock_page = MagicMock()
mock_page.query_selector_all.side_effect = lambda sel: (
[mock_el] if "stException" in sel else []
)
errors = get_page_errors(mock_page)
assert len(errors) == 1
assert errors[0].type == "exception"
assert "boom" in errors[0].message
def test_get_page_errors_finds_alert_errors(monkeypatch):
"""get_page_errors returns ErrorRecord for stAlert with stAlertContentError child."""
from tests.e2e.conftest import get_page_errors
mock_child = MagicMock()
mock_el = MagicMock()
mock_el.query_selector.return_value = mock_child
mock_el.inner_text.return_value = "Something went wrong"
mock_el.inner_html.return_value = "<div>Something went wrong</div>"
mock_page = MagicMock()
mock_page.query_selector_all.side_effect = lambda sel: (
[] if "stException" in sel else [mock_el]
)
errors = get_page_errors(mock_page)
assert len(errors) == 1
assert errors[0].type == "alert"
def test_get_page_errors_ignores_non_error_alerts(monkeypatch):
"""get_page_errors does NOT flag st.warning() or st.info() alerts."""
from tests.e2e.conftest import get_page_errors
mock_el = MagicMock()
mock_el.query_selector.return_value = None
mock_el.inner_text.return_value = "Just a warning"
mock_page = MagicMock()
mock_page.query_selector_all.side_effect = lambda sel: (
[] if "stException" in sel else [mock_el]
)
errors = get_page_errors(mock_page)
assert errors == []
def test_get_console_errors_filters_noise():
"""get_console_errors filters benign Streamlit WebSocket reconnect messages."""
from tests.e2e.conftest import get_console_errors
messages = [
MagicMock(type="error", text="WebSocket connection closed"),
MagicMock(type="error", text="TypeError: cannot read property"),
MagicMock(type="log", text="irrelevant"),
]
errors = get_console_errors(messages)
assert errors == ["TypeError: cannot read property"]