bid-doctor/server/tests/test_project_info.py

"""项目基本信息修改和分包（FR-M2-03、04，project-info-plan）：人工值优先、撤销恢复 AI 值、校验、只读、留痕；
分包增删改、默认包建回、「不分包」、删包时条目归通用、选包；清单按所投的包筛选；检查记下当时的项目依据（PI5）。"""

from __future__ import annotations

from typing import Any
from uuid import UUID

from sqlalchemy import select

from tests.conftest import User, project_with_tender, run_parse
from tests.test_check_flow import run_check, upload_bid


async def audit_actions(u: User) -> list[str]:
    from bid_doctor.audit.models import AuditEvent
    from bid_doctor.core.db import TenantContext, tenant_tx

    ctx = TenantContext(tenant_id=UUID(u.tenant_id), user_id=UUID(int=0))
    async with tenant_tx(ctx) as s:
        return list((await s.execute(select(AuditEvent.action).order_by(AuditEvent.created_at))).scalars())


def field(detail: dict[str, Any], key: str) -> dict[str, Any] | None:
    return next((f for f in detail["fields"] if f["key"] == key), None)


async def test_fields_human_value_reset_and_validation(user: User) -> None:
    project_id, tender_id = await project_with_tender(user)
    assert (await run_parse(user, tender_id))["status"] == "succeeded"
    before = (await user.get(f"/projects/{project_id}")).json()

    r = await user.put(f"/projects/{project_id}/fields/deadline_at", json={"value": "2026-12-01T09:30:00+08:00"})
    assert r.status_code == 200, r.text
    f = r.json()
    assert f["source"] == "human" and f["value"] == "2026-12-01T09:30:00+08:00" and f["updated_by"]["name"] == "张工"
    detail = (await user.get(f"/projects/{project_id}")).json()
    assert detail["deadline_at"].startswith("2026-12-01T01:30")  # project 表上冗余的列跟着改
    (bid,) = [d for d in detail["key_dates"] if d["label"] == "投标截止"]
    assert bid["edited"] and bid["item_id"] is None and bid["at"] == "2026-12-01T09:30:00+08:00"

    r = await user.put(f"/projects/{project_id}/fields/ceiling_total", json={"value": "286万"})
    assert r.json()["value"] == "2860000.00"  # PI7：按元存
    r = await user.put(f"/projects/{project_id}/fields/accept_consortium", json={"value": False})
    assert r.json() == {**r.json(), "value": False, "source": "human"}
    r = await user.put(f"/projects/{project_id}/fields/name", json={"value": "  某市政务云扩容项目 "})
    assert (await user.get(f"/projects/{project_id}")).json()["name"] == "某市政务云扩容项目"

    for key, value in (
        ("name", " "),
        ("ceiling_total", "-1"),
        ("ceiling_total", "两百万"),
        ("deadline_at", "明天"),
        ("legal_regime", "civil"),
        ("accept_imported", "是"),
    ):
        r = await user.put(f"/projects/{project_id}/fields/{key}", json={"value": value})
        assert r.status_code == 422, (key, value, r.text)

    # 撤销人工值：有 AI 值的恢复 AI 值，冗余的列也恢复；名称没有 AI 值时保持现在的名称
    await user.put(f"/projects/{project_id}/fields/legal_regime", json={"value": "gov_procurement"})
    assert (await user.get(f"/projects/{project_id}")).json()["legal_regime"] == "gov_procurement"
    r = await user.put(f"/projects/{project_id}/fields/legal_regime", json={"value": None})
    assert r.json()["source"] == "ai" and r.json()["value"] == "tendering"
    assert (await user.get(f"/projects/{project_id}")).json()["legal_regime"] == "tendering"
    r = await user.put(f"/projects/{project_id}/fields/name", json={"value": None})
    assert r.json()["source"] == "none"
    assert (await user.get(f"/projects/{project_id}")).json()["name"] == "某市政务云扩容项目"
    r = await user.put(f"/projects/{project_id}/fields/deadline_at", json={"value": None})
    after = (await user.get(f"/projects/{project_id}")).json()
    assert after["deadline_at"] == before["deadline_at"]
    assert not any(d.get("edited") for d in after["key_dates"])
    assert (await audit_actions(user)).count("project_field.set") == 8

    # 已递交的项目只读
    assert (await user.post(f"/projects/{project_id}/transitions", json={"action": "submit"})).status_code == 200
    r = await user.put(f"/projects/{project_id}/fields/code", json={"value": "X-1"})
    assert r.status_code == 409 and r.json()["code"] == "project_read_only"


