"""Regressions from the post-theme review of 安全宣贯.

Every test in here pins a hole that the five feature PRs left open and
that nothing else in the suite would have caught. They are grouped by
the question each one answers.

* **Can somebody claim understanding without demonstrating it?** The web
  and mobile pages hide the confirm button when a material carries a
  quiz, but a hidden button is not a rule — the acknowledge endpoint and
  the ``acknowledge_security_awareness`` MCP tool were both still open.
* **Can a browser write the ledger's strong claim?** ``detail`` is the
  surface that drives read/unread, the participation report and the
  reading badges. It has to be the server's word, not the client's.
* **Can a reported number exceed its own denominator?** The open rate
  divided every opener by the bubble-reached, which is not a rate.
* **Does a refusal come from this layer or from the driver?** A material
  carrying a campaign hit a RESTRICT foreign key and surfaced as a 500.
* **Is a manager digest about one thing?** It was keyed per manager and
  titled with whichever campaign happened to be first.
* **Do the quiz routes belong to the library?** They are addressable on
  their own, so they need their own publication check.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy import select

from app.domains.security_awareness.models import (
    SecawCampaign,
    SecawImpression,
    SecawMaterial,
)
from app.domains.security_quiz.models import SecQuizQuestion
from tests.test_security_awareness.conftest import (
    ADMIN_BASE,
    SELF_BASE,
    article_payload,
    fanout_double,
    headers_for,
    make_person,
    send_now,
)

pytestmark = pytest.mark.asyncio

CAMPAIGNS = f"{ADMIN_BASE}/campaigns"
MY = f"{SELF_BASE}/my/campaigns"

_FANOUT = "app.notifications.fanout.enqueue_then_drain"
_FANOUT_OK = {"enqueued": 5, "delivered": 5, "failed": 0}


async def _publish(client, headers) -> str:
    r = await client.post(
        f"{ADMIN_BASE}/materials", json=article_payload(), headers=headers,
    )
    assert r.status_code == 201, r.text
    return r.json()["data"]["id"]


async def _question(db, correct: str = "A") -> SecQuizQuestion:
    q = SecQuizQuestion(
        id=uuid.uuid4(),
        prompt="点开陌生附件之前先做什么",
        options=[{"key": "A", "text": "对"}, {"key": "B", "text": "错"}],
        qtype="single",
        correct_key=correct,
        explanation="因为 A",
        category="phishing",
    )
    db.add(q)
    await db.flush()
    await db.commit()
    return q


async def _send(client, db, headers, material, **over) -> str:
    body = {
        "material_id": material,
        "name": "必修宣贯",
        "audience_config": {},
        "channels": ["feishu"],
        "mandatory": True,
        "completion_mode": "ack",
        **over,
    }
    campaign = (await client.post(
        CAMPAIGNS, json=body, headers=headers,
    )).json()["data"]["id"]
    with patch(_FANOUT, new=fanout_double()):
        await send_now(
            client, db, campaign, headers, force=over.pop("force", True),
        )
    return campaign


# ════════════════════════════════════════════════════════════════════
# The quiz gate is a rule, not a hidden button
# ════════════════════════════════════════════════════════════════════


async def test_a_quiz_material_cannot_be_acknowledged_by_pressing_confirm(
    client, db_session, security_officer, reader,
):
    """The hole the whole quiz feature exists to close.

    Both pages hide the confirm control when ``has_quiz`` is true, but
    the endpoint stayed reachable directly — and so did the MCP tool that
    calls the same service method. Either one would let a person (or an
    agent acting for them) assert understanding without demonstrating it,
    which is precisely the claim the quiz was added to make unassertable.
    """
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    campaign = await _send(client, db_session, oh, material)

    blocked = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert blocked.status_code == 409, blocked.text
    assert blocked.json()["errors"][0]["code"] == "SECAW_QUIZ_REQUIRED"
    # Still outstanding: a refused confirm must not half-settle anything.
    assert len((await client.get(MY, headers=rh)).json()["data"]) == 1

    # Passing is the way in, and afterwards the plain endpoint is
    # idempotent again like any other acknowledgement.
    passed = await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )
    assert passed.json()["data"]["passed"] is True
    again = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert again.status_code == 200
    assert again.json()["data"]["already"] is True


async def test_attaching_a_quiz_later_does_not_unconfirm_past_readers(
    client, db_session, security_officer, reader,
):
    """The gate sits AFTER the idempotent return.

    Somebody who confirmed while the material had no quiz did what was
    asked at the time; a question added next month must not turn their
    record back into outstanding work, and must not make a repeat press
    of their own button start failing.
    """
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    campaign = await _send(client, db_session, oh, material)

    first = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert first.status_code == 200
    stamp = first.json()["data"]["acknowledged_at"]

    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    after = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert after.status_code == 200, after.text
    assert after.json()["data"]["acknowledged_at"] == stamp
    assert (await client.get(MY, headers=rh)).json()["data"] == []


async def test_one_pass_settles_every_campaign_asking_for_that_material(
    client, db_session, security_officer, reader,
):
    """A second campaign for the same material must not strand anybody.

    The quiz component refuses to re-offer a quiz somebody has already
    passed — re-offering would imply the record was not kept — so if a
    pass settled only the most urgent campaign, the other would sit in
    the todo list with no control left anywhere to clear it. One passing
    attempt is one demonstration of understanding of one piece of
    content; it settles every campaign asking for exactly that.
    """
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    first = await _send(client, db_session, oh, material, name="第一波")
    second = await _send(client, db_session, oh, material, name="第二波")
    assert len((await client.get(MY, headers=rh)).json()["data"]) == 2

    result = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )).json()["data"]
    assert set(result["acknowledged_campaign_ids"]) == {first, second}
    # Singular still names one of them, for the banner that was on screen.
    assert result["acknowledged_campaign_id"] in {first, second}
    assert (await client.get(MY, headers=rh)).json()["data"] == []


async def test_a_campaign_sent_after_the_pass_can_still_be_confirmed(
    client, db_session, security_officer, reader,
):
    """The other half of the same problem, from the other direction.

    Here the pass happened first and the campaign arrived afterwards, so
    there was nothing for it to settle at the time. The page renders the
    plain confirm control in that state (``has_quiz && quiz_passed``),
    and the gate lets it through because the demonstration is on file.
    """
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )
    campaign = await _send(client, db_session, oh, material)

    detail = (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).json()["data"]
    # What the page renders its control from.
    assert detail["has_quiz"] is True
    assert detail["quiz_passed"] is True
    assert detail["pending_ack"]["campaign_id"] == campaign

    r = await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)
    assert r.status_code == 200, r.text
    assert (await client.get(MY, headers=rh)).json()["data"] == []


# ════════════════════════════════════════════════════════════════════
# The ledger's strong claim is the server's word
# ════════════════════════════════════════════════════════════════════


async def test_a_browser_cannot_report_itself_as_having_read_a_material(
    client, db_session, security_officer, reader,
):
    """``detail`` is not decoration.

    It drives the library's read/unread marker, the ``opened`` column of
    the participation report and the ``awareness_materials_read`` badge.
    Accepting a self-report would let anybody mark fifteen materials read
    without opening one — and those numbers only mean anything because
    "read" is a stronger claim than "shown".
    """
    _officer, oh = security_officer
    person, rh = reader
    material = await _publish(client, oh)

    refused = await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": material, "surface": "detail"}, headers=rh,
    )
    assert refused.status_code == 422, refused.text
    # And nothing was written behind the refusal.
    assert (await db_session.execute(
        select(SecawImpression).where(
            SecawImpression.person_id == person.id,
            SecawImpression.surface == "detail",
        )
    )).scalars().first() is None

    # The ambient surface — the one the server genuinely cannot observe —
    # is still reportable.
    assert (await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": material, "surface": "bubble"}, headers=rh,
    )).status_code == 202


async def test_campaign_reach_cannot_be_self_reported_either(
    client, security_officer, reader,
):
    """Whether a push landed is the outbox's ``delivered``/``partial``
    status — the thing that owns the retry ladder and the leaver gate.
    A browser asserting it into the exposure ledger would be exactly the
    mirror the campaign module refuses to keep."""
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    r = await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": material, "surface": "campaign_card"},
        headers=rh,
    )
    assert r.status_code == 422, r.text


# ════════════════════════════════════════════════════════════════════
# A rate cannot exceed its own denominator
# ════════════════════════════════════════════════════════════════════


async def test_open_rate_counts_only_the_people_the_bubble_reached(
    client, db_session, security_officer, reader,
):
    """The numerator used to be every opener.

    Somebody can open a material straight from the library without the
    bubble ever offering it, so dividing all openers by the bubble-reached
    produced rates over 100% — a progress bar running off the end, and a
    "worth its slot?" number that answered a different question than its
    own label.
    """
    _officer, oh = security_officer
    bubble_reader, brh = reader
    material = await _publish(client, oh)

    # One person meets it in the bubble and never opens it.
    assert (await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": material, "surface": "bubble"}, headers=brh,
    )).status_code == 202

    # Two OTHER people open it straight from the library. Under the old
    # arithmetic that is 3 openers over 1 bubble-reached = 300%.
    for i in range(2):
        other = await make_person(db_session, f"direct{i}@test.com")
        await db_session.commit()
        await client.get(
            f"{SELF_BASE}/materials/{material}",
            headers=headers_for(other, ["employee"]),
        )

    row = next(
        r for r in (await client.get(
            f"{ADMIN_BASE}/participation", headers=oh,
        )).json()["data"]["materials"]
        if r["material_id"] == material
    )
    assert row["bubble_reach"] == 1
    assert row["opened"] == 2
    assert row["opened_after_bubble"] == 0
    assert row["open_rate_pct"] == 0

    # Now the person the bubble reached opens it: 1 of 1.
    await client.get(f"{SELF_BASE}/materials/{material}", headers=brh)
    row = next(
        r for r in (await client.get(
            f"{ADMIN_BASE}/participation", headers=oh,
        )).json()["data"]["materials"]
        if r["material_id"] == material
    )
    assert row["bubble_reach"] == 1
    assert row["opened"] == 3
    assert row["opened_after_bubble"] == 1
    assert row["open_rate_pct"] == 100


# ════════════════════════════════════════════════════════════════════
# Refusals come from this layer, not from the driver
# ════════════════════════════════════════════════════════════════════


async def test_a_material_carrying_a_campaign_is_refused_not_500(
    client, security_officer,
):
    """``secaw_campaigns.material_id`` is RESTRICT on purpose — a sent
    campaign points at the material it sent. A campaign nobody has opened
    yet leaves zero impressions, so an impressions-only guard let the
    delete through and MySQL answered with a 1451, which reaches the
    admin as a 500 and poisons the transaction."""
    _officer, oh = security_officer
    material = await _publish(client, oh)
    # Draft campaign, never sent: no targets, no impressions, nothing the
    # old guard could see.
    await client.post(
        CAMPAIGNS,
        json={
            "material_id": material, "name": "草稿", "audience_config": {},
            "channels": [], "mandatory": False, "completion_mode": "ack",
        },
        headers=oh,
    )
    r = await client.delete(f"{ADMIN_BASE}/materials/{material}", headers=oh)
    assert r.status_code == 409, r.text
    assert r.json()["errors"][0]["code"] == "SECAW_MATERIAL_HAS_CAMPAIGNS"


async def test_an_answered_material_is_refused_not_500(
    client, db_session, security_officer, reader,
):
    """Same shape on the other RESTRICT: a graded attempt is evidence."""
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "B"}}, headers=rh,
    )
    # Wipe the impressions the reader's own page view would have left, so
    # the ONLY thing standing in the way is the attempt.
    await db_session.execute(
        SecawImpression.__table__.delete().where(
            SecawImpression.material_id == uuid.UUID(material),
        )
    )
    await db_session.commit()

    r = await client.delete(f"{ADMIN_BASE}/materials/{material}", headers=oh)
    assert r.status_code == 409, r.text
    assert r.json()["errors"][0]["code"] == "SECAW_MATERIAL_HAS_QUIZ_ATTEMPTS"


# ════════════════════════════════════════════════════════════════════
# The quiz routes belong to the library
# ════════════════════════════════════════════════════════════════════


@pytest.mark.parametrize("status", ["draft", "archived"])
async def test_the_quiz_routes_apply_the_librarys_own_publication_check(
    client, db_session, security_officer, reader, status,
):
    """Addressable on their own, so they cannot inherit the check from
    whichever page usually links to them. A draft's questions are
    unreleased content, and a bookmark to an archived material must not
    still be able to earn points or clear an acknowledgement."""
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    await client.post(
        f"{ADMIN_BASE}/materials/{material}/status",
        json={"status": status}, headers=oh,
    )

    assert (await client.get(
        f"{SELF_BASE}/materials/{material}/quiz", headers=rh,
    )).status_code == 404
    assert (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )).status_code == 404

    # The admin console still sees it — that is the whole point of a
    # draft.
    assert (await client.get(
        f"{ADMIN_BASE}/materials/{material}/quiz", headers=oh,
    )).status_code == 200


# ════════════════════════════════════════════════════════════════════
# A manager digest is about one campaign
# ════════════════════════════════════════════════════════════════════


async def test_a_manager_with_two_campaigns_outstanding_gets_two_digests(
    client, db_session, security_officer,
):
    """Grouping per manager alone forced the digest to pick a campaign to
    title itself with, then listed people who owed a DIFFERENT piece of
    reading underneath it — and both campaigns shared one cadence
    counter, so the second could never chase on its own schedule."""
    from app.domains.security_awareness.reminders import (
        TEMPLATE_MANAGER,
        ReminderService,
    )

    _officer, oh = security_officer
    manager = await make_person(db_session, "boss@test.com")
    await db_session.commit()
    reports = []
    for i in range(2):
        person = await make_person(db_session, f"report{i}@test.com")
        person.manager_person_id = manager.id
        reports.append(person)
    await db_session.commit()

    first = await _send(client, db_session, oh, await _publish(client, oh), name="宣贯一")
    second = await _send(client, db_session, oh, await _publish(client, oh), name="宣贯二")

    # Age both past the manager-escalation rung.
    old = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=30)
    for cid in (first, second):
        campaign = await db_session.get(SecawCampaign, uuid.UUID(cid))
        campaign.sent_at = old
    await db_session.commit()

    captured: list = []

    async def _capture(_db, envelopes, **_kw):
        captured.extend(envelopes)
        return {"enqueued": len(envelopes), "delivered": len(envelopes),
                "failed": 0}

    with patch(_FANOUT, new=_capture):
        summary = await ReminderService(db_session).run()

    assert summary["manager_pushes"] >= 2
    manager_cards = [
        e for e in captured
        if getattr(e, "template_key", "") == TEMPLATE_MANAGER
        and e.recipient.id == manager.id
    ]
    # One per campaign, each naming its own — not one card standing for
    # both.
    assert len(manager_cards) == 2
    assert {
        str(e.related_resource_id) for e in manager_cards
    } == {first, second}
    assert {
        e.context["campaign_name"] for e in manager_cards
    } == {"宣贯一", "宣贯二"}


# ════════════════════════════════════════════════════════════════════
# The question-bank picker has no silent horizon
# ════════════════════════════════════════════════════════════════════


async def test_the_question_bank_search_is_not_capped_before_it_filters(
    client, db_session, security_officer,
):
    """Fetching the newest N and searching those in Python gives a search
    box a horizon: past N questions, a word that only occurs in an older
    one returns nothing, which reads as "we do not have that question"
    rather than as "we did not look"."""
    _officer, oh = security_officer
    old = SecQuizQuestion(
        id=uuid.uuid4(),
        prompt="U 盘捡到了要怎么处理",
        options=[{"key": "A", "text": "交给 IT"}, {"key": "B", "text": "插上"}],
        qtype="single",
        correct_key="A",
        category="endpoint",
    )
    db_session.add(old)
    await db_session.flush()
    # Everything added after it is newer, so a "newest 500 then filter"
    # implementation would still find it here — the assertion that bites
    # is the one below about the SQL doing the filtering.
    for i in range(5):
        db_session.add(SecQuizQuestion(
            id=uuid.uuid4(),
            prompt=f"无关问题 {i}",
            options=[{"key": "A", "text": "x"}, {"key": "B", "text": "y"}],
            qtype="single",
            correct_key="A",
            category="policy",
        ))
    await db_session.commit()

    # ``limit`` is the size of the ANSWER, applied after the filter — not
    # the size of the window the filter got to look at.
    r = await client.get(
        f"{ADMIN_BASE}/question-bank?q=U 盘&limit=1", headers=oh,
    )
    assert r.status_code == 200, r.text
    rows = r.json()["data"]
    assert len(rows) == 1
    assert rows[0]["question_id"] == str(old.id)

    # AND over tokens: more words narrow.
    assert (await client.get(
        f"{ADMIN_BASE}/question-bank?q=U 盘 不存在的词", headers=oh,
    )).json()["data"] == []


async def test_an_archived_material_stops_recording_new_exposures(
    client, db_session, security_officer, reader,
):
    """Belt and braces on the ledger's own gate, from the outside: the
    bubble report path and the read path both go through the same
    published-only check, so archiving really does stop the counters."""
    _officer, oh = security_officer
    person, rh = reader
    material = await _publish(client, oh)
    await client.post(
        f"{ADMIN_BASE}/materials/{material}/status",
        json={"status": "archived"}, headers=oh,
    )
    r = await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": material, "surface": "bubble"}, headers=rh,
    )
    assert r.status_code == 202
    assert r.json()["data"]["recorded"] is False
    assert (await db_session.execute(
        select(SecawImpression).where(
            SecawImpression.material_id == uuid.UUID(material),
        )
    )).scalars().first() is None
    # And the material itself is gone from the library.
    assert (await db_session.get(
        SecawMaterial, uuid.UUID(material),
    )).status == "archived"


async def test_the_worklist_says_which_items_need_a_quiz(
    client, db_session, security_officer, reader,
):
    """``/my/campaigns`` carries ``has_quiz``.

    Without it neither the todo list nor the ``list_my_security_awareness``
    MCP tool can tell somebody what kind of work an item is; they would
    find out from a refused confirm, which is the worst place to learn it.
    """
    _officer, oh = security_officer
    _person, rh = reader
    plain = await _publish(client, oh)
    quizzed = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{quizzed}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    await _send(client, db_session, oh, plain, name="无小测")
    await _send(client, db_session, oh, quizzed, name="有小测")

    by_material = {
        r["material_id"]: r
        for r in (await client.get(MY, headers=rh)).json()["data"]
    }
    assert by_material[plain]["has_quiz"] is False
    assert by_material[quizzed]["has_quiz"] is True


async def test_the_console_says_which_materials_can_actually_be_deleted(
    client, db_session, security_officer, reader,
):
    """``deletable`` mirrors the server's guard, all three branches.

    The console withholds the delete button rather than offering one
    whose only outcome is a 409 — but an impressions-only test would
    still have offered it for a campaign nobody has opened yet, which is
    exactly the case that used to reach MySQL and come back a 500.
    """
    _officer, oh = security_officer
    _person, rh = reader

    async def _row(material_id: str) -> dict:
        rows = (await client.get(
            f"{ADMIN_BASE}/materials?page_size=100", headers=oh,
        )).json()["data"]
        return next(r for r in rows if r["id"] == material_id)

    fresh = await _publish(client, oh)
    assert (await _row(fresh))["deletable"] is True

    # (1) shown to somebody
    shown = await _publish(client, oh)
    await client.get(f"{SELF_BASE}/materials/{shown}", headers=rh)
    assert (await _row(shown))["deletable"] is False

    # (2) carries a campaign, never opened by anyone
    campaigned = await _publish(client, oh)
    await client.post(
        CAMPAIGNS,
        json={
            "material_id": campaigned, "name": "草稿", "audience_config": {},
            "channels": [], "mandatory": False, "completion_mode": "ack",
        },
        headers=oh,
    )
    assert (await _row(campaigned))["deletable"] is False

    # (3) somebody answered its quiz
    answered = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{answered}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    await client.post(
        f"{SELF_BASE}/materials/{answered}/quiz",
        json={"answers": {str(q.id): "B"}}, headers=rh,
    )
    assert (await _row(answered))["deletable"] is False

    # The flag and the endpoint agree — that is the whole point of it.
    assert (await client.delete(
        f"{ADMIN_BASE}/materials/{fresh}", headers=oh,
    )).status_code == 204


# ════════════════════════════════════════════════════════════════════
# Retiring content does not retract an obligation
# ════════════════════════════════════════════════════════════════════


@pytest.mark.parametrize("retire_to", ["archived", "draft"])
async def test_archiving_a_material_does_not_strand_a_live_campaign(
    client, db_session, security_officer, reader, retire_to,
):
    """The straggler's todo list, the reminder cron and the link have to
    agree.

    A material retired from the library 404s for everyone — which used to
    include the people a live campaign was still chasing about it. Their
    todo row named it, the cron kept pushing about it, and the link went
    nowhere. Retiring content is a decision about the LIBRARY; it does
    not retract an obligation somebody was already given.
    """
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session)
    await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=oh,
    )
    campaign = await _send(client, db_session, oh, material)
    await client.post(
        f"{ADMIN_BASE}/materials/{material}/status",
        json={"status": retire_to}, headers=oh,
    )

    # Still owed, and still readable BY THEM.
    assert len((await client.get(MY, headers=rh)).json()["data"]) == 1
    detail = await client.get(f"{SELF_BASE}/materials/{material}", headers=rh)
    assert detail.status_code == 200, detail.text
    assert detail.json()["data"]["pending_ack"]["campaign_id"] == campaign
    # …including the quiz that is the only way to complete it.
    assert (await client.get(
        f"{SELF_BASE}/materials/{material}/quiz", headers=rh,
    )).status_code == 200
    passed = await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )
    assert passed.status_code == 200, passed.text
    assert passed.json()["data"]["passed"] is True
    assert (await client.get(MY, headers=rh)).json()["data"] == []

    # Somebody who owes nothing still gets a 404 — the exception is the
    # obligation, not the material.
    stranger = await make_person(db_session, "stranger@test.com")
    await db_session.commit()
    sh = headers_for(stranger, ["employee"])
    assert (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=sh,
    )).status_code == 404
    assert (await client.get(
        f"{SELF_BASE}/materials/{material}/quiz", headers=sh,
    )).status_code == 404


async def test_closing_the_campaign_retires_the_material_for_them_too(
    client, db_session, security_officer, reader,
):
    """The exception is the live obligation, and closing ends it.

    Otherwise "archived + closed" would still be readable forever by
    whoever happened to be on the target list, which is not what either
    act means.
    """
    _officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    campaign = await _send(client, db_session, oh, material)
    await client.post(
        f"{ADMIN_BASE}/materials/{material}/status",
        json={"status": "archived"}, headers=oh,
    )
    assert (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).status_code == 200

    await client.post(f"{CAMPAIGNS}/{campaign}/close", headers=oh)
    assert (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).status_code == 404


async def test_the_evidence_file_counts_automatic_reminders_too(
    client, db_session, security_officer,
):
    """``reminders_sent`` is "did we chase them", not "did an admin press
    resend".

    The ladder is the chasing — it is the whole reminder design — and a
    file that printed 0 next to somebody the cron nudged would answer one
    of the questions the file exists for with the wrong number.
    """
    from app.domains.security_awareness.reminders import ReminderService

    _officer, oh = security_officer
    person = await make_person(db_session, "straggler@test.com")
    await db_session.commit()
    rh = headers_for(person, ["employee"])
    material = await _publish(client, oh)
    campaign = await _send(client, db_session, oh, material)

    async def evidence_for(email: str) -> dict:
        raw = (await client.get(
            f"{ADMIN_BASE}/campaigns/{campaign}/evidence.csv", headers=oh,
        )).text
        rows = [r.split(",") for r in raw.lstrip("").splitlines()]
        header = rows[0]
        for row in rows[1:]:
            entry = dict(zip(header, row, strict=False))
            if entry.get("email") == email:
                return entry
        raise AssertionError(f"{email} not in the evidence file")

    assert (await evidence_for(person.work_email))["reminders_sent"] == "0"

    # Age the campaign past the first chase rung and run the cron.
    old = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=10)
    (await db_session.get(SecawCampaign, uuid.UUID(campaign))).sent_at = old
    await db_session.commit()
    with patch(_FANOUT, new=AsyncMock(return_value=dict(_FANOUT_OK))):
        summary = await ReminderService(db_session).run()
    assert summary["employee_pushes"] > 0

    assert (await evidence_for(person.work_email))["reminders_sent"] == "1"
    # And the acknowledged reader is untouched by the chase.
    await client.post(f"{MY}/{campaign}/acknowledge", headers=rh)