"""安全问卷 · 供应商安全评估 — scoring, lifecycle, xlsx round-trip, RBAC.

The test schema is built via ``create_all`` (no migrations), so the two
seed system templates from migration 374 are absent — every test seeds
its own template + questions, keeping them hermetic.
"""
from __future__ import annotations

import io
import uuid
from datetime import date

import pytest
from httpx import AsyncClient
from openpyxl import load_workbook
from sqlalchemy import select

from app.domains.auditor import vendor_assessment_service as svc
from app.domains.auditor.models import (
    AuditorFinding,
    AuditorSqTemplate,
    AuditorSqTemplateQuestion,
    AuditorVendorAssessmentAnswer,
    AuditorVendorAssessmentAttachment,
)

# ── pure scoring ──────────────────────────────────────────────────────


class _Ans:
    """Lightweight stand-in for scoring unit tests."""

    def __init__(self, answer_type, weight, value, expected=None, options=None):
        self.answer_type = answer_type
        self.weight = weight
        self.answer_value = value
        self.expected_answer = expected
        self.options = options


def test_score_answer_yes_no():
    assert svc.score_answer(_Ans("yes_no", 4, "yes", "yes")) == (4, 4)
    assert svc.score_answer(_Ans("yes_no", 4, "no", "yes")) == (0, 4)
    # blank scorable question = 0 out of weight (undemonstrated control)
    assert svc.score_answer(_Ans("yes_no", 4, "", "yes")) == (0, 4)


def test_score_answer_na_excluded():
    assert svc.score_answer(_Ans("yes_no_na", 3, "na", "yes")) == (None, None)
    assert svc.score_answer(_Ans("yes_no_na", 3, "yes", "yes")) == (3, 3)


def test_score_answer_scale_and_text():
    assert svc.score_answer(_Ans("scale_1_5", 5, "3")) == (3, 5)
    assert svc.score_answer(_Ans("scale_1_5", 5, "bad")) == (0, 5)
    assert svc.score_answer(_Ans("text", 0, "anything")) == (None, None)


def test_score_answer_single_choice():
    opts = [
        {"value": "iso27001", "score": 1.0},
        {"value": "in_progress", "score": 0.5},
        {"value": "none", "score": 0.0},
    ]
    assert svc.score_answer(_Ans("single_choice", 4, "iso27001", options=opts)) == (4, 4)
    assert svc.score_answer(_Ans("single_choice", 4, "in_progress", options=opts)) == (2, 4)
    assert svc.score_answer(_Ans("single_choice", 4, "none", options=opts)) == (0, 4)


def test_compute_rating_thresholds():
    assert svc.compute_rating(95) == "low"
    assert svc.compute_rating(80) == "medium"
    assert svc.compute_rating(65) == "high"
    assert svc.compute_rating(30) == "critical"
    assert svc.compute_rating(None) is None


def test_add_months_clamps_month_end():
    assert svc._add_months(date(2026, 1, 31), 1) == date(2026, 2, 28)
    assert svc._add_months(date(2026, 7, 2), 12) == date(2027, 7, 2)


# ── helpers ──────────────────────────────────────────────────────────


async def _seed_template(db, *, key="t_test"):
    t = AuditorSqTemplate(
        key=key, name_zh="测试模板", name_en="Test template",
        kind="vendor", tier="full", is_system=False, enabled=True, sort_order=0,
    )
    db.add(t)
    await db.flush()
    specs = [
        ("Q1", "yes_no", 4, "yes", None),
        ("Q2", "yes_no_na", 3, "yes", None),
        ("Q3", "scale_1_5", 5, None, None),
        ("Q4", "single_choice", 4, None,
         [{"value": "a", "label_zh": "甲", "label_en": "A", "score": 1.0},
          {"value": "b", "label_zh": "乙", "label_en": "B", "score": 0.0}]),
        ("Q5", "text", 0, None, None),
    ]
    for i, (code, at, w, exp, opts) in enumerate(specs, 1):
        db.add(AuditorSqTemplateQuestion(
            template_id=t.id, code=code, section_key="s1",
            section_zh="分区", section_en="Section", text_zh=f"问题{code}",
            text_en=f"Question {code}", answer_type=at, options=opts,
            weight=w, expected_answer=exp, sort_order=i * 10,
        ))
    await db.flush()
    return t


