"""Shared fixtures and test helpers for the Scout suite. Runs against a REAL Postgres database (SCOUT_DB_URL, a dedicated `scout_test` database provided by the orchestrator) — nothing here mocks the ORM. The guard below refuses to run at all unless that URL unambiguously points at the test database, so a misconfigured environment cannot point this suite at production. """ import base64 import json import os import tempfile import zipfile from io import BytesIO from types import SimpleNamespace import pytest # --- guard: refuse anything but the dedicated test database ----------------- # This MUST run before any `app.*` module is imported, because importing # app.db creates the SQLAlchemy engine bound to SCOUT_DB_URL at import time. _db_url = os.environ.get("SCOUT_DB_URL", "") if not _db_url or not _db_url.endswith("/scout_test"): pytest.exit("refusing: SCOUT_DB_URL must point at scout_test") os.environ.setdefault("SCOUT_DATA_DIR", tempfile.mkdtemp(prefix="scout-test-data")) # BW Auth must LOOK configured so the invite/reset/report paths run — but no # call may ever leave the process: the autouse _no_bw_network fixture below # stubs the single outbound funnel (bw_auth._post_json). if not os.environ.get("BW_CLIENT_ID"): os.environ["BW_CLIENT_ID"] = "t-client" if not os.environ.get("BW_CLIENT_SECRET"): os.environ["BW_CLIENT_SECRET"] = "t-secret" # --- app imports (safe now that the guard above has run) -------------------- from fastapi.testclient import TestClient # noqa: E402 from itsdangerous import TimestampSigner # noqa: E402 from sqlalchemy import text # noqa: E402 import app.main as main # noqa: E402 import app.models # noqa: E402,F401 registers every table on Base.metadata from app.config import get_settings # noqa: E402 from app.constants import LEVEL_ADMIN, LEVEL_LEAD, LEVEL_REVIEWER, SEED_LEVELS # noqa: E402 from app.db import SessionLocal, get_engine # noqa: E402 from app.models.account import Account # noqa: E402 from app.models.base import Base # noqa: E402 from app.models.level import Level # noqa: E402 from app.services import projects as projects_service # noqa: E402 # --- schema lifecycle --------------------------------------------------- @pytest.fixture(scope="session", autouse=True) def _schema(): """Rebuild the schema once for the whole run, against scout_test only. The access levels are seeded here to match what migration 0003 writes into a real database. `test_levels.py` asserts the two still agree, so this cannot drift silently. """ engine = get_engine() Base.metadata.drop_all(engine) Base.metadata.create_all(engine) session = SessionLocal() try: for name, spec in SEED_LEVELS.items(): session.merge( Level(name=name, permissions=spec["permissions"], assignable=spec["assignable"]) ) session.commit() finally: session.close() yield @pytest.fixture(autouse=True) def _isolate_t_prefixed_data(): """Wipe everything a test created, keyed off the t_ username prefix. Deleting the project(s) first cascades (real ON DELETE CASCADE constraints, not ORM-side) through project_members, options, reviews, aspect_votes, final_picks, and materials. The account rows themselves are deleted after, once nothing still references them as `created_by`. """ yield session = SessionLocal() try: session.execute(text("DELETE FROM projects WHERE created_by LIKE 't\\_%'")) session.execute(text("DELETE FROM app_accounts WHERE username LIKE 't\\_%'")) session.commit() finally: session.close() # --- outbound BW calls: stubbed and recorded -------------------------------- # Every server-to-server BW call goes through bw_auth._post_json. This fixture # replaces it with a recorder so (a) no test can touch the network and (b) a # test can assert WHAT would have been sent. The return shapes mirror the # gateway's real responses closely enough for the code under test. @pytest.fixture(autouse=True) def bw_calls(monkeypatch): calls: list[tuple[str, dict]] = [] def fake_post_json(path, payload): calls.append((path, dict(payload))) if path == "/app/userinfo": username = payload.get("username", "") return {"username": username, "email": f"{username}@example.com", "first": "T", "last": "User"} if path == "/app/invite-user": return {"success": True, "created": True, "emailed_to": payload.get("email", ""), "user": {"username": payload.get("username", "")}} if path == "/app/user-access": return {"success": True, "access": []} return {"success": True} import app.bw_auth as bw_auth_module monkeypatch.setattr(bw_auth_module, "_post_json", fake_post_json) return calls # --- client / session ---------------------------------------------------- @pytest.fixture def client(): with TestClient(main.app, raise_server_exceptions=False) as test_client: yield test_client def session_cookie(username: str) -> str: """Mint a cookie value byte-for-byte compatible with Starlette's SessionMiddleware, so tests can sign in without a login route.""" signer = TimestampSigner(str(get_settings().scout_session_secret)) payload = base64.b64encode(json.dumps({"user": username}).encode("utf-8")) return signer.sign(payload).decode("utf-8") def as_user(client, username: str) -> None: """Sign `client` in as `username` for subsequent requests.""" client.cookies.set("scout_session", session_cookie(username)) # --- data fixtures --------------------------------------------------------- @pytest.fixture def db(): session = SessionLocal() try: yield session finally: session.close() @pytest.fixture def seed(db): """The standard cast: * `t_admin` — app-wide `admin` with all_instances, so they reach every project without a grant (staff). * `t_member` — `reviewer` on the project (a client). * `t_lead` — `lead` on the same project: reviews AND sees results. * `t_outsider`— a real account on no project at all. """ admin = Account(username="t_admin", email="t_admin@example.com", level=LEVEL_ADMIN, all_instances=True, active=True) member = Account(username="t_member", email="t_member@example.com", level=LEVEL_REVIEWER, all_instances=False, active=True) lead = Account(username="t_lead", email="t_lead@example.com", level=LEVEL_REVIEWER, all_instances=False, active=True) outsider = Account(username="t_outsider", email="t_outsider@example.com", level=LEVEL_REVIEWER, all_instances=False, active=True) db.add_all([admin, member, lead, outsider]) db.commit() project = projects_service.create( db, name="Test Project", client_name="Test Client", brief="A brief.", status="active", created_by="t_admin", ) projects_service.add_member(db, project.id, "t_member", "t_admin", level=LEVEL_REVIEWER) projects_service.add_member(db, project.id, "t_lead", "t_admin", level=LEVEL_LEAD) db.commit() return SimpleNamespace( project=project, admin=admin, member=member, lead=lead, outsider=outsider ) # --- research-package builder ---------------------------------------------- # A minimal, valid 1x1 PNG — small enough to keep the fixture files tiny. PNG_1PX = base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" ) def make_package( project_slug: str, options: list[dict], *, schema_version: int = 1, extra_entries: dict[str, bytes] | None = None, omit_screenshot: bool = False, ) -> bytes: """Build an in-memory research-package zip per the import contract. `options` is a list of dicts, each at minimum needing "slug"; any of display_label/descriptor/source_name/source_url/why_selected/design_notes may also be supplied and are passed through verbatim. `omit_screenshot=True` skips writing the mobile capture (both the file and its manifest.json entry) for every option — the SCREENSHOT_MISSING case. `extra_entries` are written into the zip raw, untouched by the contract logic — used to build zip-slip payloads. """ buf = BytesIO() with zipfile.ZipFile(buf, "w") as zf: manifest_options = [] for opt in options: slug = opt["slug"] desktop_path = f"screenshots/{slug}-desktop.png" mobile_path = f"screenshots/{slug}-mobile.png" zf.writestr(desktop_path, PNG_1PX) screenshots = {"desktop": desktop_path} if not omit_screenshot: zf.writestr(mobile_path, PNG_1PX) screenshots["mobile"] = mobile_path manifest_options.append( { "slug": slug, "display_label": opt.get("display_label", ""), "descriptor": opt.get("descriptor", ""), "source_name": opt.get("source_name", ""), "source_url": opt.get("source_url", ""), "why_selected": opt.get("why_selected", ""), "design_notes": opt.get("design_notes", ""), "screenshots": screenshots, } ) manifest = { "schema_version": schema_version, "project_slug": project_slug, "options": manifest_options, } zf.writestr("manifest.json", json.dumps(manifest)) if extra_entries: for name, content in extra_entries.items(): zf.writestr(name, content) return buf.getvalue() def import_package(client, project, options: list[dict], **kwargs): """POST a package built with make_package(); returns the raw response. Caller is responsible for being signed in as an admin first.""" payload = make_package(project.slug, options, **kwargs) files = {"package": ("package.zip", payload, "application/zip")} return client.post(f"/api/projects/{project.id}/import", files=files) def publish_option(client, project, option_id: int): return client.patch( f"/api/projects/{project.id}/options/{option_id}", json={"status": "published"}, ) def admin_import_and_publish(client, project, options: list[dict]) -> dict[str, int]: """Import then publish every option in `options`. Caller must already be signed in as an admin. Returns {slug: option_id}.""" resp = import_package(client, project, options) assert resp.status_code == 200, resp.text listing = client.get(f"/api/projects/{project.id}/options").json() wanted_slugs = {o["slug"] for o in options} ids: dict[str, int] = {} for row in listing: if row["slug"] in wanted_slugs: r = publish_option(client, project, row["id"]) assert r.status_code == 200, r.text ids[row["slug"]] = row["id"] return ids