async def test_lots_default_clear_and_checklist_scope(user: User) -> None:
    project_id, tender_id = await project_with_tender(user)
    await run_parse(user, tender_id)
    (default,) = (await user.get(f"/projects/{project_id}/lots")).json()["items"]
    assert default["is_default"]

    r = await user.post(
        f"/projects/{project_id}/lots", json={"lot_no": "1", "name": "核心网络", "ceiling_amount": "120万"}
    )
    assert r.status_code == 201, r.text
    lot1 = r.json()
    assert lot1["ceiling_amount"] == "1200000.00" and lot1["origin"] == "human"
    lot2 = (await user.post(f"/projects/{project_id}/lots", json={"lot_no": "2", "name": "机房配套"})).json()
    lots = (await user.get(f"/projects/{project_id}/lots")).json()["items"]
    assert [x["lot_no"] for x in lots] == ["1", "2"]  # 手工分包：默认包不要了
    assert (await user.post(f"/projects/{project_id}/lots", json={"lot_no": "1"})).status_code == 422
    assert (await user.patch(f"/lots/{lot2['id']}", json={"lot_no": "1"})).status_code == 422
    r = await user.patch(f"/lots/{lot2['id']}", json={"budget_amount": "80,000"})
    assert r.json()["budget_amount"] == "80000.00"

    # 条目归属：包 2 的条目不在所投的包里时不列出
    created = []
    for title, lot in (("包1：交换机须支持硬件BFD", lot1["id"]), ("包2：精密空调制冷量不低于40kW", lot2["id"])):
        r = await user.post(
            f"/projects/{project_id}/checklist/items",
            json={"categories": ["tech_param"], "title": title, "lot_id": lot},
        )
        assert r.status_code == 201, r.text
        created.append(r.json()["id"])
    everything = (await user.get(f"/projects/{project_id}/checklist/items", params={"lot_scope": "all"})).json()
    assert (await user.put(f"/projects/{project_id}/bidding-lots", json={"lot_ids": []})).status_code == 422
    assert (
        await user.put(f"/projects/{project_id}/bidding-lots", json={"lot_ids": [default["id"]]})
    ).status_code == 422
    r = await user.put(f"/projects/{project_id}/bidding-lots", json={"lot_ids": [lot1["id"]]})
    assert [x["is_bidding"] for x in r.json()["items"]] == [True, False]
    shown = (await user.get(f"/projects/{project_id}/checklist/items")).json()
    assert shown["hidden_count"] == 1 and len(shown["items"]) == len(everything["items"]) - 1
    assert created[1] not in {i["id"] for i in shown["items"]}
    assert (await user.get(f"/projects/{project_id}")).json()["bidding_lots"] == ["1"]

    # 删包：原属这个包的条目归为通用（PI3）
    assert (await user.delete(f"/lots/{lot2['id']}")).status_code == 204
    item = (await user.get(f"/checklist-items/{created[1]}")).json()
    assert item["lot_id"] is None
    # 删掉最后一个包：回到不分包（PI2），条目都列出
    assert (await user.delete(f"/lots/{lot1['id']}")).status_code == 204
    (back,) = (await user.get(f"/projects/{project_id}/lots")).json()["items"]
    assert back["is_default"]
    assert (await user.get(f"/projects/{project_id}/checklist/items")).json()["hidden_count"] == 0

    # 一键「不分包」
    for no in ("A", "B", "C"):
        await user.post(f"/projects/{project_id}/lots", json={"lot_no": no})
    r = await user.post(f"/projects/{project_id}/lots/clear")
    assert [x["is_default"] for x in r.json()["items"]] == [True]
    actions = await audit_actions(user)
    assert {"lot.create", "lot.update", "lot.delete", "lot.clear", "project.bidding_lots"} <= set(actions)


async def test_check_records_basis_and_flags_changes(user: User) -> None:
    project_id, tender_id = await project_with_tender(user)
    await run_parse(user, tender_id)
    await user.put(f"/projects/{project_id}/fields/deadline_at", json={"value": "2026-12-01T09:30:00+08:00"})
    bid = await upload_bid(user, project_id, "bid-mini-v1.pdf")
    run = await run_check(user, project_id, [bid["id"]])
    basis = run["coverage"]["basis"]
    assert basis["deadline_at"] == {"value": "2026-12-01T01:30:00+00:00", "source": "human"}
    assert basis["lots"] == [] and not run["basis_changed"]

    await user.put(f"/projects/{project_id}/fields/deadline_at", json={"value": "2026-12-08T09:30:00+08:00"})
    assert (await user.get(f"/check-runs/{run['id']}")).json()["basis_changed"]

    # 报告按检查时的截止时间写
    from bid_doctor.report.service import _deadline  # pyright: ignore[reportPrivateUsage]

    class Run:
        coverage = run["coverage"]

    assert _deadline(Run()).isoformat() == "2026-12-01T01:30:00+00:00"  # type: ignore[arg-type]