286 lines
13 KiB
Python
286 lines
13 KiB
Python
# ============================================================
|
||
# 可重生算法流水线黄金测试（矩阵 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
|