aps-agent/tests/golden/test_rebuild_pipeline.py

# ============================================================
# 可重生算法流水线黄金测试（矩阵 91，方向 T）：
#   T1 候选构建 → 隔离黄金通过 → shadow 灰度 → 灰度转正（promote）
#   T2 黄金失败 → 自动回滚（注册表恢复 + 报告）
#   T3 未登记 moduleId → 显式失败
#   T4 灰度期失败（shadow 探针）→ 自动回退到上一可用注册版本
#   T5 审计事件齐全（started/passed/failed/gray/rolled_back）且链完整
#   T6 安全扫描：缺可重生注释 → 阻断；危险原语 / 未登记写世界 → 告警
#   T7 网关 /api/rebuild/* 端点接线（404 显式失败 / 状态列表）
# 隔离：假 moduleId 通过 source_path 注入（不跑真实全部黄金，避免慢）；
#       隔离运行只跑最小黄金子集，APS_* 指向随机临时目录。
# ============================================================
from __future__ import annotations

from pathlib import Path

import pytest

from server.agent_core.registry import (
    active_version,
    baseline_version,
    get_rebuild_candidate,
    verify_audit_chain,
)
from server.agent_core.rebuild_orchestrator import (
    RebuildLookupError,
    RebuildOrchestrator,
    run_isolated_golden,
    static_safety_scan,
)

ROOT = Path(__file__).resolve().parents[2]
MODULE_ID = "fixture-rebuild-demo"


class _Store:
    """最小可审计 store 替身（同 test_audit_anchor_api 模式，隔离生产 world）。"""

    def __init__(self) -> None:
        self.tenant_uuid = "platform"
        self.world_key = "default"
        self.data: dict = {}
        self._ids: dict[str, int] = {}

    def next_id(self, kind: str) -> int:
        self._ids[kind] = self._ids.get(kind, 0) + 1
        return self._ids[kind]

    def save(self) -> None:
        pass


@pytest.fixture()
def fixture_module(tmp_path):
    """自带可重生头注释的假模块 + 最小黄金子集（隔离环境自检 + 冒烟）。"""
    src = tmp_path / "fixture_rebuild_demo.py"
    src.write_text(
        "# ============================================================\n"
        "# 测试用可重生模块（moduleId: fixture-rebuild-demo, 可重生 ✅）\n"
        "# ============================================================\n"
        "from __future__ import annotations\n\n"
        "def demo_value() -> int:\n"
        "    return 42\n",
        encoding="utf-8",
    )
    golden = tmp_path / "test_fixture_rebuild_demo.py"
    src_dir = str(src.parent).replace("\\", "\\\\")
    golden.write_text(
        "import os\n"
        "import sys\n"
        "import tempfile\n\n"
        "def test_isolated_env() -> None:\n"
        "    wp = os.environ.get('APS_WORLD_PATH', '')\n"
        "    assert 'aps-rebuild-' in wp, wp\n"
        "    assert os.environ.get('APS_DB_PATH', '')\n"
        "    assert os.environ.get('APS_KNOWLEDGE_PATH', '')\n\n"
        "def test_candidate_smoke() -> None:\n"
        f"    sys.path.insert(0, r'{src_dir}')\n"
        "    from fixture_rebuild_demo import demo_value\n"
        "    assert demo_value() == 42\n",
        encoding="utf-8",
    )
    return {"moduleId": MODULE_ID, "source": str(src), "golden": str(golden)}


def _rebuild_actions(store) -> list[str]:
    return [ev.get("action") for ev in store.data.get("auditEvents", [])]


# ---------------- T1：候选构建 → 隔离黄金通过 → 灰度 → 转正 ----------------
def test_rebuild_full_flow_promote(fixture_module):
    store = _Store()
    orch = RebuildOrchestrator(store, repo_root=str(ROOT))
    report = orch.start_rebuild(MODULE_ID, golden_tests=[fixture_module["golden"]],
                                source_path=fixture_module["source"], actor="tester")
    assert report["status"] == "GRAY", report
    assert report["golden"]["passed"] is True
    assert report["golden"]["elapsedSec"] >= 0
    assert report["safety"]["ok"] is True
    # 构建包：源文件清单 + 依赖 + 黄金测试映射
    assert report["package"]["sourceFiles"] == [fixture_module["source"].replace("\\", "/")]
    assert report["package"]["goldenTests"] == [fixture_module["golden"]]
    # shadow 灰度：候选与现行并存（激活指针不动，基线 current）
    assert active_version(store.data, MODULE_ID) is None
    assert baseline_version(store.data, MODULE_ID) == "current"
    cand = get_rebuild_candidate(store.data, MODULE_ID)
    assert cand["status"] == "GRAY"
    assert cand["version"] == f"{MODULE_ID}-v1"
    actions = _rebuild_actions(store)
    assert "rebuild.started" in actions and "rebuild.passed" in actions
    # 灰度转正
    promoted = orch.promote(MODULE_ID, actor="tester")
    assert promoted["status"] == "ACTIVE"
    assert active_version(store.data, MODULE_ID) == cand["version"]
    assert "rebuild.gray" in _rebuild_actions(store)
    assert verify_audit_chain(store.data["auditEvents"])["ok"] is True