# ── lifecycle ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_create_materialises_answers(db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "Acme Cloud",
                             "tier": "high"}, actor_id=None,
    )
    await db_session.commit()
    answers = (await db_session.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
        )
    )).scalars().all()
    assert len(answers) == 5
    assert a.status == "draft"
    assert a.review_period_months == 12  # high tier default


@pytest.mark.asyncio
async def test_scoring_and_complete(db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "V", "tier": "medium"},
        actor_id=None,
    )
    await db_session.commit()
    answers = {
        x.question_code: x
        for x in (await db_session.execute(
            select(AuditorVendorAssessmentAnswer).where(
                AuditorVendorAssessmentAnswer.assessment_id == a.id,
            )
        )).scalars().all()
    }
    # Q1 yes(4/4), Q2 na(excluded), Q3 scale 5(5/5), Q4 'a'(4/4), Q5 text(none)
    await svc.set_answer(db_session, answers["Q1"].id, payload={"answer_value": "yes"})
    await svc.set_answer(db_session, answers["Q2"].id, payload={"answer_value": "na"})
    await svc.set_answer(db_session, answers["Q3"].id, payload={"answer_value": "5"})
    await svc.set_answer(db_session, answers["Q4"].id, payload={"answer_value": "a"})
    await db_session.commit()
    a = await svc.complete_assessment(
        db_session, a.id, summary="ok", reviewer_person_id=None,
    )
    await db_session.commit()
    # earned 13 / max 13 → 100 → low
    assert a.score_earned == 13
    assert a.score_max == 13
    assert a.score_pct == 100
    assert a.risk_rating == "low"
    assert a.status == "completed"
    assert a.next_review_date is not None


@pytest.mark.asyncio
async def test_rating_override_sticks(db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "V", "tier": "low"},
        actor_id=None,
    )
    await db_session.commit()
    await svc.override_rating(db_session, a.id, rating="critical")
    await db_session.commit()
    a = await svc.get_assessment(db_session, a.id)
    assert a.rating_overridden is True
    assert a.risk_rating == "critical"
    # completing must NOT clobber the manual override
    a = await svc.complete_assessment(db_session, a.id, summary=None, reviewer_person_id=None)
    await db_session.commit()
    assert a.risk_rating == "critical"


@pytest.mark.asyncio
async def test_finding_promotion_and_resolution(db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "RiskCo", "tier": "high"},
        actor_id=None,
    )
    await db_session.commit()
    ans = (await db_session.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
            AuditorVendorAssessmentAnswer.question_code == "Q1",
        )
    )).scalar_one()
    await svc.review_answer(db_session, ans.id, review_flag="finding", reviewer_note="gap")
    await db_session.commit()
    await svc.complete_assessment(db_session, a.id, summary=None, reviewer_person_id=None)
    await db_session.commit()
    findings = (await db_session.execute(
        select(AuditorFinding).where(
            AuditorFinding.control_key == "vendor_security_assessment",
        )
    )).scalars().all()
    assert len(findings) == 1
    assert findings[0].status == "open"
    # un-flag + re-complete → finding auto-resolves
    await svc.review_answer(db_session, ans.id, review_flag="ok", reviewer_note=None)
    await db_session.commit()
    await svc.complete_assessment(db_session, a.id, summary=None, reviewer_person_id=None)
    await db_session.commit()
    f = (await db_session.execute(
        select(AuditorFinding).where(
            AuditorFinding.control_key == "vendor_security_assessment",
        )
    )).scalar_one()
    assert f.status == "resolved"


# ── xlsx round-trip ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_questionnaire_roundtrip(db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "V", "tier": "medium"},
        actor_id=None,
    )
    await db_session.commit()
    answers = list((await db_session.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
        ).order_by(AuditorVendorAssessmentAnswer.sort_order)
    )).scalars().all())
    # Build the blank workbook, fill Q1's answer cell, re-parse.
    buf = svc.build_blank_questionnaire(a, answers)
    wb = load_workbook(io.BytesIO(buf.getvalue()))
    ws = wb.active
    # find the Q1 row and its answer column (col 5), set "Yes"
    for row in ws.iter_rows():
        if row[0].value == "Q1":
            ws.cell(row=row[0].row, column=5, value="Yes")
            ws.cell(row=row[0].row, column=6, value="we do this")
    out = io.BytesIO()
    wb.save(out)
    result = await svc.import_answers(db_session, a.id, payload=out.getvalue())
    await db_session.commit()
    assert result["updated"] == 1
    q1 = (await db_session.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
            AuditorVendorAssessmentAnswer.question_code == "Q1",
        )
    )).scalar_one()
    assert q1.answer_value == "yes"  # normalised from "Yes"
    assert q1.answer_comment == "we do this"


