"""素材小测 —— grading, the acknowledgement it replaces, and the points.

The load-bearing assertions:

* **The answer key never reaches the employee's browser.** The kind of
  thing only noticed after somebody reads the network tab.
* **Passing IS the acknowledgement.** Two controls for one claim would
  let somebody assert understanding without demonstrating it.
* **Grading matches the security-quiz domain's own normalisation.** The
  helper is restated here (domains do not import each other), so one test
  pins the two implementations against the same inputs.
* **Points never cost somebody the compliance record.**
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import select

from app.domains.security_awareness.models import (
    SecawCampaignTarget,
    SecawQuizAttempt,
)
from app.domains.security_quiz.models import SecQuizQuestion
from tests.test_security_awareness.conftest import (
    ADMIN_BASE,
    SELF_BASE,
    article_payload,
    fanout_double,
    send_now,
)

pytestmark = pytest.mark.asyncio

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


async def _question(db, prompt: str, correct: str = "A", qtype: str = "single"):
    q = SecQuizQuestion(
        id=uuid.uuid4(),
        prompt=prompt,
        options=[
            {"key": "A", "text": "对的"},
            {"key": "B", "text": "错的"},
            {"key": "C", "text": "也错"},
        ],
        qtype=qtype,
        correct_key=correct,
        explanation=f"因为 {correct}",
        category="phishing",
    )
    db.add(q)
    await db.flush()
    await db.commit()
    return q


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 _attach(client, headers, material, question_ids) -> None:
    r = await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q) for q in question_ids]},
        headers=headers,
    )
    assert r.status_code == 200, r.text


# ── The answer key stays server-side ────────────────────────────────


async def test_the_employee_endpoint_never_returns_the_answer_key(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "钓鱼邮件最常见的破绽是?")
    await _attach(client, oh, material, [q.id])

    body = (await client.get(
        f"{SELF_BASE}/materials/{material}/quiz", headers=rh,
    )).json()["data"]
    assert body["questions"][0]["prompt"]
    assert "correct_key" not in body["questions"][0]
    assert "explanation" not in body["questions"][0]

    # The admin endpoint does return it — that is the editor's whole job.
    admin_body = (await client.get(
        f"{ADMIN_BASE}/materials/{material}/quiz", headers=oh,
    )).json()["data"]
    assert admin_body["questions"][0]["correct_key"] == "A"


# ── Grading ─────────────────────────────────────────────────────────


async def test_grading_matches_the_security_quiz_domains_normalisation():
    """The helper is restated in this domain (domains do not import each
    other). This pins the two implementations against the same inputs so
    they cannot drift into disagreeing about what a right answer is."""
    from app.domains.security_awareness.quiz import normalize_answer
    from app.domains.security_quiz.service import _normalize_keys

    for raw in ("ab", "BA", "a b", " A ", "", "ACB", "aab"):
        assert normalize_answer(raw) == _normalize_keys(raw), raw
    # The list form the frontend sends for a multi-select normalises to
    # the same string as its concatenation.
    assert normalize_answer(["b", "a"]) == _normalize_keys("ab")


async def test_a_wrong_answer_returns_the_explanation_and_can_be_retried(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    miss = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "B"}}, headers=rh,
    )).json()["data"]
    assert miss["passed"] is False
    assert miss["correct_count"] == 0
    assert miss["results"][0]["correct_key"] == "A"
    assert miss["results"][0]["explanation"] == "因为 A"

    hit = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )).json()["data"]
    assert hit["passed"] is True
    # A right answer does NOT hand back the key — otherwise one wrong
    # answer on a retry becomes a full answer sheet.
    assert hit["results"][0]["correct_key"] is None


async def test_every_question_must_be_right(
    client, db_session, security_officer, reader,
):
    """Two or three questions written for one short piece do not support
    a meaningful partial score, and "2 of 3 counts" is a weaker claim
    than the plain acknowledgement it replaces."""
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q1 = await _question(db_session, "一", correct="A")
    q2 = await _question(db_session, "二", correct="B")
    await _attach(client, oh, material, [q1.id, q2.id])

    partial = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q1.id): "A", str(q2.id): "C"}}, headers=rh,
    )).json()["data"]
    assert partial["correct_count"] == 1
    assert partial["passed"] is False


async def test_a_blank_answer_is_not_a_pass(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选点什么", correct="A")
    await _attach(client, oh, material, [q.id])

    r = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {}}, headers=rh,
    )).json()["data"]
    assert r["passed"] is False


async def test_every_attempt_is_kept_including_the_failures(
    client, db_session, security_officer, reader,
):
    """Which questions people get wrong the first time is the most direct
    evidence the security team has that a passage is unclear."""
    officer, oh = security_officer
    person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    for answer in ("B", "C", "A"):
        await client.post(
            f"{SELF_BASE}/materials/{material}/quiz",
            json={"answers": {str(q.id): answer}}, headers=rh,
        )

    attempts = (await db_session.execute(
        select(SecawQuizAttempt).where(
            SecawQuizAttempt.person_id == person.id,
            SecawQuizAttempt.material_id == uuid.UUID(material),
        )
        # Explicit ORDER BY, because the assertion below is about
        # sequence. Without it the rows come back in whatever order the
        # chosen index happens to yield — and which index the optimiser
        # picks depends on table statistics, i.e. on what OTHER tests in
        # the session inserted. This file passed alone and failed inside
        # a larger run for exactly that reason. ``created_at`` is
        # second-granularity and all three attempts land in the same
        # second, so ``attempt_no`` is the only ordering that is actually
        # a sequence — which is the same reason the column exists.
        .order_by(SecawQuizAttempt.attempt_no.asc())
    )).scalars().all()
    assert len(attempts) == 3
    assert [a.attempt_no for a in attempts] == [1, 2, 3]
    assert [a.passed for a in attempts] == [False, False, True]
    # And the version they answered against is on the row — the material
    # gets edited afterwards, and that is what a dispute is about.
    assert all(a.material_version == 1 for a in attempts)


# ── Passing IS the acknowledgement ──────────────────────────────────


async def test_passing_the_quiz_acknowledges_the_pending_campaign(
    client, db_session, security_officer, reader,
):
    """Two controls for one claim would let somebody assert understanding
    without demonstrating it — which is the gap the quiz exists to
    close."""
    from unittest.mock import patch

    officer, oh = security_officer
    person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    campaign = (await client.post(
        CAMPAIGNS,
        json={
            "material_id": material, "name": "必修", "audience_config": {},
            "channels": ["feishu"], "mandatory": True,
            "completion_mode": "ack",
        },
        headers=oh,
    )).json()["data"]["id"]
    with patch(
        "app.notifications.fanout.enqueue_then_drain", new=fanout_double(),
    ):
        await send_now(client, db_session, campaign, oh)

    # A miss does NOT acknowledge.
    miss = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "B"}}, headers=rh,
    )).json()["data"]
    assert miss["acknowledged_campaign_id"] is None
    assert len((await client.get(MY, headers=rh)).json()["data"]) == 1

    hit = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )).json()["data"]
    assert hit["acknowledged_campaign_id"] == campaign
    assert (await client.get(MY, headers=rh)).json()["data"] == []


async def test_the_material_page_says_which_control_to_render(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)

    before = (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).json()["data"]
    assert before["has_quiz"] is False
    assert before["quiz_passed"] is False

    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    after = (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).json()["data"]
    assert after["has_quiz"] is True
    assert after["quiz_passed"] is False

    await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )
    done = (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=rh,
    )).json()["data"]
    assert done["quiz_passed"] is True


async def test_a_material_without_a_quiz_refuses_a_submission(
    client, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    r = await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {}}, headers=rh,
    )
    assert r.status_code == 409
    assert r.json()["errors"][0]["code"] == "SECAW_NO_QUIZ"


# ── Points ──────────────────────────────────────────────────────────


async def test_points_are_awarded_once_per_material(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    first = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )).json()["data"]
    assert first["points_awarded"] > 0

    # Passing again credits nothing — otherwise somebody could farm the
    # wallet by re-submitting, and the moment points become the reason to
    # click, the acknowledgement stops meaning anything.
    again = (await client.post(
        f"{SELF_BASE}/materials/{material}/quiz",
        json={"answers": {str(q.id): "A"}}, headers=rh,
    )).json()["data"]
    assert again["points_awarded"] == 0


async def test_a_points_failure_never_costs_the_acknowledgement(
    client, db_session, security_officer, reader,
):
    """The caller has just written a compliance record; a wallet hiccup
    must not undo it.

    This covers the *Python* half only — a ``RuntimeError`` is swallowed
    cleanly by any handler. The DB half, which needs a SAVEPOINT and is
    the half that actually broke, is the test below.
    """
    from unittest.mock import AsyncMock, patch

    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    with patch(
        "app.domains.arcade.economy.get_wallet_locked",
        new=AsyncMock(side_effect=RuntimeError("wallet is down")),
    ):
        r = await client.post(
            f"{SELF_BASE}/materials/{material}/quiz",
            json={"answers": {str(q.id): "A"}}, headers=rh,
        )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["passed"] is True
    assert data["points_awarded"] == 0


async def test_a_failed_points_WRITE_still_lets_the_ack_commit(
    client, db_session, security_officer, reader,
):
    """Regression: the points write must be SAVEPOINT-isolated.

    A failed *flush* is not a Python exception — it deactivates the
    caller's WHOLE transaction. So a bare ``except Exception`` around
    the wallet write "succeeds", the function dutifully returns 0, and
    then the caller's own commit — the one carrying the acknowledgement
    written moments earlier — raises ``PendingRollbackError``. The
    compliance record is lost by the very code that promised to protect
    it. That is the 2026-08-06 Zoom-seat bug class (PR #2571), and
    ``scripts/check_savepoint_isolation.py`` exists to keep it out.

    The failure is injected as a REAL MySQL error on the ledger INSERT
    (a value too long for ``reason``, ``VARCHAR(32)``) rather than a
    mocked ``raise``: only a real flush failure poisons a transaction,
    so only a real one can tell the fix from the bug. It is deliberately
    a *value*-level error rather than a duplicate-key one — the
    coordinator's own "already credited?" pre-check would see a planted
    duplicate and return early, which would make this test pass without
    ever reaching a flush.
    """
    from unittest.mock import patch

    from app.domains.arcade.models import ArcadePointsLedger

    officer, oh = security_officer
    person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    campaign = (await client.post(
        CAMPAIGNS,
        json={
            "material_id": material, "name": "必修", "audience_config": {},
            "channels": ["feishu"], "mandatory": True,
            "completion_mode": "ack",
        },
        headers=oh,
    )).json()["data"]["id"]
    with patch(
        "app.notifications.fanout.enqueue_then_drain", new=fanout_double(),
    ):
        await send_now(client, db_session, campaign, oh)

    with patch(
        "app.core.coordinators.awareness_points_coordinator"
        ".REASON_AWARENESS_QUIZ",
        new="x" * 64,  # VARCHAR(32) — MySQL 1406 on the INSERT
    ):
        r = await client.post(
            f"{SELF_BASE}/materials/{material}/quiz",
            json={"answers": {str(q.id): "A"}}, headers=rh,
        )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["passed"] is True
    assert data["acknowledged_campaign_id"] == campaign
    assert data["points_awarded"] == 0

    # The load-bearing assertion: the session the acknowledgement was
    # written on is still usable, so the request's commit lands. Without
    # the savepoint this raises PendingRollbackError.
    await db_session.commit()
    target = (await db_session.execute(
        select(SecawCampaignTarget).where(
            SecawCampaignTarget.person_id == person.id,
        )
    )).scalar_one()
    assert target.acknowledged_at is not None

    # And nothing was credited — the savepoint rolled the whole attempt
    # back, planted row included.
    assert (await db_session.execute(
        select(ArcadePointsLedger).where(
            ArcadePointsLedger.person_id == person.id,
        )
    )).scalars().all() == []


# ── Admin: attaching, and the difficulty read ───────────────────────


async def test_setting_the_question_set_replaces_it_in_order(
    client, db_session, security_officer,
):
    officer, oh = security_officer
    material = await _publish(client, oh)
    q1 = await _question(db_session, "一")
    q2 = await _question(db_session, "二")
    q3 = await _question(db_session, "三")

    await _attach(client, oh, material, [q1.id, q2.id])
    body = (await client.get(
        f"{ADMIN_BASE}/materials/{material}/quiz", headers=oh,
    )).json()["data"]
    assert [q["prompt"] for q in body["questions"]] == ["一", "二"]

    await _attach(client, oh, material, [q3.id, q1.id])
    body = (await client.get(
        f"{ADMIN_BASE}/materials/{material}/quiz", headers=oh,
    )).json()["data"]
    assert [q["prompt"] for q in body["questions"]] == ["三", "一"]

    # Removing them all falls the control back to the plain button.
    await _attach(client, oh, material, [])
    assert (await client.get(
        f"{SELF_BASE}/materials/{material}", headers=oh,
    )).json()["data"]["has_quiz"] is False


async def test_a_quiz_longer_than_the_cap_is_refused(
    client, db_session, security_officer,
):
    """A longer quiz is a test, and this is not a test."""
    officer, oh = security_officer
    material = await _publish(client, oh)
    ids = [(await _question(db_session, f"q{n}")).id for n in range(11)]
    r = await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(i) for i in ids]}, headers=oh,
    )
    assert r.status_code == 400
    assert r.json()["errors"][0]["code"] == "SECAW_QUIZ_TOO_LONG"


async def test_difficulty_counts_only_the_first_attempt_per_person(
    client, db_session, security_officer, reader,
):
    """Later attempts are contaminated by the explanation the reader was
    just shown, so counting them would make every question look easy."""
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "选 A", correct="A")
    await _attach(client, oh, material, [q.id])

    # Miss, then pass — the first attempt is what counts.
    for answer in ("B", "A"):
        await client.post(
            f"{SELF_BASE}/materials/{material}/quiz",
            json={"answers": {str(q.id): answer}}, headers=rh,
        )

    body = (await client.get(
        f"{ADMIN_BASE}/materials/{material}/quiz", headers=oh,
    )).json()["data"]
    row = body["difficulty"][0]
    assert row["asked"] == 1
    assert row["missed"] == 1
    assert row["miss_rate_pct"] == 100


async def test_employees_cannot_edit_the_question_set(
    client, db_session, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)
    q = await _question(db_session, "一")
    assert (await client.put(
        f"{ADMIN_BASE}/materials/{material}/quiz",
        json={"question_ids": [str(q.id)]}, headers=rh,
    )).status_code == 403
    assert (await client.get(
        f"{ADMIN_BASE}/materials/{material}/quiz", headers=rh,
    )).status_code == 403


# ── Participation ───────────────────────────────────────────────────


async def test_participation_separates_bubble_reach_from_opens(
    client, db_session, security_officer, reader,
):
    """"Shown 900 times" and "reached 300 people" answer different
    questions, and only the second one is engagement."""
    officer, oh = security_officer
    _person, rh = reader
    material = await _publish(client, oh)

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

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


async def test_participation_is_read_gated(client, security_officer, reader):
    officer, _oh = security_officer
    _person, rh = reader
    assert (await client.get(
        f"{ADMIN_BASE}/participation", headers=rh,
    )).status_code == 403


# ── Badges ──────────────────────────────────────────────────────────


async def test_the_awareness_badges_measure_reading_not_being_shown(
    client, db_session, security_officer, reader,
):
    """A badge that rewarded passive exposure would reward leaving a tab
    open."""
    from app.domains.gamification.catalog import BADGES
    from app.domains.gamification.facts import compute_facts_for

    officer, oh = security_officer
    person, rh = reader
    material = await _publish(client, oh)

    await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": material, "surface": "bubble"}, headers=rh,
    )
    facts = await compute_facts_for(db_session, [person])
    assert facts[person.id]["awareness_materials_read"] == 0

    await client.get(f"{SELF_BASE}/materials/{material}", headers=rh)
    facts = await compute_facts_for(db_session, [person])
    assert facts[person.id]["awareness_materials_read"] == 1

    reader_badge = next(b for b in BADGES if b.code == "awareness_reader")
    assert reader_badge.is_earned(facts[person.id]) is True
    quiz_badge = next(b for b in BADGES if b.code == "awareness_quiz_first")
    assert quiz_badge.is_earned(facts[person.id]) is False


async def test_badge_metrics_are_registered_facts():
    """A typo in a badge's ``metric`` would make it permanently
    unearnable, silently."""
    from app.domains.gamification.catalog import BADGES
    from app.domains.gamification.facts import FACT_KEYS

    for badge in BADGES:
        if badge.metric is not None:
            assert badge.metric in FACT_KEYS, badge.code


async def test_the_security_desk_can_read_the_question_bank(
    client, db_session, security_officer, reader,
):
    """The attach picker needs the question bank, and the security-quiz
    domain's own admin list gates on ``global_admin`` / ``auditor`` —
    which does NOT admit ``security_officer``, the role that actually
    runs awareness.

    Rather than widening that gate (which would hand the desk write
    access to the daily-quiz bank as a side effect), this domain serves a
    narrow read-only picker. Same shape of hole as the identity-sync
    lookups in PR2, found the same way: by asking who can actually reach
    the endpoint the new UI depends on.
    """
    _officer, oh = security_officer
    _person, rh = reader
    await _question(db_session, "题库里的一道题")

    r = await client.get(f"{ADMIN_BASE}/question-bank", headers=oh)
    assert r.status_code == 200, r.text
    assert any(q["prompt"] == "题库里的一道题" for q in r.json()["data"])

    # It is a read surface, not an open one.
    assert (await client.get(
        f"{ADMIN_BASE}/question-bank", headers=rh,
    )).status_code == 403


async def test_the_question_bank_picker_is_read_only_and_active_only(
    client, db_session, security_officer,
):
    """A retired question must not be offered for attachment — it would
    render to employees as a question the bank no longer stands behind."""
    _officer, oh = security_officer
    live = await _question(db_session, "在用的")
    retired = await _question(db_session, "退役的")
    retired.status = "archived"
    await db_session.commit()

    prompts = {
        q["prompt"] for q in (await client.get(
            f"{ADMIN_BASE}/question-bank", headers=oh,
        )).json()["data"]
    }
    assert "在用的" in prompts
    assert "退役的" not in prompts
    assert live.id