#!/usr/bin/env python3
"""三个 Master 新增契约校验器的最小正反例回归测试。"""

from __future__ import annotations

import hashlib
import json
import struct
import subprocess
import tempfile
import unittest
import zlib
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
PRD_VALIDATOR = ROOT / "prd-master/validators/check_requirements_contract.py"
PRD_CONSISTENCY_VALIDATOR = ROOT / "prd-master/validators/check_consistency.py"
DESIGN_VALIDATOR = ROOT / "design-master/validators/check_traceability.py"
DELIVERY_VALIDATOR = ROOT / "tdd-master/validators/check_delivery_contract.py"
REQUIREMENTS_TEMPLATE = ROOT / "prd-master/templates/requirements.md"
ANALYSIS_TEMPLATE = ROOT / "prd-master/templates/requirements-analysis.md"


def run_validator(script: Path, *args: str | Path) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["python3", str(script), *(str(arg) for arg in args)],
        check=False,
        capture_output=True,
        text=True,
    )


def write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


def digest(path: Path) -> str:
    return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest()


def write_png(path: Path, width: int = 320, height: int = 200) -> None:
    def chunk(kind: bytes, data: bytes) -> bytes:
        payload = kind + data
        return (
            struct.pack(">I", len(data))
            + payload
            + struct.pack(">I", zlib.crc32(payload) & 0xFFFFFFFF)
        )

    pixel = b"\x2f\x6f\xff"
    raw = b"".join(b"\x00" + pixel * width for _ in range(height))
    png = (
        b"\x89PNG\r\n\x1a\n"
        + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0))
        + chunk(b"IDAT", zlib.compress(raw))
        + chunk(b"IEND", b"")
    )
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(png)


TRACEABILITY = """# 设计追溯矩阵

| Source ID | Source Revision | Design IDs | Artifact | Error/Invariant | Verification Surface | Status |
| :--- | :--- | :--- | :--- | :--- | :--- | :--- |
| REQ-F01-01 | 1 | API-F01-01 | 技术方案.md §4 | 请求必须可校验 | integration | covered |
| AC-F01-01 | 1 | API-F01-01 | 技术方案.md §4 | 错误结果可观察 | integration | covered |
| NFR-001 | 1 | DEC-001 | 技术方案.md §6 | 延迟不超过阈值 | performance | covered |
"""

STAGED_REQUIREMENTS = """# 分阶段产品 Requirements Contract

## Metadata

- work_type: feature
- workflow_mode: standard
- revision: 1
- source_prd: PRD详细版.md
- status: active

## Feature F01 · MVP 功能

### REQ-F01-01 · MVP 行为

- Stage: MVP
- Revision: 1
- Status: active

### AC-F01-01 · MVP 结果

- Parent: REQ-F01-01
- Priority: P0
- EARS: WHEN 用户提交 THE SYSTEM SHALL 保存结果

## Feature F02 · 后续功能

### REQ-F02-01 · 后续行为

- Stage: 阶段2
- Revision: 1
- Status: active

### AC-F02-01 · 后续结果

- Parent: REQ-F02-01
- Priority: P1
- EARS: WHEN 用户触发 THE SYSTEM SHALL 返回结果

### NFR-002 · 后续指标

- Applies-to: REQ-F02-01
- Revision: 1
- Status: active
- Measure: 阶段2启动时定义并验证指标
"""

STAGED_TRACEABILITY = """# 设计追溯矩阵

| Source ID | Source Revision | Design IDs | Artifact | Error/Invariant | Verification Surface | Status |
| :--- | :--- | :--- | :--- | :--- | :--- | :--- |
| REQ-F01-01 | 1 | PAGE-F01-01, API-F01-01 | pages/阶段1_核心_首页.md; 技术方案.md §4 | 保存失败不得显示成功 | integration + e2e | covered |
| AC-F01-01 | 1 | PAGE-F01-01, API-F01-01 | pages/阶段1_核心_首页.md; 技术方案.md §4 | 保存结果可重新读取 | integration + e2e | covered |
| REQ-F02-01 | 1 | PAGE-F02-01, DEC-002 | 页面清单.md; 设计决策蓝图.md §4 | 依赖MVP主体ID保持稳定 | 阶段2启动时重新深设计与校验 | planned |
| AC-F02-01 | 1 | PAGE-F02-01, DEC-002 | 页面清单.md; 设计决策蓝图.md §4 | 依赖MVP主体ID保持稳定 | 阶段2启动时重新深设计与校验 | planned |
| NFR-002 | 1 | DEC-002 | 技术方案.md §8.3 | 阶段2指标不得反推MVP提前建设 | 阶段2启动时重新深设计与校验 | planned |
"""