def test_parse_rejects_wrong_workbook():
    from openpyxl import Workbook
    wb = Workbook()
    wb.active["A1"] = "not a questionnaire"
    buf = io.BytesIO()
    wb.save(buf)
    with pytest.raises(Exception):
        svc.parse_filled_questionnaire(buf.getvalue())


# ── review reminder staging ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_review_reminder_stages(db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "V", "tier": "medium"},
        actor_id=None,
    )
    a.status = "completed"
    a.next_review_date = date(2026, 8, 1)
    await db_session.commit()
    # 40 days out → not yet in any stage
    due = await svc.find_due_review_reminders(db_session, ref=date(2026, 6, 22))
    assert due == []
    # 20 days out → "14"? no, 20 > 14 but <= 30 → stage "30"
    due = await svc.find_due_review_reminders(db_session, ref=date(2026, 7, 12))
    assert len(due) == 1 and due[0][1] == "30"
    svc.mark_reminder_sent(due[0][0], "30")
    await db_session.commit()
    # same window again → already reminded at "30", nothing new
    due = await svc.find_due_review_reminders(db_session, ref=date(2026, 7, 13))
    assert due == []
    # overdue → new stage fires
    due = await svc.find_due_review_reminders(db_session, ref=date(2026, 8, 5))
    assert len(due) == 1 and due[0][1] == "overdue"


def test_reminder_stage_for_brackets_are_reachable():
    """Regression (2026-07-03 audit): tightest bracket must win so the
    14-day and due-day reminders are reachable — not shadowed by "30"."""
    assert svc._reminder_stage_for(40) is None
    assert svc._reminder_stage_for(30) == "30"
    assert svc._reminder_stage_for(20) == "30"
    assert svc._reminder_stage_for(14) == "14"
    assert svc._reminder_stage_for(10) == "14"
    assert svc._reminder_stage_for(1) == "14"
    assert svc._reminder_stage_for(0) == "due"
    assert svc._reminder_stage_for(-1) == "overdue"


@pytest.mark.asyncio
async def test_review_reminder_14_and_due_fire(db_session):
    """The 14-day and due-day escalations must fire as the review nears
    (previously shadowed by the "30" stage → the owner got one 30-day
    nudge then silence until the review had already lapsed)."""
    t = await _seed_template(db_session, key="t_rem2")
    await db_session.commit()
    a = await svc.create_assessment(
        db_session,
        payload={"template_id": t.id, "vendor_name": "V2", "tier": "medium"},
        actor_id=None,
    )
    a.status = "completed"
    a.next_review_date = date(2026, 8, 1)
    await db_session.commit()
    # 20 days out → "30", mark sent
    due = await svc.find_due_review_reminders(db_session, ref=date(2026, 7, 12))
    assert len(due) == 1 and due[0][1] == "30"
    svc.mark_reminder_sent(due[0][0], "30")
    await db_session.commit()
    # 12 days out → "14" (previously returned "30" → guard-skipped → nothing)
    due = await svc.find_due_review_reminders(db_session, ref=date(2026, 7, 20))
    assert len(due) == 1 and due[0][1] == "14"
    svc.mark_reminder_sent(due[0][0], "14")
    await db_session.commit()
    # due day → "due"
    due = await svc.find_due_review_reminders(db_session, ref=date(2026, 8, 1))
    assert len(due) == 1 and due[0][1] == "due"


