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 def test_overview_promotes_two_fresh_federated_receipts_without_claiming_12() -> None: catalog = load_mcp_control_plane_catalog(_OPERATIONS_DIR) matrix = load_latest_product_governance_matrix_readback(_OPERATIONS_DIR) receipts = [ { "product_id": product_id, "canonical_repo": canonical_repo, "runtime_role": runtime_role, "runtime_version": "V10.796", "health_status": "partial_degraded", "mcp": { "enabled": True, "server_count": 3, "server_ids": ["firecrawl", "omnisearch", "postgres"], "caller_count": 2, "tool_count": 9, }, "rag": { "enabled": True, "vector_store": "pgvector", "embedding_model": "bge-m3:latest", "embedding_dim": 1024, "embedding_version_state": "floating_tag_detected", }, "pixelrag": { "enabled": True, "platform_count": 3, "platforms": ["momoshop", "pchome", "shopee"], "visual_rag_stage": "phase1_visual_evidence_receipts", }, "blockers": ["rag_embedding_model_not_immutable"], "fresh": True, } for product_id, canonical_repo, runtime_role in ( ( "ewoooc", "wooo/ewoooc", "mcp_rag_automation_source_plane", ), ( "momo-pro-system", "wooo/momo-pro-system", "commerce_runtime_consumer_plane", ), ) ] runtime = { "status": "ready", "provider_registry": {"provider_count": 9, "tool_count": 39}, "gateway_audit_24h": {"call_count": 1}, "rag": {"total_chunks": 1}, "version_lifecycle": {"status": "pending_first_run", "blockers": []}, "federated_runtime": { "status": "ready", "verified_fresh_receipt_count": 2, "expected_receipt_count": 2, "receipts": receipts, "blockers": [], }, "federated_product_runtime_receipt_count": 2, "federated_product_runtime_receipt_expected": 12, } payload = build_mcp_control_plane_overview( catalog=catalog, product_matrix=matrix, runtime=runtime, ) assert payload["summary"]["federated_product_runtime_receipt_count"] == 2 assert payload["summary"]["federated_product_runtime_receipt_expected"] == 12 assert "cross_product_runtime_federation_receipts_missing" in payload[ "active_blockers" ] products = {row["product_id"]: row for row in payload["products"]} for product_id in ("ewoooc", "momo-pro-system"): product = products[product_id] assert ( product["inventory_state"] == "federated_runtime_readback_partial_degraded" ) assert "awoooi_federated_runtime_readback_missing" not in product["blockers"] assert "rag_embedding_model_not_immutable" in product["blockers"] assert product["federated_runtime_receipt"]["mcp"]["server_ids"] == [ "firecrawl", "omnisearch", "postgres", ] assert payload["next_actions"][0] == ( "resolve_ewoooc_momo_runtime_blockers_and_expand_next_product_federation_receipts" )