bid-doctor/server/tests/test_report_flow.py

"""检查报告（查标第 5 步，FR-M3-09）：导出 → 后台生成 PDF → 列表 → 下载；致命问题未处理也能导出（红线 4）；
模板文字不出现「通过 / 合格 / 保证」（设计规范 §7.17）。需要 Gotenberg。"""

from __future__ import annotations

import hashlib
import io
from datetime import UTC, datetime
from urllib.parse import urlsplit
from uuid import UUID

import pypdfium2 as pdfium

from bid_doctor.report import template
from tests.conftest import User
from tests.test_check_flow import checked_project

FORBIDDEN = ("通过", "合格", "保证")


async def export(u: User, run_id: str, **body: bool) -> dict[str, object]:
    from bid_doctor.report import runner

    r = await u.post(f"/check-runs/{run_id}/reports", json=body or None)
    assert r.status_code == 202, r.text
    job = r.json()
    assert job["status"] == "queued" and job["report_id"] is None
    await runner.run(UUID(u.tenant_id), UUID(job["id"]))
    return (await u.get(f"/report-jobs/{job['id']}")).json()


def pdf_text(data: bytes) -> str:
    pdf = pdfium.PdfDocument(io.BytesIO(data))
    return "".join(pdf[i].get_textpage().get_text_range() for i in range(len(pdf)))


async def test_export_report_and_download(user: User) -> None:
    project_id, _, run = await checked_project(user)
    assert run["counts"]["open_fatal"] == 1

    job = await export(user, run["id"])
    assert job["status"] == "succeeded" and job["report_id"], job
    (rep,) = (await user.get(f"/projects/{project_id}/reports")).json()["items"]
    assert rep["id"] == job["report_id"] and rep["run_no"] == 1 and rep["page_count"] >= 2
    assert rep["open_fatal_at_export"] == 1 and rep["template_version"] == template.TEMPLATE_VERSION

    link = (await user.get(f"/reports/{rep['id']}/file")).json()
    parts = urlsplit(link["url"])
    got = await user.client.get(parts.path.removeprefix("/api/v1") + "?" + parts.query)
    assert got.status_code == 200 and got.headers["content-type"] == "application/pdf"
    assert hashlib.sha256(got.content).hexdigest() == rep["sha256"] == link["sha256"]
    text = pdf_text(got.content)
    for want in ("检查覆盖范围", "免责声明", "问题明细", template.FOOTER, "交换机支持硬件BFD"):
        assert want in text.replace("\r", "").replace("\n", ""), want

    # 重复导出就是新的一份，旧的还在（随项目归档）
    await export(user, run["id"], citations=False, manual_checks=False)
    assert len((await user.get(f"/projects/{project_id}/reports")).json()["items"]) == 2

    notes = (await user.get("/notifications")).json()["items"]
    assert notes[0]["kind"] == "report_succeeded"


async def test_cannot_export_unfinished_run_or_other_tenant(user: User, app_client: object) -> None:
    from sqlalchemy import select

    from bid_doctor.check.models import CheckRun
    from bid_doctor.core.db import TenantContext, tenant_tx

    project_id, _, run = await checked_project(user)
    ctx = TenantContext(tenant_id=UUID(user.tenant_id), user_id=UUID(str(user.me["user"]["id"])))
    async with tenant_tx(ctx) as s:
        r = (await s.execute(select(CheckRun).where(CheckRun.id == UUID(run["id"])))).scalar_one()
        r.status = "failed"
    resp = await user.post(f"/check-runs/{run['id']}/reports")
    assert resp.status_code == 409 and resp.json()["code"] == "run_not_succeeded"
    del project_id, app_client


def test_template_text_has_no_conclusion_words() -> None:
    """模板自己的文字（标题、页脚、免责声明、表头、提示）不出现「通过 / 合格 / 保证」；原文引用不在此列。"""
    data = {
        "project": {
            "name": "某市数据中心扩容项目",
            "code": "ZC-1",
            "purchaser": "某市大数据局",
            "legal_regime": "gov_procurement",
            "procurement_method": "公开招标",
            "deadline_at": datetime(2026, 12, 1, 9, 30, tzinfo=UTC),
            "lots": [],
        },
        "run": {
            "run_no": 2,
            "created_at": datetime.now(UTC),
            "finished_at": datetime.now(UTC),
            "rule_set_version": "r",
            "checklist_version_no": 1,
            "checklist_confirmed": False,
            "coverage": {"checked": [], "not_checked": []},
            "documents": [{"label": "技术分册", "filename": "a.pdf"}],
        },
        "summary": {
            s: {"total": 0, "pending": 0, "confirmed_ok": 0, "modified": 0, "ignored": 0}
            for s in ("fatal", "important", "hint")
        },
        "resolved_this_run": 0,
        "findings": [],
        "manual": [],
        "options": {},
        "exported_by": "张工",
        "exported_at": datetime.now(UTC),
    }
    html = template.page(data) + template.header() + template.footer()
    for w in FORBIDDEN:
        assert w not in html, w


async def test_project_submit_archive_and_undo(user: User) -> None:
    """项目状态流（FR-M6-02）：致命问题未处理也能递交（红线 4），递交后只读、可撤销；
    归档后撤销回到归档前的状态；都留痕。"""
    project_id, _, run = await checked_project(user)
    assert run["counts"]["open_fatal"] == 1

    r = await user.post(f"/projects/{project_id}/transitions", json={"action": "submit"})
    assert r.status_code == 200 and r.json()["status"] == "submitted", r.text
    # 只读：不能再发起检查、勾选需人工核对
    blocked = await user.post(
        f"/projects/{project_id}/checklist/items", json={"categories": ["schedule"], "title": "x"}
    )
    assert blocked.status_code == 409 and blocked.json()["code"] == "project_read_only"
    # 报告照样能导出（随项目归档）
    assert (await export(user, run["id"]))["status"] == "succeeded"

    assert (await user.post(f"/projects/{project_id}/transitions", json={"action": "submit"})).status_code == 409
    r = await user.post(f"/projects/{project_id}/transitions", json={"action": "archive"})
    assert r.json()["status"] == "archived"
    r = await user.post(f"/projects/{project_id}/transitions", json={"action": "unarchive"})
    assert r.json()["status"] == "submitted"  # 回到归档前
    r = await user.post(f"/projects/{project_id}/transitions", json={"action": "unsubmit"})
    assert r.json()["status"] == "checked"