# ── HTTP RBAC ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_overview_rbac(client: AsyncClient, auditor_headers, employee_headers):
    r = await client.get(
        "/api/v1/auditor/vendor-assessments/overview", headers=auditor_headers,
    )
    assert r.status_code == 200
    assert "summary" in r.json()["data"]
    r = await client.get(
        "/api/v1/auditor/vendor-assessments/overview", headers=employee_headers,
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_employee_cannot_create(client: AsyncClient, employee_headers, db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    r = await client.post(
        "/api/v1/auditor/vendor-assessments",
        headers=employee_headers,
        json={"template_id": str(t.id), "vendor_name": "X", "tier": "low"},
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_create_and_get_flow(client: AsyncClient, auditor_headers, db_session):
    t = await _seed_template(db_session)
    await db_session.commit()
    r = await client.post(
        "/api/v1/auditor/vendor-assessments",
        headers=auditor_headers,
        json={"template_id": str(t.id), "vendor_name": "Acme", "tier": "critical"},
    )
    assert r.status_code == 200, r.text
    aid = r.json()["data"]["id"]
    r = await client.get(
        f"/api/v1/auditor/vendor-assessments/{aid}", headers=auditor_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["vendor_name"] == "Acme"
    assert data["total_count"] == 5
    assert len(data["sections"]) == 1


@pytest.mark.asyncio
async def test_attachment_download_writes_an_audit_row(
    client: AsyncClient, auditor_headers, db_session, monkeypatch,
):
    """Sibling of the self-check test — same egress, same trace (#2161).

    Vendor evidence is the more sensitive of the two (pen-test reports,
    certificates), and ``management`` can reach it through
    ``can_view_security``, so an untraced presign was the worse half.
    """
    from app.audit.models import AuditLog

    t = await _seed_template(db_session, key="t_dl_audit")
    await db_session.commit()
    created = await client.post(
        "/api/v1/auditor/vendor-assessments",
        headers=auditor_headers,
        json={"template_id": str(t.id), "vendor_name": "Egress",
              "tier": "critical"},
    )
    aid = created.json()["data"]["id"]

    async def _fake_stat(_db, *, object_key, **_kw):
        return {"size_bytes": 1024, "content_type": "application/pdf"}

    async def _fake_presign(_db, *, object_key, **_kw):
        return f"https://s3.example/{object_key}?sig=x"

    monkeypatch.setattr(
        "app.domains.auditor.vendor_assessment_service.stat_object",
        _fake_stat,
    )
    claimed_resource_ids = []

    async def _fake_claim(_db, *, object_keys, **_kw):
        from types import SimpleNamespace

        claimed_resource_ids.append(_kw["resource_id"])
        return {
            key: SimpleNamespace(
                declared_size_bytes=1024, content_type="image/png",
            )
            for key in object_keys
        }
    monkeypatch.setattr("app.core.storage_uploads.mark_uploads_referenced", _fake_claim)
    monkeypatch.setattr(
        "app.domains.auditor.vendor_assessment_service.presign_get",
        _fake_presign,
    )

    obj_key = f"dev/vendor_assessment_evidence/2026/07/31/{aid}/pentest.pdf"
    reg = await client.post(
        f"/api/v1/auditor/vendor-assessments/{aid}/attachments",
        headers=auditor_headers,
        json={"object_key": obj_key, "filename": "pentest.pdf",
              "content_type": "application/pdf", "size_bytes": 1024},
    )
    assert reg.status_code == 200, reg.text
    att_id = reg.json()["data"]["id"]
    attachment = await db_session.get(
        AuditorVendorAssessmentAttachment, uuid.UUID(att_id),
    )
    assert attachment is not None
    assert attachment.content_type == "image/png"
    assert claimed_resource_ids == [att_id]

    resp = await client.get(
        f"/api/v1/auditor/vendor-assessments/attachments/{att_id}/download",
        headers=auditor_headers,
    )
    assert resp.status_code == 200, resp.text

    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action
            == "auditor.vendor_assessment.attachment.download",
        )
    )).scalars().all()
    assert len(rows) == 1
    assert rows[0].detail["attachment_id"] == att_id
    assert rows[0].resource_id == aid


# ── 2026-07-31 lifecycle-gate batch ──────────────────────────────────


async def _completed_with_finding(db, *, key, vendor):
    """An assessment completed with one promoted open finding."""
    t = await _seed_template(db, key=key)
    await db.commit()
    a = await svc.create_assessment(
        db, payload={"template_id": t.id, "vendor_name": vendor,
                     "tier": "high"},
        actor_id=None,
    )
    await db.commit()
    ans = (await db.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
            AuditorVendorAssessmentAnswer.question_code == "Q1",
        )
    )).scalar_one()
    await svc.review_answer(db, ans.id, review_flag="finding",
                            reviewer_note="gap")
    await db.commit()
    await svc.complete_assessment(db, a.id, summary=None,
                                  reviewer_person_id=None)
    await db.commit()
    return a, ans


async def _findings_for(db, assessment_id):
    return (await db.execute(
        select(AuditorFinding).where(
            AuditorFinding.finding_type == f"assessment:{assessment_id}",
        )
    )).scalars().all()


