feat(governance): 新增全產品 Code Review 防木馬 Gate
Some checks failed
Code Review / ai-code-review (push) Successful in 13s
CD Pipeline / tests (push) Successful in 1m49s
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled

This commit is contained in:
Your Name
2026-06-19 00:26:04 +08:00
parent 9ebab2db6e
commit 4a14860c60
9 changed files with 1496 additions and 0 deletions

View File

@@ -0,0 +1,199 @@
from __future__ import annotations
import json
import pytest
from src.services.product_code_review_gate import load_latest_product_code_review_gate
def test_load_latest_product_code_review_gate_enriches_tenant_inventory(tmp_path):
_write_snapshot(tmp_path, _snapshot())
loaded = load_latest_product_code_review_gate(tmp_path)
assert loaded["schema_version"] == "product_code_review_gate_v1"
assert loaded["program_status"]["current_task_id"] == "P2-111"
assert loaded["program_status"]["next_task_id"] == "P2-112"
assert loaded["program_status"]["read_only_mode"] is True
assert loaded["rollups"]["product_scope_count"] == len(loaded["product_review_matrix"]) >= 16
assert loaded["tenant_asset_inventory_summary"]["public_route_count"] >= 31
assert loaded["rollups"]["source_candidate_repo_count"] == 10
assert loaded["gate_boundaries"]["auto_merge_allowed"] is False
assert loaded["gate_boundaries"]["aider_auto_patch_allowed"] is False
assert loaded["gate_boundaries"]["elephantalpha_write_allowed"] is False
assert all(item["runtime_gate_count"] == 0 for item in loaded["product_review_matrix"])
assert all(item["action_button_count"] == 0 for item in loaded["product_review_matrix"])
assert any(item["product_id"] == "PRD-001" for item in loaded["product_review_matrix"])
assert any(tool["tool_id"] == "codeql" for tool in loaded["mainstream_tool_lanes"])
assert any(tool["tool_id"] == "gitleaks" for tool in loaded["mainstream_tool_lanes"])
def test_product_code_review_gate_requires_read_only_mode(tmp_path):
snapshot = _snapshot()
snapshot["program_status"]["read_only_mode"] = False
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="read_only_mode"):
load_latest_product_code_review_gate(tmp_path)
def test_product_code_review_gate_blocks_auto_merge(tmp_path):
snapshot = _snapshot()
snapshot["gate_boundaries"]["auto_merge_allowed"] = True
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="gate boundaries"):
load_latest_product_code_review_gate(tmp_path)
def test_product_code_review_gate_requires_rollup_consistency(tmp_path):
snapshot = _snapshot()
snapshot["rollups"]["pre_deploy_gate_count"] = 99
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="pre_deploy_gate_count"):
load_latest_product_code_review_gate(tmp_path)
def test_product_code_review_gate_requires_gate_evidence(tmp_path):
snapshot = _snapshot()
snapshot["pre_deploy_gates"][0]["evidence_refs"] = []
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="evidence_refs"):
load_latest_product_code_review_gate(tmp_path)
def test_product_code_review_gate_requires_reviewer_no_write(tmp_path):
snapshot = _snapshot()
snapshot["ai_reviewer_lanes"][0]["write_allowed"] = True
_write_snapshot(tmp_path, snapshot)
with pytest.raises(ValueError, match="write_allowed"):
load_latest_product_code_review_gate(tmp_path)
def test_product_code_review_gate_fails_when_missing(tmp_path):
with pytest.raises(FileNotFoundError):
load_latest_product_code_review_gate(tmp_path)
def _write_snapshot(tmp_path, payload: dict) -> None:
(tmp_path / "product_code_review_gate_2026-06-19.json").write_text(
json.dumps(payload),
encoding="utf-8",
)
def _snapshot() -> dict:
return {
"schema_version": "product_code_review_gate_v1",
"generated_at": "2026-06-19T00:42:00+08:00",
"program_status": {
"overall_completion_percent": 100,
"current_priority": "P2",
"current_task_id": "P2-111",
"next_task_id": "P2-112",
"read_only_mode": True,
"runtime_authority": "repo_only",
"status_note": "只讀 Code Review Gate。",
},
"source_refs": [".gitea/workflows/code-review.yaml"],
"rollups": {
"product_scope_count": 16,
"public_route_count_minimum": 31,
"source_candidate_repo_count": 10,
"pre_deploy_gate_count": 1,
"post_deploy_gate_count": 1,
"ai_reviewer_count": 2,
"mainstream_tool_count": 2,
"owner_review_required_count": 1,
"critical_gap_count": 1,
"blocked_operation_count": 17,
"active_write_gate_count": 0,
"action_button_count": 0,
},
"pre_deploy_gates": [_gate("pre", "hermes")],
"post_deploy_gates": [_gate("post", "nemotron")],
"ai_reviewer_lanes": [
_reviewer("hermes"),
_reviewer("aider"),
],
"mainstream_tool_lanes": [
_tool("codeql"),
_tool("gitleaks"),
],
"decision_matrix": [
{
"risk_lane": "low",
"reviewer": "Hermes",
"aider_role": "draft",
"required_gate": "owner",
"post_deploy": "smoke",
}
],
"gate_boundaries": {
"read_only_api_allowed": True,
"workflow_write_allowed": False,
"external_scanner_activation_allowed": False,
"paid_ai_review_allowed": False,
"repo_app_install_allowed": False,
"auto_merge_allowed": False,
"production_deploy_authorized": False,
"aider_auto_patch_allowed": False,
"elephantalpha_write_allowed": False,
"secret_read_allowed": False,
"post_deploy_write_allowed": False,
"runtime_execution_allowed": False,
"telegram_send_allowed": False,
"gateway_queue_write_allowed": False,
"host_probe_allowed": False,
"registry_push_allowed": False,
"artifact_signing_allowed": False,
"action_buttons_allowed": False,
},
"next_actions": [
{
"task_id": "P2-112",
"priority": "P1",
"summary": "下一步",
"gate": "owner",
}
],
}
def _gate(gate_id: str, owner_agent: str) -> dict:
return {
"gate_id": gate_id,
"label": gate_id,
"coverage": "repo",
"status": "wired",
"owner_agent": owner_agent,
"evidence_refs": ["docs/LOGBOOK.md"],
"current_gap": "gap",
"next_action": "review",
}
def _reviewer(agent_id: str) -> dict:
return {
"agent_id": agent_id,
"label": agent_id,
"role": "review",
"allowed_output": "packet",
"write_allowed": False,
}
def _tool(tool_id: str) -> dict:
return {
"tool_id": tool_id,
"label": tool_id,
"category": "scanner",
"source_url": "https://example.com",
"integration_status": "candidate",
"recommended_role": "scan",
"blocked_now": ["external lookup"],
}

