"""NovaPay 续约复审（**整改后**）数据房间的纯业务层查询测试。 > 整改前（"故意埋入 5 条风险线索"）快照位于 > ``data-room/NOVAPAY.original-2026-08-19/_fixtures-snapshot/``。 > 当前 ``fixtures/`` 是整改后版本，对应 `policies/vendor-renewal.md` 的 > ``renewal`` 档 —— 数据应当让 Agent 顺利得出可提交审批的结论。 覆盖以下场景（无需调用模型）： 1. 供应商主数据：NovaPay 已从 ``renewal-review`` 回到 ``active``， 延迟交付率回落至历史基线，``notes`` 字段记录 PT-01 修复时间线。 2. 安全事故：2026-02-15 撞库事件状态为 ``closed``，且 ``investigation_report`` 字段已由 ``null`` 变为 Atrium Labs 出具的 独立调查报告引用 —— 关键信息缺失被补齐。 3. 负面新闻：NovaPay 共 6 条记录，包含 ``recovered``、``denied``、 ``confirmed`` 三种状态，覆盖 ``labor`` / ``financial`` / ``operational`` 三个分类；声誉 Agent 应能识别"历史事件已结清"的语义。 4. 制裁名单：NovaPay 仍无命中，必须返回空数组而非抛错。 5. 跨供应商隔离：新增 NovaPay 记录不能影响 ACME 的查询结果数量。 """ from __future__ import annotations from pathlib import Path import unittest from vendor_guard.repository import JsonRepository PROJECT_ROOT = Path(__file__).resolve().parents[1] class NovapayFixtureTests(unittest.TestCase): """针对 ``fixtures/`` 中**整改后** NovaPay 记录的业务层查询行为。""" def setUp(self) -> None: self.repository = JsonRepository(PROJECT_ROOT / "fixtures") # -------- vendor_registry.json -------- def test_vendor_registry_finds_novapay(self) -> None: vendor = self.repository.get_vendor("NOVAPAY") self.assertTrue(vendor["found"]) self.assertEqual(vendor["vendor"], "NOVAPAY") self.assertEqual(vendor["status"], "active") self.assertTrue(vendor["beneficial_owner_verified"]) def test_vendor_registry_lookup_is_case_insensitive_for_novapay(self) -> None: for variant in ("novapay", "NovaPay", "NOVAPAY", "NovaPAY"): with self.subTest(variant=variant): vendor = self.repository.get_vendor(variant) self.assertTrue(vendor["found"], msg=f"未命中：{variant}") self.assertEqual(vendor["vendor"], "NOVAPAY") def test_vendor_registry_shows_healthy_operating_signals(self) -> None: """整改后 ``late_delivery_rate`` 应当低于基线警戒值（0.10）， ``disputes_24m`` 不应超过 2，且 ``notes`` 反映 PT-01 修复完成。""" vendor = self.repository.get_vendor("NOVAPAY") self.assertLess(vendor["late_delivery_rate"], 0.10) self.assertLessEqual(vendor["disputes_24m"], 2) self.assertIn("PT-01", vendor["notes"]) self.assertIn("复测通过", vendor["notes"]) def test_vendor_registry_addition_does_not_change_acme(self) -> None: acme = self.repository.get_vendor("ACME") self.assertTrue(acme["found"]) self.assertEqual(acme["late_delivery_rate"], 0.08) self.assertEqual(acme["disputes_24m"], 2) self.assertNotIn("notes", acme) # -------- security_incidents.json -------- def test_security_incident_returns_single_novapay_record(self) -> None: incidents = self.repository.get_incidents("NOVAPAY") self.assertEqual(len(incidents), 1) incident = incidents[0] self.assertEqual(incident["severity"], "high") self.assertEqual(incident["status"], "closed") self.assertEqual(incident["date"], "2026-02-15") def test_security_incident_now_carries_investigation_report(self) -> None: """整改前 ``investigation_report`` 为 ``null``（关键信息缺失）。 整改后必须引用 Atrium Labs 2026-04-10 独立调查报告， 让 Agent 能看到客观证据而不是空白。""" incidents = self.repository.get_incidents("NOVAPAY") self.assertEqual(len(incidents), 1) self.assertIn("investigation_report", incidents[0]) report = incidents[0]["investigation_report"] self.assertIsNotNone(report) self.assertIn("Atrium Labs", report) self.assertIn("2026-04-10", report) def test_security_incident_lookup_is_case_insensitive_for_novapay(self) -> None: for variant in ("novapay", "NovaPay"): with self.subTest(variant=variant): incidents = self.repository.get_incidents(variant) self.assertEqual(len(incidents), 1) self.assertEqual(incidents[0]["vendor"].upper(), "NOVAPAY") def test_security_incident_isolation_acme_count_unchanged(self) -> None: incidents = self.repository.get_incidents("ACME") self.assertEqual(len(incidents), 1) self.assertEqual(incidents[0]["severity"], "medium") # -------- adverse_news.json -------- def test_adverse_news_returns_all_six_novapay_records(self) -> None: news = self.repository.get_adverse_news("NOVAPAY") self.assertEqual(len(news), 6) dates = {item["date"] for item in news} self.assertEqual( dates, { "2026-04-22", "2026-06-10", "2026-07-03", "2026-07-19", "2026-08-05", "2026-08-12", }, ) def test_adverse_news_distinguishes_recovered_denied_confirmed(self) -> None: """整改后三类状态必须被透传，便于声誉 Agent 区分 "历史事件已结清"（``recovered``）、"传言被否认"（``denied``）、 "新近已验证事件"（``confirmed``）。""" news = self.repository.get_adverse_news("NOVAPAY") statuses = {item["status"] for item in news} self.assertEqual(statuses, {"recovered", "denied", "confirmed"}) def test_adverse_news_covers_labor_financial_and_operational(self) -> None: news = self.repository.get_adverse_news("NOVAPAY") categories = {item["category"] for item in news} self.assertEqual(categories, {"labor", "financial", "operational"}) def test_adverse_news_keeps_denied_record_distinct(self) -> None: """``denied`` 记录（TechRabbit 收购传闻 + NovaPay 官方否认）必须 与 ``confirmed`` / ``recovered`` 记录分开，便于声誉 Agent 按 ``skills/reputation-review/SKILL.md`` 的来源优先级排序。""" news = self.repository.get_adverse_news("NOVAPAY") denied = [item for item in news if item["status"] == "denied"] confirmed = [item for item in news if item["status"] == "confirmed"] recovered = [item for item in news if item["status"] == "recovered"] self.assertEqual(len(denied), 1) self.assertEqual(len(confirmed), 2) self.assertEqual(len(recovered), 3) self.assertIn("收购", denied[0]["summary"]) def test_adverse_news_recovered_records_carry_resolution_evidence(self) -> None: """``recovered`` 记录必须包含客观的恢复/结清证据，否则 Agent 无法判断历史事件的影响是否仍存在。允许使用 ``恢复`` / ``已完成`` / ``已结清`` 等多种表达。""" news = self.repository.get_adverse_news("NOVAPAY") recovered = [item for item in news if item["status"] == "recovered"] resolution_markers = ("恢复", "已完成", "已结清", "recovered") for item in recovered: with self.subTest(date=item["date"]): haystack = item["summary"] + item.get("source", "") self.assertTrue( any(marker in haystack for marker in resolution_markers), msg=f"recovered 记录缺少恢复证据：{item['date']}", ) def test_adverse_news_lookup_is_case_insensitive_for_novapay(self) -> None: for variant in ("novapay", "NovaPay", "NOVAPAY"): with self.subTest(variant=variant): news = self.repository.get_adverse_news(variant) self.assertEqual(len(news), 6) def test_adverse_news_acme_count_unchanged(self) -> None: news = self.repository.get_adverse_news("ACME") self.assertEqual(len(news), 1) self.assertEqual(news[0]["status"], "settled") # -------- sanctions.json -------- def test_sanctions_return_empty_for_novapay(self) -> None: result = self.repository.get_sanctions("NOVAPAY") self.assertEqual(result, []) def test_sanctions_lookup_is_case_insensitive_for_novapay(self) -> None: for variant in ("novapay", "NovaPay", "NOVAPAY"): with self.subTest(variant=variant): self.assertEqual(self.repository.get_sanctions(variant), []) def test_sanctions_still_empty_for_acme(self) -> None: self.assertEqual(self.repository.get_sanctions("ACME"), []) # -------- 跨供应商隔离（防止记录泄漏） -------- def test_novapay_records_do_not_leak_into_acme_queries(self) -> None: self.assertEqual(len(self.repository.get_incidents("ACME")), 1) self.assertEqual(len(self.repository.get_adverse_news("ACME")), 1) # 反向：ACME 的事件也不应出现在 NovaPay 的查询中。 novapay_incidents = self.repository.get_incidents("NOVAPAY") for incident in novapay_incidents: self.assertEqual(incident["vendor"], "NOVAPAY") novapay_news = self.repository.get_adverse_news("NOVAPAY") for item in novapay_news: self.assertEqual(item["vendor"], "NOVAPAY") if __name__ == "__main__": unittest.main()