"""Shared fixtures + builders for the security-awareness suite.""" from __future__ import annotations import uuid from typing import Any import pytest import pytest_asyncio from sqlalchemy import select from app.auth.models import AuthRole, AuthUserRole from app.core.security import create_access_token from app.identity.person_models import IdentityPerson ADMIN_BASE = "/api/v1/admin/security-awareness" SELF_BASE = "/api/v1/security-awareness" async def make_person(db, email: str, *, role: str | None = None, status: str = "active") -> IdentityPerson: person = IdentityPerson( id=uuid.uuid4(), work_email=email, display_name=email.split("@")[0], employment_status=status, source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(person) await db.flush() if role: role_id = (await db.execute( select(AuthRole.id).where(AuthRole.name == role), )).scalar_one_or_none() if role_id is None: role_id = uuid.uuid4() db.add(AuthRole(id=role_id, name=role, is_system=True)) await db.flush() db.add(AuthUserRole(user_id=person.id, role_id=role_id)) await db.flush() return person def headers_for(person: IdentityPerson, roles: list[str]) -> dict[str, str]: return { "Authorization": ( f"Bearer {create_access_token(str(person.id), roles)}" ), } @pytest_asyncio.fixture async def security_officer(db_session): """安全员 — the role that actually runs awareness campaigns.""" person = await make_person( db_session, "officer@test.com", role="security_officer", ) await db_session.commit() return person, headers_for(person, ["security_officer"]) @pytest_asyncio.fixture async def reader(db_session): """A plain employee — the audience.""" person = await make_person(db_session, "reader@test.com", role="employee") await db_session.commit() return person, headers_for(person, ["employee"]) def awareness_key(person_id: uuid.UUID | str, name: str = "p1.png") -> str: """A well-formed ``awareness_material`` object key. Shape must match ``storage_router._build_relative_key`` exactly — ``{purpose}/{yyyy}/{mm}/{dd}/{person_id}/{filename}`` — because the service re-parses it with the router's own parser before persisting. Building it by hand here is deliberate: a test that mints keys through the upload endpoint would need S3 configured, and the thing under test is the *parser gate*, not S3. """ return f"awareness_material/2026/09/03/{person_id}/{name}" def comic_payload(person_id: uuid.UUID, *, publish: bool = True) -> dict: return { "kind": "comic", "category": "phishing", "tags": ["钓鱼", "phishing"], "publish": publish, "localized": { "zh": { "title": "识破钓鱼邮件", "summary": "五个细节", "image_keys": [ awareness_key(person_id, "zh1.png"), awareness_key(person_id, "zh2.png"), ], }, "en": { "title": "Spotting a phishing mail", "image_keys": [awareness_key(person_id, "en1.png")], }, }, } def article_payload(*, publish: bool = True, body: str = "## 正文\n内容") -> dict: return { "kind": "article", "category": "account", "publish": publish, "localized": {"zh": {"title": "口令怎么设", "body": body}}, } @pytest.fixture def secaw_bases() -> tuple[str, str]: return SELF_BASE, ADMIN_BASE @pytest.fixture(autouse=True) def spawned_dispatches(monkeypatch): """Neutralise the background hand-off in ``/send-now``, and record it. Left live, the spawned task races every assertion in the suite: it claims the campaign on ITS OWN session the moment the test next awaits, so the dispatch :func:`send_now` runs afterwards gets ``skipped: not_scheduled`` and the target ledger reads empty — or half-written, depending on how many times the test yielded. That is not a hypothetical; it is what this suite did the first time the endpoint became a 202 (27 failures). Tests own the clock instead. Returns the list of task names handed to ``spawn`` so a test can assert the endpoint really did delegate. """ spawned: list[str] = [] def _record(coro: Any, *, name: str | None = None) -> None: # Close it rather than leaving an un-awaited coroutine behind — # otherwise every send in the suite emits a RuntimeWarning and # the real ones get lost in the noise. coro.close() spawned.append(name or "") monkeypatch.setattr( "app.domains.security_awareness.campaign_router.spawn", _record, ) return spawned # ════════════════════════════════════════════════════════════════════ # Dispatch helpers # ════════════════════════════════════════════════════════════════════ CAMPAIGNS = f"{ADMIN_BASE}/campaigns" #: What the real primitive returns, for a double that needs a shape. FANOUT_KEYS = ("enqueued", "delivered", "failed") def fanout_double( *, failed_for: set[str] | None = None, ) -> Any: """A stand-in for ``fanout.enqueue_then_drain`` that HONOURS its contract instead of just returning a plausible dict. An ``AsyncMock(return_value={...})`` is the obvious double and it is the wrong one: the primitive's contract includes *calling ``on_enqueued`` once per recipient, inside that recipient's transaction*, and the campaign dispatcher stamps ``queued_at`` / ``last_sent_at`` there. A double that skips the hook leaves every ledger row unstamped, so a test asserting "the funnel says 2 queued" fails for a reason that has nothing to do with the code under test — and worse, a REGRESSION that dropped the hook would go green. ``failed_for`` — work emails whose enqueue "fails": the hook is not called for them and they are counted in ``failed``, which is exactly how a real enqueue failure looks from the outside. """ bad = failed_for or set() async def _fake(db, envelopes, **kw): # noqa: ANN001, ANN003 on_enqueued = kw.get("on_enqueued") enqueued = failed = 0 for env in envelopes: if (env.recipient.work_email or "") in bad: failed += 1 continue if on_enqueued is not None: await on_enqueued(env, uuid.uuid4()) await db.commit() enqueued += 1 return {"enqueued": enqueued, "delivered": enqueued, "failed": failed} return _fake async def resend( client, db, campaign_id: str, headers: dict[str, str], *, actor_id: uuid.UUID | None = None, expect: int = 202, ) -> Any: """``/resend`` over HTTP, then run the chase inline — same two halves as :func:`send_now`, and off-request for the same reason: on day one "everybody who has not acknowledged" IS most of the audience.""" from app.domains.security_awareness.campaign_service import CampaignService r = await client.post( f"{CAMPAIGNS}/{campaign_id}/resend", headers=headers, ) assert r.status_code == expect, r.text if r.status_code == 202: await CampaignService(db).resend_pending( uuid.UUID(str(campaign_id)), actor_id=actor_id or uuid.uuid4(), trace_id="test", ) return r async def send_now( client, db, campaign_id: str, headers: dict[str, str], *, force: bool = False, actor_id: uuid.UUID | None = None, expect: int = 202, ) -> Any: """Arm a campaign over HTTP, then run the fan-out inline. ``/send-now`` is a **202**: it arms the campaign and hands the fan-out to ``background.spawn``, because a company-wide audience takes minutes to drain and outlived nginx's 300s ``proxy_read_timeout`` back when it was awaited in-request (2026-09-07 — 416 recipients, 15.5 minutes, and an admin staring at a failure toast over a send that was still running). Tests drive the two halves explicitly rather than racing the spawned asyncio task: an assertion about the target ledger must not be a coin flip on scheduler timing, and running the dispatch on the test's own session is what makes the fan-out double above observable. """ from app.domains.security_awareness.campaign_service import CampaignService url = f"{CAMPAIGNS}/{campaign_id}/send-now" r = await client.post( url + ("?force=true" if force else ""), headers=headers, ) assert r.status_code == expect, r.text if r.status_code == 202: await CampaignService(db).dispatch( uuid.UUID(str(campaign_id)), trace_id="test", actor_id=actor_id, ) return r