@pytest.mark.asyncio
async def test_unflagging_on_a_completed_assessment_closes_the_finding(
    db_session,
):
    """#2165 — a flag change after completion must reconcile right away.

    ``complete_assessment`` is the only caller of ``_sync_findings``, so
    un-flagging without re-completing used to leave the finding open on
    the worklist for ever, still linked from ``answer.finding_id``.
    """
    a, ans = await _completed_with_finding(
        db_session, key="t_unflag", vendor="UnflagCo")
    assert [f.status for f in await _findings_for(db_session, a.id)] == ["open"]

    await svc.review_answer(db_session, ans.id, review_flag="ok",
                            reviewer_note=None)
    await db_session.commit()

    assert [f.status for f in await _findings_for(db_session, a.id)] == [
        "resolved"
    ]


@pytest.mark.asyncio
async def test_review_answer_rejected_on_archived_assessment(db_session):
    """#2165 — archived/cancelled have no reconcile path out."""
    a, ans = await _completed_with_finding(
        db_session, key="t_arch_rev", vendor="ArchCo")
    await svc.set_status(db_session, a.id, status="archived")
    await db_session.commit()

    with pytest.raises(Exception, match="archived"):
        await svc.review_answer(db_session, ans.id, review_flag="ok",
                                reviewer_note=None)


@pytest.mark.asyncio
async def test_cancelling_closes_the_open_findings(db_session):
    """#2166 — cancel/archive must not strand findings on the worklist.

    ``complete_assessment`` refuses cancelled/archived, so the only
    reconcile entry point is closed to them: whatever is open at cancel
    time stays open for ever, pointing at an assessment nobody will
    finish.
    """
    a, ans = await _completed_with_finding(
        db_session, key="t_cancel", vendor="CancelCo")
    assert [f.status for f in await _findings_for(db_session, a.id)] == ["open"]

    await svc.set_status(db_session, a.id, status="cancelled")
    await db_session.commit()

    assert [f.status for f in await _findings_for(db_session, a.id)] == [
        "resolved"
    ]
    await db_session.refresh(ans)
    assert ans.finding_id is None


@pytest.mark.asyncio
async def test_back_to_draft_clears_the_completion_verdict(db_session):
    """#2166 — a draft carries no completion time and no risk grade."""
    a, _ans = await _completed_with_finding(
        db_session, key="t_draft", vendor="DraftCo")
    assert a.completed_at is not None
    assert a.risk_rating is not None

    await svc.set_status(db_session, a.id, status="draft")
    await db_session.commit()
    await db_session.refresh(a)

    assert a.status == "draft"
    assert a.completed_at is None
    assert a.score_pct is None
    assert a.risk_rating is None
    assert a.next_review_date is None


@pytest.mark.asyncio
async def test_import_with_zero_matches_leaves_status_alone(db_session):
    """#2167 — the wrong workbook must not clear the overdue alarm.

    ``is_response_overdue`` only fires while status == 'sent', so
    flipping to 'answered' on a zero-match import silently retired the
    alarm on an assessment the vendor had not answered at all.
    """
    from openpyxl import Workbook

    t = await _seed_template(db_session, key="t_zero_match")
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "ZeroCo",
                             "tier": "high"},
        actor_id=None,
    )
    a.status = "sent"
    await db_session.commit()

    # A structurally valid workbook whose codes belong to another
    # template — exactly the "uploaded the wrong file" case.
    wb = Workbook()
    ws = wb.active
    ws.append(["编号/Code", "回答/Answer", "备注/Comment"])
    ws.append(["ZZ-1", "yes", ""])
    ws.append(["ZZ-2", "no", ""])
    buf = io.BytesIO()
    wb.save(buf)

    with pytest.raises(Exception, match="no answers matched"):
        await svc.import_answers(db_session, a.id, payload=buf.getvalue())

    # The guard fires before any status mutation, so the row in this
    # session is untouched — no rollback needed to observe it.
    assert a.status == "sent", "the overdue alarm was silently cleared"
    assert a.answered_at is None


@pytest.mark.asyncio
async def test_attachment_rejects_an_answer_from_another_assessment(
    db_session, monkeypatch,
):
    """#2168 — ``answer_id`` is a soft ref; nothing else checks it."""
    t = await _seed_template(db_session, key="t_xref")
    await db_session.commit()
    a = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "A",
                             "tier": "high"}, actor_id=None)
    b = await svc.create_assessment(
        db_session, payload={"template_id": t.id, "vendor_name": "B",
                             "tier": "high"}, actor_id=None)
    await db_session.commit()
    b_answer = (await db_session.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == b.id,
        ).limit(1)
    )).scalars().first()

    async def _fake_stat(_db, *, object_key, **_kw):
        return {"size_bytes": 10, "content_type": "application/pdf"}

    monkeypatch.setattr(
        "app.domains.auditor.vendor_assessment_service.stat_object",
        _fake_stat,
    )

    with pytest.raises(Exception, match="does not belong"):
        await svc.register_attachment(
            db_session, assessment_id=a.id,
            object_key=f"dev/vendor_assessment_evidence/2026/07/31/{a.id}/x.pdf",
            filename="x.pdf", content_type="application/pdf",
            uploaded_by_person_id=None,
            answer_id=b_answer.id,
        )