# ---------------- T2：黄金失败 → 自动回滚 ----------------
def test_rebuild_golden_failure_auto_rollback(fixture_module):
    bad = Path(fixture_module["golden"]).parent / "test_fails.py"
    bad.write_text("def test_must_fail() -> None:\n    assert 1 == 2\n", encoding="utf-8")
    store = _Store()
    orch = RebuildOrchestrator(store, repo_root=str(ROOT))
    report = orch.start_rebuild(MODULE_ID, golden_tests=[str(bad)],
                                source_path=fixture_module["source"], actor="tester")
    assert report["status"] == "FAILED", report
    assert report["golden"]["passed"] is False
    assert report["rollback"]["restoredTo"] == "current"
    cand = get_rebuild_candidate(store.data, MODULE_ID)
    assert cand["status"] == "FAILED"
    assert cand["autoRolledBack"] is True
    actions = _rebuild_actions(store)
    assert "rebuild.failed" in actions and "rebuild.rolled_back" in actions
    assert verify_audit_chain(store.data["auditEvents"])["ok"] is True


# ---------------- T3：未登记 moduleId → 显式失败 ----------------
def test_rebuild_unregistered_module_explicit_failure():
    store = _Store()
    orch = RebuildOrchestrator(store, repo_root=str(ROOT))
    with pytest.raises(RebuildLookupError):
        orch.start_rebuild("fixture-no-such-module-xyz", actor="tester")
    # 注册表无候选、无审计（未启动流水线）
    assert get_rebuild_candidate(store.data, "fixture-no-such-module-xyz") is None
    assert store.data.get("auditEvents") is None


# ---------------- T4：灰度期失败 → 自动回退上一可用注册版本 ----------------
def test_rebuild_gray_failure_auto_rollback(fixture_module):
    store = _Store()
    orch = RebuildOrchestrator(store, repo_root=str(ROOT))

    def bad_probe(report):
        return {"ok": False, "reason": "shadow 流量异常"}

    report = orch.start_rebuild(MODULE_ID, golden_tests=[fixture_module["golden"]],
                                source_path=fixture_module["source"],
                                gray_probe=bad_probe, actor="tester")
    assert report["status"] == "ROLLED_BACK", report
    assert report["rollback"]["restoredTo"] == "current"
    cand = get_rebuild_candidate(store.data, MODULE_ID)
    assert cand["status"] == "ROLLED_BACK"
    actions = _rebuild_actions(store)
    for expected in ("rebuild.started", "rebuild.passed", "rebuild.failed",
                     "rebuild.rolled_back"):
        assert expected in actions, f"缺审计事件 {expected}"
    assert verify_audit_chain(store.data["auditEvents"])["ok"] is True


# ---------------- T5：审计事件齐全 + 链完整 ----------------
def test_rebuild_audit_events_complete(fixture_module):
    store = _Store()
    orch = RebuildOrchestrator(store, repo_root=str(ROOT))
    orch.start_rebuild(MODULE_ID, golden_tests=[fixture_module["golden"]],
                       source_path=fixture_module["source"], actor="tester")
    orch.promote(MODULE_ID, actor="tester")
    orch.rollback(MODULE_ID, reason="manual", actor="tester")
    events = store.data["auditEvents"]
    actions = [ev["action"] for ev in events]
    for expected in ("rebuild.started", "rebuild.passed", "rebuild.gray",
                     "rebuild.rolled_back"):
        assert expected in actions, f"缺审计事件 {expected}"
    assert verify_audit_chain(events)["ok"] is True
    for ev in events:                                   # REBUILD 事件结构合法
        assert ev["category"] == "REBUILD"
        assert ev["target"]["type"] == "REBUILD"
        assert ev["power"] == "P1"


