165 lines
6.0 KiB
Python
165 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
# ruff: noqa: E402
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
|
|
|
|
from src.api.v1.mcp_control_plane import router
|
|
from src.services.mcp_control_plane_service import (
|
|
build_mcp_control_plane_overview,
|
|
load_mcp_control_plane_catalog,
|
|
)
|
|
from src.services.product_governance_matrix_readback import (
|
|
load_latest_product_governance_matrix_readback,
|
|
)
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
_OPERATIONS_DIR = _REPO_ROOT / "docs" / "operations"
|
|
|
|
|
|
def test_catalog_contains_exact_12_products_and_explicit_missing_rows() -> None:
|
|
catalog = load_mcp_control_plane_catalog(_OPERATIONS_DIR)
|
|
|
|
product_ids = {row["product_id"] for row in catalog["products"]}
|
|
assert len(product_ids) == 12
|
|
assert {"ewoooc", "momo-pro-system"}.issubset(product_ids)
|
|
assert catalog["architecture"]["separate_gateway_allowed"] is False
|
|
assert catalog["architecture"]["separate_rag_database_allowed"] is False
|
|
assert catalog["source_policy"]["github_allowed"] is False
|
|
assert catalog["security_policy"]["external_rag_auto_write_allowed"] is False
|
|
|
|
capabilities = {row["capability_id"]: row for row in catalog["capabilities"]}
|
|
assert capabilities["rejected.github"]["decision"] == "rejected"
|
|
assert capabilities["quarantine.agent-bounty-mcp"]["decision"] == "quarantine"
|
|
assert (
|
|
capabilities["external.smithery-site-audit-shadow"]["decision"]
|
|
== "evaluation_shadow_candidate"
|
|
)
|
|
assert (
|
|
capabilities["external.smithery-site-audit-shadow"]["deployment_allowed"]
|
|
is False
|
|
)
|
|
assert (
|
|
capabilities["rejected.smithery-technical-analysis"]["decision"] == "rejected"
|
|
)
|
|
assert (
|
|
"tool_output_contains_cross_tool_prompt_injection"
|
|
in capabilities["quarantine.agent-bounty-mcp"]["blockers"]
|
|
)
|
|
assert not any(
|
|
row["deployment_allowed"]
|
|
for row in catalog["capabilities"]
|
|
if row["source_type"] == "external"
|
|
)
|
|
|
|
|
|
def test_overview_joins_committed_product_matrix_without_runtime_claims() -> None:
|
|
catalog = load_mcp_control_plane_catalog(_OPERATIONS_DIR)
|
|
matrix = load_latest_product_governance_matrix_readback(_OPERATIONS_DIR)
|
|
|
|
payload = build_mcp_control_plane_overview(
|
|
catalog=catalog,
|
|
product_matrix=matrix,
|
|
generated_at="2026-07-15T19:00:00+08:00",
|
|
)
|
|
|
|
assert payload["schema_version"] == "mcp_control_plane_overview_v1"
|
|
assert payload["status"] == "partial_degraded_with_safe_next_action"
|
|
assert payload["summary"]["product_count"] == 12
|
|
assert payload["summary"]["mcp_source_detected_product_count"] == 7
|
|
assert payload["summary"]["external_promotion_ready_count"] == 0
|
|
assert payload["summary"]["github_in_scope_count"] == 0
|
|
assert payload["runtime"]["status"] == "not_requested"
|
|
|
|
products = {row["product_id"]: row for row in payload["products"]}
|
|
assert products["ewoooc"]["canonical_repo"] == "wooo/ewoooc"
|
|
assert products["momo-pro-system"]["canonical_repo"] == "wooo/momo-pro-system"
|
|
assert (
|
|
products["agent-bounty-protocol"]["inventory_state"]
|
|
== "source_detected_quarantined"
|
|
)
|
|
|
|
|
|
def test_catalog_rejects_external_deployment_without_immutable_receipts(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
catalog = load_mcp_control_plane_catalog(_OPERATIONS_DIR)
|
|
external = next(
|
|
row for row in catalog["capabilities"] if row["source_type"] == "external"
|
|
)
|
|
external["deployment_allowed"] = True
|
|
(tmp_path / "mcp-control-plane-catalog.snapshot.json").write_text(
|
|
json.dumps(catalog),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
try:
|
|
load_mcp_control_plane_catalog(tmp_path)
|
|
except ValueError as exc:
|
|
assert "cannot deploy without immutable receipts" in str(exc)
|
|
else:
|
|
raise AssertionError("unsafe external deployment must be rejected")
|
|
|
|
|
|
def test_overview_endpoint_returns_real_committed_catalog_without_runtime() -> None:
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/v1")
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/api/v1/mcp-control-plane/overview?include_runtime=false")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["summary"]["product_count"] == 12
|
|
assert data["runtime"]["status"] == "not_requested"
|
|
assert data["operation_boundaries"]["github_allowed"] is False
|
|
assert (
|
|
data["operation_boundaries"]["public_response_redacted_aggregate_only"] is True
|
|
)
|
|
assert data["operation_boundaries"]["mutation_endpoint_exposed"] is False
|
|
assert (
|
|
data["operation_boundaries"]["operator_auth_required_for_future_mutation"]
|
|
is True
|
|
)
|
|
|
|
|
|
def test_overview_promotes_durable_lifecycle_receipt_not_static_policy() -> None:
|
|
catalog = load_mcp_control_plane_catalog(_OPERATIONS_DIR)
|
|
matrix = load_latest_product_governance_matrix_readback(_OPERATIONS_DIR)
|
|
runtime = {
|
|
"status": "ready",
|
|
"provider_registry": {"provider_count": 9, "tool_count": 39},
|
|
"gateway_audit_24h": {"call_count": 1},
|
|
"rag": {"total_chunks": 1},
|
|
"version_lifecycle": {
|
|
"status": "ready",
|
|
"latest_run": {
|
|
"run_id": "11111111-1111-1111-1111-111111111111",
|
|
"status": "completed_no_write_policy_blocked",
|
|
"fresh": True,
|
|
},
|
|
"blockers": ["external_candidates_not_promotion_ready"],
|
|
},
|
|
}
|
|
|
|
payload = build_mcp_control_plane_overview(
|
|
catalog=catalog,
|
|
product_matrix=matrix,
|
|
runtime=runtime,
|
|
)
|
|
|
|
assert payload["version_lifecycle"]["runtime_state"] == "running_fail_closed"
|
|
assert payload["version_lifecycle"]["runtime"]["status"] == "ready"
|
|
assert "external_candidates_not_promotion_ready" in payload["active_blockers"]
|
|
assert (
|
|
payload["operation_boundaries"]["normalized_version_receipt_write_allowed"]
|
|
is True
|
|
)
|
|
assert payload["operation_boundaries"]["production_upgrade_allowed"] is False
|