f98fd51128
- 合并时待确认条目不再借正式条目进清单，没通过的类别记在 pending_categories - 结构识别：章（按章跳过不相干的部分）、表后「上述……否则无效」说明、★ 节标题 - INVALID 补写法；输出用到上限 92% 按截断处理，拆到底时采用已有输出 - 格式范本正文、整表概括句不收；项目信息多看评标方法所在分块 - WPS 旧版 DOC 直接修正字体表字符集；表格按单元格切句（s3-v4） - 口径：★ 节继承实质性、★ 定义为计分项时降为重要、「必须/应」声明记 keyword 规则、 政府采购资格条件按 87 号令第六十三条带无效类别 - 提示词 substantive v5，qualification / schedule / tech_param v4 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
206 lines
10 KiB
Python
206 lines
10 KiB
Python
"""最小链路：上传 → 解析 → 清单；以及额度的预扣、结算、退回（红线 9）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import httpx
|
||
|
||
from tests.conftest import FIXTURES, User, balance, project_with_tender, register, run_parse, upload
|
||
|
||
|
||
async def test_upload_parse_and_checklist(user: User) -> None:
|
||
project_id, doc_id = await project_with_tender(user)
|
||
assert (await user.get(f"/projects/{project_id}")).json()["next_action"] == "wait_parse"
|
||
|
||
job = await run_parse(user, doc_id)
|
||
assert job["status"] == "succeeded", job
|
||
assert job["result"]["items_added"] == 3
|
||
|
||
items = (await user.get(f"/projects/{project_id}/checklist/items")).json()["items"]
|
||
formal = [i for i in items if i["zone"] == "formal"]
|
||
assert len(formal) == 3
|
||
for it in formal:
|
||
assert it["importance"] == "substantive" and "invalidity" in it["categories"]
|
||
(c,) = it["citations"]
|
||
assert c["page_no"] == 1 and c["quads"] and c["quote"].startswith("★")
|
||
assert set(c["sentence_keys"][0].split(":")[:1]) == {doc_id}
|
||
assert any("32GB" in i["citations"][0]["quote"] for i in formal)
|
||
|
||
checklist = (await user.get(f"/projects/{project_id}/checklist")).json()
|
||
assert [r["marker"] for r in checklist["marking_rules"]] == ["star"]
|
||
|
||
project = (await user.get(f"/projects/{project_id}")).json()
|
||
assert project["status"] == "drafting" and project["next_action"] == "confirm_checklist"
|
||
|
||
doc = (await user.get(f"/documents/{doc_id}")).json()
|
||
assert doc["status"] == "ready" and doc["page_count"] == 1 and doc["normalized_sha256"]
|
||
|
||
sentences = (await user.get(f"/documents/{doc_id}/pages/1/sentences")).json()["items"]
|
||
assert any(s["text"].startswith("★交换机") for s in sentences)
|
||
|
||
notes = (await user.get("/notifications")).json()
|
||
assert notes["unread_count"] == 1 and notes["items"][0]["kind"] == "parse_succeeded"
|
||
|
||
|
||
async def test_quota_hold_then_settle(user: User) -> None:
|
||
_, doc_id = await project_with_tender(user)
|
||
assert await balance(user) == 3
|
||
await run_parse(user, doc_id)
|
||
assert await balance(user) == 2
|
||
accounts = (await user.get("/quota")).json()["accounts"]
|
||
assert accounts[0]["period_used"] == 1 and accounts[0]["held"] == 0
|
||
|
||
|
||
async def test_failed_parse_refunds_quota(user: User) -> None:
|
||
"""没有文字层的 PDF（扫描件）：现在报「需要 OCR」，额度退回，净扣点为 0（FR-M6-03）。"""
|
||
_, doc_id = await project_with_tender(user, "blank.pdf")
|
||
job = await run_parse(user, doc_id)
|
||
assert job["status"] == "failed"
|
||
assert job["error"] == {"code": "ocr_required", "message": job["error"]["message"], "refunded": True}
|
||
assert await balance(user) == 3
|
||
|
||
from sqlalchemy import text
|
||
|
||
from bid_doctor.core.db import TenantContext, tenant_tx
|
||
|
||
ctx = TenantContext(tenant_id=user.me["tenant"]["id"], user_id=user.me["user"]["id"])
|
||
async with tenant_tx(ctx) as s:
|
||
kinds = (
|
||
await s.execute(text("SELECT kind, amount FROM quota_entry WHERE kind <> 'grant' ORDER BY created_at"))
|
||
).all()
|
||
assert [tuple(k) for k in kinds] == [("hold", -1), ("release", 1)]
|
||
|
||
|
||
async def test_retry_after_failure_holds_again(user: User) -> None:
|
||
_, doc_id = await project_with_tender(user, "blank.pdf")
|
||
first = await run_parse(user, doc_id)
|
||
second = await run_parse(user, doc_id)
|
||
assert (first["attempt_no"], second["attempt_no"]) == (1, 2)
|
||
assert await balance(user) == 3
|
||
|
||
|
||
async def test_duplicate_parse_is_rejected(user: User) -> None:
|
||
_, doc_id = await project_with_tender(user)
|
||
assert (await user.post(f"/documents/{doc_id}/parse-jobs", json={"expected_charge": 1})).status_code == 201
|
||
r = await user.post(f"/documents/{doc_id}/parse-jobs", json={"expected_charge": 1})
|
||
assert r.status_code == 409 and r.json()["code"] == "job_already_running" and r.json()["job_id"]
|
||
assert await balance(user) == 2 # 只预扣了一次
|
||
|
||
|
||
async def test_billing_mismatch_is_rejected_without_charge(user: User) -> None:
|
||
_, doc_id = await project_with_tender(user)
|
||
r = await user.post(f"/documents/{doc_id}/parse-jobs", json={"expected_charge": 0})
|
||
assert r.status_code == 409 and r.json()["code"] == "billing_changed"
|
||
assert await balance(user) == 3
|
||
|
||
|
||
async def test_quota_exhausted(app_client: httpx.AsyncClient) -> None:
|
||
u = await register(app_client)
|
||
_, doc_id = await project_with_tender(u)
|
||
for _ in range(3):
|
||
await run_parse(u, doc_id)
|
||
r = await u.post(f"/documents/{doc_id}/parse-jobs", json={"expected_charge": 1})
|
||
assert r.status_code == 409 and r.json()["code"] == "quota_exhausted"
|
||
assert await balance(u) == 0
|
||
|
||
|
||
async def test_reparse_keeps_item_ids(user: User) -> None:
|
||
"""重新解析时清单项编号不变（架构 §3.8）。"""
|
||
project_id, doc_id = await project_with_tender(user)
|
||
await run_parse(user, doc_id)
|
||
before = {i["id"] for i in (await user.get(f"/projects/{project_id}/checklist/items")).json()["items"]}
|
||
job = await run_parse(user, doc_id)
|
||
assert job["result"]["items_updated"] == 3 and job["result"]["items_added"] == 0
|
||
after = {i["id"] for i in (await user.get(f"/projects/{project_id}/checklist/items")).json()["items"]}
|
||
assert before == after
|
||
|
||
|
||
async def test_free_plan_allows_one_project(user: User) -> None:
|
||
assert (await user.post("/projects", json={})).status_code == 201
|
||
r = await user.post("/projects", json={})
|
||
assert r.status_code == 409 and r.json()["code"] == "plan_limit_exceeded"
|
||
|
||
|
||
async def test_upload_rejects_unsupported_and_corrupt_files(user: User) -> None:
|
||
r = await upload(user, "a.txt", "这不是招标文件".encode())
|
||
assert r.status_code == 422 and r.json()["code"] == "file_format_unsupported"
|
||
r = await upload(user, "a.pdf", b"%PDF-1.7\nnot really a pdf")
|
||
assert r.status_code == 422 and r.json()["code"] == "file_corrupted"
|
||
r = await upload(user, "看起来是.doc", (FIXTURES / "tender-mini.pdf").read_bytes())
|
||
assert r.status_code == 200
|
||
assert r.json()["detected_format"] == "pdf" and r.json()["extension_mismatch"] is True
|
||
|
||
|
||
async def test_upload_size_limit(user: User) -> None:
|
||
r = await user.post(
|
||
"/uploads", json={"filename": "big.pdf", "size_bytes": 51 * 1024 * 1024, "purpose": "project_document"}
|
||
)
|
||
assert r.status_code == 409 and r.json()["code"] == "plan_limit_exceeded"
|
||
|
||
|
||
async def test_signed_download(user: User, app_client: httpx.AsyncClient) -> None:
|
||
_, doc_id = await project_with_tender(user)
|
||
await run_parse(user, doc_id)
|
||
url = (await user.get(f"/documents/{doc_id}/file", params={"variant": "normalized"})).json()["url"]
|
||
path = url.split("/api/v1", 1)[1]
|
||
r = await app_client.get(path)
|
||
assert r.status_code == 200 and r.content.startswith(b"%PDF")
|
||
tampered = path.replace("sig=", "sig=x")
|
||
assert (await app_client.get(tampered)).status_code == 404
|
||
|
||
|
||
async def test_objects_are_encrypted_at_rest(user: User) -> None:
|
||
import os
|
||
from pathlib import Path
|
||
|
||
_, doc_id = await project_with_tender(user)
|
||
await run_parse(user, doc_id)
|
||
root = Path(os.environ["BD_STORAGE_DIR"]) / "t" / user.tenant_id
|
||
files = [p for p in root.rglob("*") if p.is_file()]
|
||
assert files and not any(p.read_bytes().startswith(b"%PDF") for p in files)
|
||
plain = (FIXTURES / "tender-mini.pdf").read_bytes()[:64]
|
||
assert not any(plain in p.read_bytes() for p in files)
|
||
|
||
|
||
async def test_all_categories_and_project_info(user: User) -> None:
|
||
"""虚构的两页样例：项目基本信息、截止时间、资格、评分、技术参数、上级标题下的无效情形、* 号兜底、页脚不进清单。"""
|
||
project_id, doc_id = await project_with_tender(user, "tender-full.pdf")
|
||
job = await run_parse(user, doc_id)
|
||
assert job["status"] == "succeeded", job
|
||
|
||
project = (await user.get(f"/projects/{project_id}")).json()
|
||
assert project["name"] == "某市智慧政务服务平台建设项目" # 建项目时没填名称：按招标文件自动填
|
||
assert project["code"] == "TEST-2026-001"
|
||
assert project["purchaser"] == "某市行政审批局"
|
||
assert project["legal_regime"] == "gov_procurement"
|
||
assert project["deadline_at"].startswith("2026-12-01T01:30") # 北京时间 9:30
|
||
fields = {f["key"]: f for f in project["fields"]}
|
||
assert fields["budget_total"]["value"] == "450000.00"
|
||
assert fields["budget_total"]["citation"]["page_no"] == 1
|
||
|
||
items = (await user.get(f"/projects/{project_id}/checklist/items")).json()["items"]
|
||
by_text = {i["requirement_text"]: i for i in items}
|
||
|
||
def item(prefix: str) -> dict[str, object]:
|
||
return next(v for k, v in by_text.items() if k and k.startswith(prefix))
|
||
|
||
for clause in ("25.3.1", "25.3.2"): # 本身没写「无效」，靠上级标题认定
|
||
it = item(clause)
|
||
assert it["zone"] == "formal" and "invalidity" in it["categories"], it
|
||
# 口径 Q4：政府采购项目不具备资格要求的投标无效（87 号令第六十三条），资格条件带「无效」类别，依据是法规
|
||
qual = item("资格要求")
|
||
assert qual["categories"] == ["qualification", "invalidity"]
|
||
assert (qual["importance"], qual["importance_basis"]) == ("substantive", "statute")
|
||
assert item("技术方案")["spec"]["scoring"]["points"] == "20"
|
||
assert item("系统支持")["spec"]["tech_param"] == {"param": "系统支持", "op": ">=", "value": "200", "unit": "个"}
|
||
deadline = item("响应文件递交截止时间")
|
||
assert deadline["spec"]["schedule"]["date_kind"] == "bid_deadline"
|
||
star = item("*供应商须提供原厂授权书") # 假模型不认 * 号：规则兜底进待确认
|
||
assert (star["zone"], star["pending_reason"], star["importance"]) == ("pending", "marker_uncovered", "substantive")
|
||
assert not any(k and ("—1—" in k or "—2—" in k) for k in by_text) # 页码不混进条款
|
||
assert not any(k and "斜体且有下划线" in k for k in by_text) # 规则说明本身不是条款
|
||
|
||
checklist = (await user.get(f"/projects/{project_id}/checklist")).json()
|
||
assert {r["marker"] for r in checklist["marking_rules"]} == {"italic_underline", "asterisk"}
|
||
sentences = (await user.get(f"/documents/{doc_id}/pages/1/sentences")).json()["items"]
|
||
assert [s["location"] for s in sentences if s["text"] == "—1—"] == ["header_footer"]
|