View File

@@ -0,0 +1,38 @@
from __future__ import annotations
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1.agents import router
def test_product_code_review_gate_endpoint_returns_all_product_gate():
app = FastAPI()
app.include_router(router, prefix="/api/v1")
client = TestClient(app)
response = client.get("/api/v1/agents/product-code-review-gate")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "product_code_review_gate_v1"
assert data["program_status"]["current_task_id"] == "P2-111"
assert data["program_status"]["next_task_id"] == "P2-112"
assert data["program_status"]["read_only_mode"] is True
assert data["rollups"]["product_scope_count"] == len(data["product_review_matrix"]) >= 16
assert data["tenant_asset_inventory_summary"]["public_route_count"] >= 31
assert data["rollups"]["pre_deploy_gate_count"] == len(data["pre_deploy_gates"]) == 8
assert data["rollups"]["post_deploy_gate_count"] == len(data["post_deploy_gates"]) == 6
assert data["rollups"]["ai_reviewer_count"] == len(data["ai_reviewer_lanes"]) == 5
assert data["rollups"]["mainstream_tool_count"] == len(data["mainstream_tool_lanes"]) == 9
assert data["rollups"]["active_write_gate_count"] == 0
assert data["rollups"]["action_button_count"] == 0
assert data["gate_boundaries"]["auto_merge_allowed"] is False
assert data["gate_boundaries"]["aider_auto_patch_allowed"] is False
assert data["gate_boundaries"]["external_scanner_activation_allowed"] is False
assert data["gate_boundaries"]["artifact_signing_allowed"] is False
assert any(product["product_id"] == "PRD-001" for product in data["product_review_matrix"])
assert any(product["product_id"] == "PRD-006" for product in data["product_review_matrix"])
assert any(tool["tool_id"] == "codeql" for tool in data["mainstream_tool_lanes"])
assert any(tool["tool_id"] == "semgrep" for tool in data["mainstream_tool_lanes"])
assert any(tool["tool_id"] == "osv_scanner" for tool in data["mainstream_tool_lanes"])