BUGFIX_CONTRACT = """# Bugfix Contract

### REPRO-001 · 稳定复现
- Status: active

### CUR-001 · 当前错误
- Status: active

### EXP-001 · 预期结果
- Status: active

### UNCH-001 · 不变行为
- Status: active

### CON-001 · 修复约束
- Status: active
"""

BUGFIX_TRACEABILITY = """# 设计追溯矩阵

| Source ID | Source Revision | Design IDs | Artifact | Error/Invariant | Verification Surface | Status |
| :--- | :--- | :--- | :--- | :--- | :--- | :--- |
| REPRO-001 | 1 | FIX-BUG-001 | 修复设计.md §2 | 相同输入稳定复现 | integration | covered |
| CUR-001 | 1 | FIX-BUG-001 | 修复设计.md §2 | 当前错误结果明确 | integration | covered |
| EXP-001 | 1 | FIX-BUG-001 | 修复设计.md §3 | 修复后结果明确 | integration | covered |
| UNCH-001 | 1 | FIX-BUG-001 | 修复设计.md §4 | 不变行为不得退化 | regression | covered |
| CON-001 | 1 | FIX-BUG-001 | 修复设计.md §5 | 禁止扩大修复面 | review | covered |
"""


class RequirementsValidatorTests(unittest.TestCase):
    def test_accepts_feature_template(self) -> None:
        result = run_validator(PRD_VALIDATOR, REQUIREMENTS_TEMPLATE, ANALYSIS_TEMPLATE)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_rejects_ac_without_parent(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "- Parent: REQ-F01-01\n", ""
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("缺少合法 Parent", result.stdout)

    def test_rejects_user_story_without_role(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "- Role: {{角色}}\n", ""
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("US-F01-01 缺少有效 Role", result.stdout)

    def test_rejects_missing_external_configuration_decision(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8")
            start = content.index("## External capability configuration")
            end = content.index("## Capability prerequisites", start)
            write(contract, content[:start] + content[end:])
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("缺少 External capability configuration", result.stdout)

    def test_rejects_missing_capability_prerequisites_decision(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8")
            start = content.index("## Capability prerequisites")
            end = content.index("## Feature F01", start)
            write(contract, content[:start] + content[end:])
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("缺少 Capability prerequisites", result.stdout)

    def test_rejects_none_when_llm_is_declared(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "- Goal: {{目标}}",
                "- Goal: 调用 LLM 生成结果",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("正文出现外部能力信号“LLM”", result.stdout)
            self.assertIn("配置责任不能填写 none", result.stdout)

    def test_accepts_explicit_llm_settings_responsibility(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "| none | - | - | - | - | - | - | - |",
                "| LLM | tenant | tenant-admin | settings | tenant | validate-mask-rotate-delete | MVP | REQ-F01-01 |",
            ).replace(
                "| none | - | - | - | - | - | - |\n",
                "| LLM 服务账号 | ready | platform-owner | 已验证可调用 | 改用本地规则 | MVP | REQ-F01-01 |\n",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_rejects_product_configuration_surface_without_requirement(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "| none | - | - | - | - | - | - | - |",
                "| LLM | tenant | tenant-admin | settings | tenant | validate-mask-rotate-delete | MVP | - |",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn(
                "外部能力 LLM 通过 settings 配置，必须关联至少一个真实 REQ",
                result.stdout,
            )

    def test_rejects_external_configuration_without_lifecycle(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "| none | - | - | - | - | - | - | - |",
                "| LLM | tenant | tenant-admin | settings | tenant | - | MVP | REQ-F01-01 |",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("外部能力 LLM 缺少 Lifecycle", result.stdout)

    def test_accepts_explicit_deployment_secret_without_product_page(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "| none | - | - | - | - | - | - | - |",
                "| LLM | platform | technical-operator | deployment-secret | system | operator-rotation | MVP | - |",
            ).replace(
                "| none | - | - | - | - | - | - |\n",
                "| LLM 服务账号 | committed | technical-operator | 2026-08-01 | 移出 MVP | MVP | REQ-F01-01 |\n",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_rejects_sms_auth_without_confirmed_channel_prerequisite(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "- Goal: {{目标}}",
                "- Goal: 使用短信验证码完成注册",
            ).replace(
                "| none | - | - | - | - | - | - | - |",
                "| SMS | platform | technical-operator | deployment-secret | system | rotate-on-expiry | MVP | REQ-F01-01 |",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("方案前置能力信号“短信验证码”", result.stdout)
            self.assertIn("前置能力不能填写 none", result.stdout)

    def test_rejects_unrelated_prerequisite_for_sms_auth(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "- Goal: {{目标}}",
                "- Goal: 使用短信验证码完成注册",
            ).replace(
                "| none | - | - | - | - | - | - | - |",
                "| SMS | platform | technical-operator | deployment-secret | system | rotate-on-expiry | MVP | REQ-F01-01 |",
            ).replace(
                "| none | - | - | - | - | - | - |\n",
                "| Apple 开发者账号 | ready | release-owner | 已验证 | Web 版 | MVP | REQ-F01-01 |\n",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("依赖短信通道，但方案前置能力表没有对应行", result.stdout)

    def test_rejects_ios_distribution_without_account_prerequisite(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "- Goal: {{目标}}",
                "- Goal: 通过 App Store 正式发布 iOS 应用",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("方案前置能力信号“App Store”", result.stdout)

    def test_accepts_committed_ios_distribution_prerequisite(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "- Goal: {{目标}}",
                "- Goal: 通过 App Store 正式发布 iOS 应用",
            ).replace(
                "| none | - | - | - | - | - | - |\n",
                "| Apple 开发者账号与签名 | committed | release-owner | 2026-08-15 | 先交付 Web 版 | MVP | REQ-F01-01 |\n",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_rejects_unresolved_prerequisite_status(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            contract = Path(directory) / "requirements.md"
            content = REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8").replace(
                "| none | - | - | - | - | - | - |\n",
                "| Apple 开发者账号 | unknown | release-owner | 待确认 | Web 版 | MVP | REQ-F01-01 |\n",
            )
            write(contract, content)
            result = run_validator(PRD_VALIDATOR, contract, ANALYSIS_TEMPLATE)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("Status 非法: unknown", result.stdout)


class PrdRoleConsistencyValidatorTests(unittest.TestCase):
    PRD = """# 角色校验示例 PRD

## 2. 用户画像

### 2.1 用户角色矩阵

| 角色 | 身份 | 目标 |
| :--- | :--- | :--- |
| 决策者 | 企业家本人 | 查看并拍板 |
| 执行者 | 业务助理 | 执行清单 |

### 2.3 用户故事

- **US-1**: 作为企业家，我希望查看决策结果，以便完成拍板。（由 FR-1 实现）

## 3. 功能需求

- **FR-1**: 展示决策结果并支持确认。
"""

    def test_accepts_story_role_alias_declared_in_role_matrix(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            prd = Path(directory) / "PRD详细版.md"
            write(prd, self.PRD)
            result = run_validator(PRD_CONSISTENCY_VALIDATOR, prd)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertIn("US-1 → 企业家 [PASS]", result.stdout)

    def test_rejects_story_role_missing_from_role_matrix(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            prd = Path(directory) / "PRD详细版.md"
            write(prd, self.PRD.replace("作为企业家", "作为平台运营员"))
            result = run_validator(PRD_CONSISTENCY_VALIDATOR, prd)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn(
                "US-1 的角色“平台运营员”未出现在 §2.1 角色矩阵",
                result.stdout,
            )


class TraceabilityValidatorTests(unittest.TestCase):
    def test_accepts_complete_traceability(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            trace = Path(directory) / "设计追溯矩阵.md"
            write(trace, TRACEABILITY)
            result = run_validator(DESIGN_VALIDATOR, REQUIREMENTS_TEMPLATE, trace)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_rejects_missing_source_row(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            trace = Path(directory) / "设计追溯矩阵.md"
            write(trace, TRACEABILITY.replace("| NFR-001 | 1 | DEC-001 | 技术方案.md §6 | 延迟不超过阈值 | performance | covered |\n", ""))
            result = run_validator(DESIGN_VALIDATOR, REQUIREMENTS_TEMPLATE, trace)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("缺少 source 行: NFR-001", result.stdout)

    def test_accepts_planned_traceability_for_future_stage(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            requirements = Path(directory) / "requirements.md"
            trace = Path(directory) / "设计追溯矩阵.md"
            write(requirements, STAGED_REQUIREMENTS)
            write(trace, STAGED_TRACEABILITY)
            result = run_validator(DESIGN_VALIDATOR, requirements, trace)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertIn("2 个 MVP source 施工级覆盖 / 3 个后续 source 已规划", result.stdout)

    def test_rejects_planned_status_for_mvp_source(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            requirements = Path(directory) / "requirements.md"
            trace = Path(directory) / "设计追溯矩阵.md"
            write(requirements, STAGED_REQUIREMENTS)
            write(
                trace,
                STAGED_TRACEABILITY.replace(
                    "| REQ-F01-01 | 1 | PAGE-F01-01, API-F01-01 | pages/阶段1_核心_首页.md; 技术方案.md §4 | 保存失败不得显示成功 | integration + e2e | covered |",
                    "| REQ-F01-01 | 1 | PAGE-F01-01, API-F01-01 | pages/阶段1_核心_首页.md; 技术方案.md §4 | 保存失败不得显示成功 | integration + e2e | planned |",
                ),
            )
            result = run_validator(DESIGN_VALIDATOR, requirements, trace)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("MVP source REQ-F01-01 状态必须为 covered/preserved", result.stdout)

    def test_rejects_planned_status_for_bugfix_source(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            bugfix = Path(directory) / "bugfix.md"
            trace = Path(directory) / "设计追溯矩阵.md"
            write(bugfix, BUGFIX_CONTRACT)
            write(trace, BUGFIX_TRACEABILITY.replace("| covered |", "| planned |", 1))
            result = run_validator(DESIGN_VALIDATOR, bugfix, trace)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("Bugfix source REPRO-001 状态必须为 covered/preserved", result.stdout)

    def test_rejects_feature_requirement_without_stage(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            requirements = Path(directory) / "requirements.md"
            trace = Path(directory) / "设计追溯矩阵.md"
            write(requirements, STAGED_REQUIREMENTS.replace("- Stage: 阶段2\n", "", 1))
            write(trace, STAGED_TRACEABILITY)
            result = run_validator(DESIGN_VALIDATOR, requirements, trace)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("REQ-F02-01 缺少 Stage", result.stdout)

    def test_rejects_planned_source_with_fake_page_artifact(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            requirements = Path(directory) / "requirements.md"
            trace = Path(directory) / "设计追溯矩阵.md"
            write(requirements, STAGED_REQUIREMENTS)
            write(
                trace,
                STAGED_TRACEABILITY.replace(
                    "页面清单.md; 设计决策蓝图.md §4",
                    "pages/阶段2_扩展_详情页.md",
                    1,
                ),
            )
            result = run_validator(DESIGN_VALIDATOR, requirements, trace)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("planned Artifact 必须指向页面清单/蓝图/技术方案骨架", result.stdout)
            self.assertIn("不得伪造尚未生成的页面文件", result.stdout)


class CompactDeliveryValidatorTests(unittest.TestCase):
    PAGE_LIST = """# 页面清单

| 页面 ID | Source IDs | 所属阶段 | 功能模块 | 页面名称 | 功能范围描述 |
| :--- | :--- | :--- | :--- | :--- | :--- |
| PAGE-F01-01 | REQ-F01-01, AC-F01-01 | 阶段1(MVP) | 核心 | 首页 | 展示并保存核心结果 |
"""

    CONTRACT = """# TDD 验收契约 · 示例

## 0. 契约元信息

- work_type: feature
- source_contract: requirements.md
- scope: MVP

## 1. 完成定义

全部 required 行通过。

## 2. 冒烟门禁

| ID | Source IDs | Design IDs / 产物 | 场景与操作 | 可观察通过标准 | 验证方式 |
| :--- | :--- | :--- | :--- | :--- | :--- |
| `SMOKE-01` | `-` | `-` | 启动应用并打开首页 | 首页可访问且无阻断错误 | 启动命令 + 浏览器 |

## 3. 功能与流程验收

| ID | Source IDs | Design IDs / 产物 | 场景与操作 | 可观察通过标准 | 验证方式 |
| :--- | :--- | :--- | :--- | :--- | :--- |
| `FLOW-01` | `REQ-F01-01, AC-F01-01` | `PAGE-F01-01, API-F01-01` | 用户保存核心结果 | 页面反馈成功且结果可重新读取 | 浏览器 E2E + API |

## 4. 设计落地验收

| ID | Source IDs | Design IDs / 产物 | 场景与操作 | 可观察通过标准 | 验证方式 |
| :--- | :--- | :--- | :--- | :--- | :--- |
| `DESIGN-01` | `REQ-F01-01, AC-F01-01` | `PAGE-F01-01; {md_path}; {html_path}` | 打开首页并操作核心区域 | 结构=页头>内容区; 组件=1个保存按钮; 内容=首页标题和核心结果; 交互=保存并重新读取; 视觉=单列栅格和主按钮层级; 原生适配=仅窗口外壳 | 同状态原型/成品 PNG 截图对 + 独立视觉复核；UI 自动化不能代替 |

## 5. 风险触发验收

本期无额外风险验收。

## 6. 本期不阻塞项

- 完整性能压测：当前无明确 NFR。
"""

    def make_output(
        self,
        directory: str,
        *,
        actual_md: str = "pages/阶段1_核心_首页.md",
        actual_html: str = "pages/阶段1_核心_首页.html",
        contract_md: str | None = None,
        contract_html: str | None = None,
    ) -> Path:
        output = Path(directory) / "output"
        write(output / "requirements.md", STAGED_REQUIREMENTS)
        write(output / "设计追溯矩阵.md", STAGED_TRACEABILITY)
        write(output / "页面清单.md", self.PAGE_LIST)
        write(
            output / actual_md,
            """# 页面名称：首页

| 字段 | 内容 |
| :--- | :--- |
| 页面 ID | `PAGE-F01-01` |
| 所属模块 | 核心 |
""",
        )
        write(
            output / actual_html,
            """<!doctype html>
<html><head><meta name="page-id" content="PAGE-F01-01"><title>首页</title></head>
<body><main><h1>首页</h1><button>保存核心结果</button></main></body></html>
""",
        )
        contract = self.CONTRACT.format(
            md_path=contract_md or actual_md,
            html_path=contract_html or actual_html,
        )
        write(output / "tests/TDD验收契约.md", contract)
        return output

    def write_result(
        self,
        output: Path,
        *,
        status: str = "pass",
        omitted_id: str | None = None,
        contract_hash: str | None = None,
        visual_evidence: bool = True,
        blocking: str = "0",
        actual_size: tuple[int, int] = (320, 200),
    ) -> None:
        contract_path = output / "tests/TDD验收契约.md"
        baseline = output / "tests/visual/PAGE-F01-01-baseline.png"
        actual = output / "tests/visual/PAGE-F01-01-actual.png"
        if visual_evidence:
            write_png(baseline)
            write_png(actual, *actual_size)
            design_evidence = (
                "baseline=tests/visual/PAGE-F01-01-baseline.png; "
                "actual=tests/visual/PAGE-F01-01-actual.png; "
                "reviewer=independent-visual; review=pass; "
                f"blocking={blocking}"
            )
        else:
            design_evidence = "Playwright UI 自动化 1 passed"
        rows = []
        for test_id, action, evidence in (
            ("SMOKE-01", "运行启动命令并打开首页", "exit 0"),
            ("FLOW-01", "执行保存与重新读取 E2E", "1 passed；API 返回已保存数据"),
            ("DESIGN-01", "独立验收官实际查看 baseline+actual 并看图核对", design_evidence),
        ):
            if test_id != omitted_id:
                rows.append(f"| `{test_id}` | `{status}` | {action} | {evidence} |")
        write(
            output / "tests/TDD验收结果.md",
            "\n".join(
                (
                    "# TDD 验收结果",
                    "",
                    f"- contract_sha256: `{contract_hash or digest(contract_path)}`",
                    "- run_started_at: `2026-07-22T10:00:00+08:00`",
                    "",
                    "| TDD ID | 状态 | 实际验证动作 | 证据 |",
                    "| :--- | :--- | :--- | :--- |",
                    *rows,
                )
            ),
        )

    def test_accepts_canonical_page_artifact_pair(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertNotIn("命名/目录不规范", result.stdout)

    def test_warns_but_accepts_unique_content_under_noncanonical_names(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(
                directory,
                actual_md="pages/specs/home-document.md",
                actual_html="pages/html/final-screen.html",
            )
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertIn("命名/目录不规范，但内容身份唯一", result.stdout)

    def test_recovers_from_stale_contract_paths_by_unique_content(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(
                directory,
                actual_md="pages/specs/home-document.md",
                actual_html="pages/html/final-screen.html",
                contract_md="pages/阶段1_核心_首页.md",
                contract_html="pages/阶段1_核心_首页.html",
            )
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertIn("未准确记录 PAGE-F01-01 的实际 MD/HTML 路径", result.stdout)

    def test_rejects_ambiguous_content_match(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            write(
                output / "pages/another-home.html",
                '<html><head><meta name="page-id" content="PAGE-F01-01"><title>首页</title></head></html>',
            )
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn(".html 内容匹配不唯一", result.stdout)

    def test_does_not_treat_cross_reference_as_page_identity(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            write(
                output / "pages/阶段1_其他_说明页.md",
                """# 页面名称：说明页

| 字段 | 内容 |
| :--- | :--- |
| 页面 ID | `PAGE-F02-01` |

相关页面：PAGE-F01-01。
""",
            )
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_accepts_complete_execution_ledger_with_fresh_evidence(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            self.write_result(output)
            result = run_validator(DELIVERY_VALIDATOR, "--result", output)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertIn("execution ledger all pass", result.stdout)

    def test_rejects_design_pass_backed_only_by_ui_automation(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            self.write_result(output, visual_evidence=False)
            result = run_validator(DELIVERY_VALIDATOR, "--result", output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("DESIGN-01 缺少视觉证据字段", result.stdout)

    def test_rejects_design_with_blocking_diff_or_mismatched_screenshot_size(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            self.write_result(output, blocking="1", actual_size=(300, 200))
            result = run_validator(DELIVERY_VALIDATOR, "--result", output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("DESIGN-01 仍有阻塞视觉差异: 1", result.stdout)
            self.assertIn("原型/成品截图尺寸不同", result.stdout)

    def test_rejects_vague_design_acceptance_without_fidelity_anchors(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            contract = output / "tests/TDD验收契约.md"
            text = contract.read_text(encoding="utf-8").replace(
                "结构=页头>内容区; 组件=1个保存按钮; 内容=首页标题和核心结果; "
                "交互=保存并重新读取; 视觉=单列栅格和主按钮层级; 原生适配=仅窗口外壳",
                "页面与高保真原型一致",
            )
            write(contract, text)
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("DESIGN-01 的通过标准缺少具体视觉锚点", result.stdout)

    def test_rejects_pending_or_missing_execution_rows(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            self.write_result(output, status="pending", omitted_id="DESIGN-01")
            result = run_validator(DELIVERY_VALIDATOR, "--result", output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("TDD验收结果漏项: DESIGN-01", result.stdout)
            self.assertIn("SMOKE-01 尚未通过: pending", result.stdout)

    def test_rejects_result_ledger_for_stale_contract(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            self.write_result(output, contract_hash="sha256:" + "0" * 64)
            result = run_validator(DELIVERY_VALIDATOR, "--result", output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("旧通过状态全部失效", result.stdout)

    def test_contract_mode_does_not_treat_an_old_result_as_contract_content(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            self.write_result(output, contract_hash="sha256:" + "0" * 64)
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            self.assertNotIn("execution ledger", result.stdout)

    def test_result_mode_rejects_missing_execution_ledger(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            result = run_validator(DELIVERY_VALIDATOR, "--result", output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("TDD验收结果.md", result.stdout)


class DeliveryValidatorTests(unittest.TestCase):
    def make_output(self, directory: str) -> Path:
        output = Path(directory) / "output"
        requirements = output / "requirements.md"
        design_trace = output / "设计追溯矩阵.md"
        test_index = output / "tests/测试索引.md"
        write(requirements, REQUIREMENTS_TEMPLATE.read_text(encoding="utf-8"))
        write(design_trace, TRACEABILITY)
        write(
            test_index,
            """# 测试索引

| ID | Name | Priority | Source IDs | Design IDs | Expected Failure | Status |
| :--- | :--- | :--- | :--- | :--- | :--- | :--- |
| U-F01-01 | 验证核心行为 | P0 | REQ-F01-01, AC-F01-01, NFR-001 | API-F01-01, DEC-001 | 接口尚未满足契约 | pending |
""",
        )
        write(
            output / "开发任务图.md",
            """# 开发任务图

| ID | Outcome | Source IDs | Design IDs | Test IDs | Depends On | Wave | Class | Done When | Status |
| :--- | :--- | :--- | :--- | :--- | :--- | :--- | :--- | :--- | :--- |
| TASK-F01-01 | 核心行为满足契约 | REQ-F01-01, AC-F01-01, NFR-001 | API-F01-01, DEC-001 | U-F01-01 | - | 1 | required | RED 原因匹配；GREEN；全量回归绿 | pending |
""",
        )
        write(
            output / "全链路追溯矩阵.md",
            """# 全链路追溯矩阵

| Source ID | Design IDs | Test IDs | Task IDs | Status |
| :--- | :--- | :--- | :--- | :--- |
| REQ-F01-01 | API-F01-01 | U-F01-01 | TASK-F01-01 | covered |
| AC-F01-01 | API-F01-01 | U-F01-01 | TASK-F01-01 | covered |
| NFR-001 | DEC-001 | U-F01-01 | TASK-F01-01 | covered |
""",
        )
        state = {
            "schema_version": "1.0",
            "work_type": "feature",
            "active_contract": "requirements.md",
            "source_hashes": {
                "requirements.md": digest(requirements),
                "设计追溯矩阵.md": digest(design_trace),
                "tests/测试索引.md": digest(test_index),
            },
            "tasks": [
                {
                    "id": "TASK-F01-01",
                    "status": "pending",
                    "wave": 1,
                    "required": True,
                    "source_ids": ["REQ-F01-01", "AC-F01-01", "NFR-001"],
                    "design_ids": ["API-F01-01", "DEC-001"],
                    "test_ids": ["U-F01-01"],
                    "last_verified_at": None,
                    "evidence": [],
                }
            ],
        }
        write(output / "task-state.json", json.dumps(state, ensure_ascii=False, indent=2))
        return output

    def test_accepts_consistent_delivery_graph(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_rejects_changed_upstream_fingerprint(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            output = self.make_output(directory)
            state_path = output / "task-state.json"
            state = json.loads(state_path.read_text(encoding="utf-8"))
            state["source_hashes"]["requirements.md"] = "sha256:" + "0" * 64
            write(state_path, json.dumps(state, ensure_ascii=False, indent=2))
            result = run_validator(DELIVERY_VALIDATOR, output)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("上游指纹不匹配: requirements.md", result.stdout)


if __name__ == "__main__":
    unittest.main()