@pytest.mark.asyncio
async def test_reopen_then_cancel_still_closes_the_findings(db_session):
    """#2232 — the other way into #2166's stranded-findings state.

    ``reopen_assessment`` sets ``status='answered'`` and clears
    ``completed_at`` while deliberately leaving the promoted findings
    open. The close gate was ``status == 'completed' or completed_at is
    not None``, so cancelling a REOPENED assessment (an auditor deciding
    to drop the vendor mid-review) computed ``was_assessed=False``, took
    no action, and stranded them for ever: ``complete_assessment`` and
    ``review_answer`` both refuse a cancelled assessment, no cron
    reconciles the ``assessment:{id}`` scope, and even sending it back to
    draft skipped the close on the same predicate.
    """
    a, _ans = await _completed_with_finding(
        db_session, key="t_reopen_cancel", vendor="ReopenCancelCo")
    assert [f.status for f in await _findings_for(db_session, a.id)] == ["open"]

    await svc.reopen_assessment(db_session, a.id)
    await db_session.commit()
    await db_session.refresh(a)
    assert a.status == "answered" and a.completed_at is None
    # Reopen leaves them open by design — that is the state cancel meets.
    assert [f.status for f in await _findings_for(db_session, a.id)] == ["open"]

    await svc.set_status(db_session, a.id, status="cancelled")
    await db_session.commit()
    assert [f.status for f in await _findings_for(db_session, a.id)] == [
        "resolved"
    ]


@pytest.mark.asyncio
async def test_reopen_then_draft_also_closes_the_findings(db_session):
    """Same predicate, the other terminal status the endpoint accepts."""
    a, _ans = await _completed_with_finding(
        db_session, key="t_reopen_draft", vendor="ReopenDraftCo")
    await svc.reopen_assessment(db_session, a.id)
    await db_session.commit()

    await svc.set_status(db_session, a.id, status="draft")
    await db_session.commit()
    assert [f.status for f in await _findings_for(db_session, a.id)] == [
        "resolved"
    ]


@pytest.mark.asyncio
async def test_status_change_with_no_findings_is_a_no_op(db_session):
    """Closing unconditionally is only safe because the close is an
    empty-spec reconcile — pin that a never-assessed draft still works.
    """
    t = await _seed_template(db_session, key="t_noop")
    await db_session.commit()
    a = await svc.create_assessment(
        db_session,
        payload={"template_id": t.id, "vendor_name": "NoopCo",
                 "tier": "low"},
        actor_id=None,
    )
    await db_session.commit()

    await svc.set_status(db_session, a.id, status="cancelled")
    await db_session.commit()
    await db_session.refresh(a)
    assert a.status == "cancelled"
    assert await _findings_for(db_session, a.id) == []


@pytest.mark.asyncio
async def test_blank_template_upload_says_the_cells_were_empty(db_session):
    """#2247 — the ``updated == 0`` guard is right, its wording was not.

    A workbook whose codes MATCH but whose answer cells are all blank
    (the vendor returned the template untouched) reported "0 unknown
    code(s)", sending the operator after a code mismatch that does not
    exist.
    """
    t = await _seed_template(db_session, key="t_blank")
    await db_session.commit()
    a = await svc.create_assessment(
        db_session,
        payload={"template_id": t.id, "vendor_name": "BlankCo",
                 "tier": "low"},
        actor_id=None,
    )
    await db_session.commit()

    answers = list((await db_session.execute(
        select(AuditorVendorAssessmentAnswer).where(
            AuditorVendorAssessmentAnswer.assessment_id == a.id,
        ).order_by(AuditorVendorAssessmentAnswer.sort_order)
    )).scalars().all())
    # The blank workbook, returned exactly as generated — right codes,
    # nothing filled in.
    buf = svc.build_blank_questionnaire(a, answers)
    with pytest.raises(Exception, match="every matched cell was empty"):
        await svc.import_answers(db_session, a.id, payload=buf.getvalue())