from __future__ import annotations import json import unittest from pathlib import Path from mais_humana.catalog import get_platform from mais_humana.cli import main from mais_humana.human_rulebook import MCP_TRANSIT_FIELDS, evaluate_rulebook from mais_humana.matrix import build_matrix, build_platform_reports from mais_humana.mcp_contract import ( CANONICAL_PROJECT_ID, MCP_CONTROL_PLANE_ID, McpContractKind, administration_route_contracts, administration_route_readiness_markdown, administration_route_readiness_payload, build_mcp_contract_report, build_mcp_execute_probe, contracts_for_kind, iter_contracts, mcp_contract_csv, mcp_contract_markdown, mcp_execute_probe_markdown, mcp_provider_payload, same_source_validation_payload, ) from mais_humana.scanner import scan_platform from tests.helpers import make_tmp class McpProviderContractTests(unittest.TestCase): def make_repo(self, root: Path, platform_id: str, text: str) -> None: platform = get_platform(platform_id) repo = root / platform.repo_name repo.mkdir(parents=True) (repo / "README.md").write_text(text, encoding="utf-8") (repo / "src").mkdir() (repo / "src" / "index.ts").write_text( "\n".join( [ "export const panel = 'admin_ui panelReady sameSource sourcePayloadHash sourceRecordsHash';", "export const trace = 'traceId auditId actor permission result timestamp';", "export const docs = 'responseReady readiness health evidence';", ] ) + "\n", encoding="utf-8", ) def make_report(self, root: Path): self.make_repo( root, "business", "business checkout billing entitlements sameSource panelReady responseReady traceId auditId", ) self.make_repo( root, "docs", "docs catalogOnly formal exception contract canonical sourcePayloadHash sourceRecordsHash", ) scans = tuple(scan_platform(root, get_platform(pid)) for pid in ("business", "docs")) cells = build_matrix(scans) reports = build_platform_reports(scans, cells) rulebook = evaluate_rulebook(reports, limit=60) return build_mcp_contract_report(rulebook) def test_generated_contract_registry_is_large_and_mcp_bound(self) -> None: contracts = iter_contracts() self.assertGreaterEqual(len(contracts), 500) self.assertTrue(any(contract.contract_id == "docs.formal-exception.docs-catalogonly" for contract in contracts)) self.assertTrue(any(contract.control_plane_id == MCP_CONTROL_PLANE_ID for contract in contracts)) self.assertTrue(all(tuple(MCP_TRANSIT_FIELDS) == contract.required_transit_fields for contract in contracts)) def test_contract_report_exports_provider_envelope(self) -> None: report = self.make_report(make_tmp()) envelope = mcp_provider_payload(report, limit=5) self.assertEqual(envelope["origin"], "tudo-para-ia-mais-humana") self.assertEqual(envelope["destination"], MCP_CONTROL_PLANE_ID) self.assertIn("traceId", envelope) self.assertIn("auditId", envelope) self.assertIn("evidenceId", envelope["result"]) self.assertEqual(envelope["payload"]["canonicalProjectId"], CANONICAL_PROJECT_ID) self.assertLessEqual(len(envelope["payload"]["contracts"]), 5) def test_contract_markdown_csv_and_same_source_payload_are_stable(self) -> None: report = self.make_report(make_tmp()) markdown = mcp_contract_markdown(report) csv_text = mcp_contract_csv(report) same_source = same_source_validation_payload(report, limit=8) self.assertIn("MCP Provider Mais Humana", markdown) self.assertIn("contract_id,kind,platform_id", csv_text) self.assertTrue(same_source["allSameSource"]) self.assertGreater(same_source["validatedCount"], 0) first = same_source["validations"][0] self.assertIn("sourceEndpoint", first) self.assertIn("sourceToolId", first) self.assertIn("evidenceId", first) def test_execute_probe_redacts_auth_and_links_evidence(self) -> None: report = self.make_report(make_tmp()) probe = build_mcp_execute_probe(report, limit=4, observed_status="http_403") payload = probe.to_dict() self.assertEqual(payload["endpoint"].endswith("/v1/execute"), True) self.assertEqual(payload["request_headers"]["Authorization"], "Bearer ") self.assertEqual(payload["request_body"]["toolId"], "mais_humana.rulebook.compact") self.assertIn("evidence-", payload["evidence_id"]) markdown = mcp_execute_probe_markdown(probe) self.assertNotIn("123456789", markdown) self.assertIn("MCP Execute Probe", markdown) def test_report_model_and_ui_contracts_exist(self) -> None: report_models = contracts_for_kind(McpContractKind.REPORT_MODEL) ui_screens = contracts_for_kind(McpContractKind.UI_SCREEN) self.assertGreater(len(report_models), 100) self.assertGreater(len(ui_screens), 20) self.assertTrue(all(contract.report_model_id for contract in report_models)) def test_access_policy_contracts_cover_profiles_and_gateway_rules(self) -> None: access_contracts = contracts_for_kind(McpContractKind.ACCESS_POLICY) self.assertGreater(len(access_contracts), 500) sample = access_contracts[0] self.assertIn("authorizationCredentialRef", sample.required_payload_fields) self.assertIn("pluginCloudflareDiagnosticIgnored", sample.required_payload_fields) self.assertTrue(any("WAF" in step or "waf" in step for step in sample.validation_steps)) self.assertTrue(all("0045_GERENCIAL__pactuar-politica-acesso-waf-gpt-mcp-gateway" in item.order_ids for item in access_contracts)) def test_administration_route_contracts_cover_mcp_only_operations(self) -> None: routes = administration_route_contracts() self.assertGreater(len(routes), 900) operations = {contract.contract_id.split(".")[-2] for contract in routes} self.assertEqual(operations, {"acao", "auditoria", "consulta", "diagnostico", "explicacao"}) sample = routes[0] self.assertIn("mcpOnlyAdministration", sample.required_payload_fields) self.assertIn("directPlatformBypassBlocked", sample.required_payload_fields) self.assertIn("permissionScope", sample.required_payload_fields) self.assertIn("0044_GERENCIAL__institucionalizar-ledger-transito-mcp-como-gate-release", sample.order_ids) readiness = administration_route_readiness_payload(platform_id=sample.platform_id) self.assertEqual(readiness["kind"], "administration_route") self.assertEqual(set(readiness["operations"].keys()), operations) self.assertTrue(readiness["evidenceId"].startswith("evidence-")) markdown = administration_route_readiness_markdown(readiness) self.assertIn("MCP Administration Routes Readiness", markdown) self.assertIn("Regra MCP-only", markdown) def test_cli_mcp_provider_returns_json(self) -> None: root = make_tmp() self.make_repo( root, "business", "business checkout billing entitlements sameSource panelReady responseReady traceId auditId", ) code = main(["mcp-provider", "--ecosystem-root", str(root), "--limit", "3"]) self.assertEqual(code, 0) probe_code = main(["mcp-execute-probe", "--ecosystem-root", str(root), "--limit", "3", "--observed-status", "prepared"]) self.assertEqual(probe_code, 0) def test_cli_mcp_contract_artifacts_writes_focused_outputs(self) -> None: root = make_tmp() project = root / "tudo-para-ia-mais-humana" self.make_repo( root, "business", "business checkout billing entitlements sameSource panelReady responseReady traceId auditId", ) code = main(["mcp-contract-artifacts", "--ecosystem-root", str(root), "--project-root", str(project)]) self.assertEqual(code, 0) provider = json.loads((project / "dados" / "mcp-provider-mais-humana-compacto.json").read_text(encoding="utf-8")) admin_routes = json.loads((project / "dados" / "mcp-administration-routes-readiness.json").read_text(encoding="utf-8")) self.assertGreaterEqual(provider["contractsCount"], 2000) self.assertGreaterEqual(admin_routes["contractsCount"], 900) self.assertTrue((project / "ecossistema" / "MCP-ADMINISTRATION-ROUTES.md").exists()) def test_provider_payload_is_serializable(self) -> None: report = self.make_report(make_tmp()) payload = mcp_provider_payload(report, limit=3) text = json.dumps(payload, ensure_ascii=False) self.assertIn("sourcePayloadHash", text) if __name__ == "__main__": unittest.main()