# ---------------- T6：安全扫描（静态规则） ----------------
def test_rebuild_safety_scan_blocks_non_regenerable(tmp_path):
    src = tmp_path / "fixture_no_regen.py"
    src.write_text(
        "# 普通模块（moduleId: fixture-no-regen，非再生模块）\n"
        "def f() -> int:\n    return 1\n",
        encoding="utf-8",
    )
    store = _Store()
    orch = RebuildOrchestrator(store, repo_root=str(ROOT))
    report = orch.start_rebuild("fixture-no-regen", golden_tests=[str(tmp_path / "t.py")],
                                source_path=str(src), actor="tester")
    assert report["status"] == "FAILED", report
    assert report["safety"]["ok"] is False
    assert any(f["rule"] == "S1" for f in report["safety"]["findings"])
    assert "rebuild.failed" in _rebuild_actions(store)


def test_rebuild_safety_scan_warns_on_dangerous_primitives(tmp_path):
    src = tmp_path / "fixture_danger.py"
    src.write_text(
        "# （moduleId: fixture-danger, 可重生 ✅）\n"
        "def f(x):\n    return eval(x)\n",
        encoding="utf-8",
    )
    report = static_safety_scan({"moduleId": "fixture-danger", "path": str(src)},
                                repo_root=str(ROOT))
    assert report["ok"] is True                        # 告警不阻断
    assert "S2" in {f["rule"] for f in report["findings"]}


def test_rebuild_safety_scan_flags_unregistered_world_write(tmp_path):
    src = tmp_path / "fixture_write.py"
    src.write_text(
        "# （moduleId: fixture-write, 可重生 ✅）\n"
        "def f(world):\n    world['orders'] = []\n",
        encoding="utf-8",
    )
    report = static_safety_scan({"moduleId": "fixture-write", "path": str(src)},
                                repo_root=str(ROOT))
    assert "S4" in {f["rule"] for f in report["findings"]}
    # 经门禁/登记标记的写路径不算未登记
    src2 = tmp_path / "fixture_write_gated.py"
    src2.write_text(
        "# （moduleId: fixture-write-gated, 可重生 ✅）\n"
        "def f(world):\n"
        "    from server.agent_core import harness\n"
        "    harness.guard('order.upsert', {}, lambda: world.__setitem__('orders', []))\n",
        encoding="utf-8",
    )
    report2 = static_safety_scan({"moduleId": "fixture-write-gated", "path": str(src2)},
                                 repo_root=str(ROOT))
    assert not any(f["rule"] == "S4" for f in report2["findings"])


# ---------------- 隔离运行：APS_* 指向临时目录 ----------------
def test_run_isolated_golden_uses_tmp_env(tmp_path):
    probe = tmp_path / "test_probe_env.py"
    probe.write_text(
        "import os\n"
        "def test_env_isolated() -> None:\n"
        "    wp = os.environ.get('APS_WORLD_PATH', '')\n"
        "    assert 'aps-rebuild-' in wp, wp\n"
        "    assert os.environ.get('APS_DB_PATH', '')\n",
        encoding="utf-8",
    )
    result = run_isolated_golden([str(probe)], repo_root=str(ROOT))
    assert result["passed"] is True, result
    assert "aps-rebuild-" in result["env"]["APS_WORLD_PATH"]


# ---------------- T7：网关 /api/rebuild/* 接线 ----------------
def test_rebuild_gateway_endpoints_wired(monkeypatch):
    import server.gateway.app as gateway_module
    from fastapi.testclient import TestClient
    from tests.auth_provider import install_test_auth

    store = _Store()
    monkeypatch.setattr(gateway_module, "get_store", lambda: store)
    install_test_auth(monkeypatch, "tenant-rebuild-api")
    client = TestClient(gateway_module.create_app())
    login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
    assert login.status_code == 200, login.text
    # 未登记 moduleId → 404 显式失败（不触发子进程）
    r = client.post("/api/rebuild/fixture-no-such-module-xyz")
    assert r.status_code == 404, r.text
    assert "未登记 moduleId" in r.json()["detail"]
    # 状态列表：空流水线占位结构
    g = client.get("/api/rebuild")
    assert g.status_code == 200, g.text
    body = g.json()
    assert "candidates" in body and "activeVersions" in body and "baselines" in body
    # 未登记模块的转正/回滚 → 404
    assert client.post("/api/rebuild/fixture-no-such-module-xyz/promote").status_code == 404
    assert client.post("/api/rebuild/fixture-no-such-module-xyz/rollback").status_code == 404