93 lines
4.2 KiB
Python
93 lines
4.2 KiB
Python
|
# ============================================================
|
|||
|
|
# 可重生流水线注册表黄金测试（矩阵 91，方向 T）：
|
|||
|
|
# 候选版本登记 / 激活版本指针 / 基线 / 回滚恢复（world["rebuildRegistry"]）。
|
|||
|
|
# ============================================================
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from server.agent_core.registry import (
|
|||
|
|
active_version,
|
|||
|
|
baseline_version,
|
|||
|
|
get_rebuild_candidate,
|
|||
|
|
list_rebuild_candidates,
|
|||
|
|
rebuild_registry,
|
|||
|
|
register_rebuild_candidate,
|
|||
|
|
rollback_registry,
|
|||
|
|
set_active_version,
|
|||
|
|
set_baseline_version,
|
|||
|
|
verify_audit_chain,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_registry_initializes_on_first_access():
|
|||
|
|
world: dict = {}
|
|||
|
|
reg = rebuild_registry(world)
|
|||
|
|
assert reg["candidates"] == []
|
|||
|
|
assert reg["activeVersions"] == {}
|
|||
|
|
assert reg["baselines"] == {}
|
|||
|
|
assert world["rebuildRegistry"] is reg # 落于 world 状态，随 WorldStore 持久化
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_register_candidate_sets_baseline_and_lists():
|
|||
|
|
world: dict = {}
|
|||
|
|
c1 = {"moduleId": "domain-orders", "version": "domain-orders-v1", "status": "GRAY"}
|
|||
|
|
register_rebuild_candidate(world, "domain-orders", c1)
|
|||
|
|
assert get_rebuild_candidate(world, "domain-orders") == c1
|
|||
|
|
assert baseline_version(world, "domain-orders") == "current" # 无激活指针 → 现场源码基线
|
|||
|
|
assert active_version(world, "domain-orders") is None
|
|||
|
|
# 同一 moduleId 再次登记 → 取最新；基线保持首个
|
|||
|
|
c2 = {"moduleId": "domain-orders", "version": "domain-orders-v2", "status": "BUILDING"}
|
|||
|
|
register_rebuild_candidate(world, "domain-orders", c2)
|
|||
|
|
assert get_rebuild_candidate(world, "domain-orders") == c2
|
|||
|
|
assert len(list_rebuild_candidates(world)) == 2
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_active_version_pointer_and_baseline_update():
|
|||
|
|
world: dict = {}
|
|||
|
|
set_baseline_version(world, "domain-orders", "v0")
|
|||
|
|
set_active_version(world, "domain-orders", "v1")
|
|||
|
|
assert active_version(world, "domain-orders") == "v1"
|
|||
|
|
assert baseline_version(world, "domain-orders") == "v0"
|
|||
|
|
# 新候选登记时若已有基线则不覆盖
|
|||
|
|
register_rebuild_candidate(world, "domain-orders",
|
|||
|
|
{"moduleId": "domain-orders", "version": "v2", "status": "GRAY"})
|
|||
|
|
assert baseline_version(world, "domain-orders") == "v0"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_rollback_registry_restores_active_pointer():
|
|||
|
|
world: dict = {}
|
|||
|
|
cand = {"moduleId": "domain-orders", "version": "v2", "status": "ACTIVE"}
|
|||
|
|
set_baseline_version(world, "domain-orders", "v1")
|
|||
|
|
set_active_version(world, "domain-orders", "v2")
|
|||
|
|
register_rebuild_candidate(world, "domain-orders", cand)
|
|||
|
|
result = rollback_registry(world, "domain-orders", reason="auto:golden")
|
|||
|
|
assert result["restoredTo"] == "v1"
|
|||
|
|
assert result["previousActive"] == "v2"
|
|||
|
|
assert active_version(world, "domain-orders") == "v1"
|
|||
|
|
assert cand["status"] == "ROLLED_BACK"
|
|||
|
|
assert cand["rollbackReason"] == "auto:golden"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_rollback_registry_without_active_pointer_restores_current():
|
|||
|
|
world: dict = {}
|
|||
|
|
register_rebuild_candidate(world, "domain-orders",
|
|||
|
|
{"moduleId": "domain-orders", "version": "v1", "status": "FAILED"})
|
|||
|
|
result = rollback_registry(world, "domain-orders")
|
|||
|
|
assert result["restoredTo"] == "current"
|
|||
|
|
assert active_version(world, "domain-orders") == "current"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_registry_functions_coexist_with_audit_chain():
|
|||
|
|
"""注册表操作不破坏审计链（独立分区，互不污染）。"""
|
|||
|
|
from server.agent_core.audit import write_audit
|
|||
|
|
world: dict = {}
|
|||
|
|
register_rebuild_candidate(world, "domain-orders",
|
|||
|
|
{"moduleId": "domain-orders", "version": "v1", "status": "GRAY"})
|
|||
|
|
write_audit(world, lambda kind: 1, actor="tester", category="REBUILD",
|
|||
|
|
action="rebuild.started", target={"type": "REBUILD", "id": "domain-orders"},
|
|||
|
|
power="P1", rationale={"stage": "build"})
|
|||
|
|
write_audit(world, lambda kind: 2, actor="tester", category="REBUILD",
|
|||
|
|
action="rebuild.passed", target={"type": "REBUILD", "id": "domain-orders"},
|
|||
|
|
power="P1", rationale={"stage": "shadow"})
|
|||
|
|
assert verify_audit_chain(world["auditEvents"])["ok"] is True
|
|||
|
|
assert world["rebuildRegistry"]["candidates"][0]["version"] == "v1"
|