from __future__ import annotations import json import unittest from pathlib import Path from mais_humana.cli import main from mais_humana.mcp_publication_gate import ( DEFAULT_GATEWAY_TOOLS, GateStatus, LiveToolProbe, ProbeStatus, build_alias_policy, build_not_run_probe, build_publication_gate_report, build_wrangler_runner_evidence, decide_publication_orders, publication_gate_csv, publication_gate_markdown, run_publication_gate, scan_gateway_source, ) from tests.helpers import make_tmp class McpPublicationGateTests(unittest.TestCase): def make_mcp_repo(self, root: Path) -> Path: repo = root / "tudo-para-ia-mcps-internos-plataform" manager = repo / "deploy" / "mcps-gateway" / "src" / "managers" / "gpt-tooling-manager.ts" catalog = repo / "deploy" / "mcps-gateway" / "src" / "providers" / "managed-catalog.ts" managed = repo / "deploy" / "mcps-gateway" / "src" / "managers" / "managed-provider-manager.ts" tests = repo / "tests" / "platform" / "mcps-gateway-mais-humana-provider.test.ts" for path in (manager, catalog, managed, tests): path.parent.mkdir(parents=True, exist_ok=True) manager.write_text( "\n".join( [ 'const PLATFORM_IDS = { maisHumana: "tudo-para-ia-mais-humana-platform" };', "const MAIS_HUMANA_PROVIDER_ID = \"mais_humana\";", "async function executeMaisHumanaGatewayTool() { return true; }", "const ids = ['mais_humana.rulebook.compact', 'mais_humana.admin_ui.same_source', 'mais_humana.mcp_transit.ledger', 'mais_humana.admin_routes.acceptance'];", ] ), encoding="utf-8", ) catalog.write_text( "\n".join( [ "const MAIS_HUMANA_CATALOG = { providerId: 'mais_humana' };", "export const tools = ['mais_humana.rulebook.compact', 'mais_humana.admin_ui.same_source', 'mais_humana.mcp_transit.ledger', 'mais_humana.admin_routes.acceptance'];", ] ), encoding="utf-8", ) managed.write_text( "\n".join( [ "case 'mais_humana:rulebook.compact': return {};", "case 'mais_humana:admin_ui.same_source': return {};", "case 'mais_humana:mcp_transit.ledger': return {};", "case 'mais_humana:admin_routes.acceptance': return {};", ] ), encoding="utf-8", ) tests.write_text("test('Mais Humana tools are published in the canonical gateway catalog', () => {});\n", encoding="utf-8") return repo def test_scan_gateway_source_detects_provider_and_tests(self) -> None: repo = self.make_mcp_repo(make_tmp()) evidence = scan_gateway_source(repo) self.assertTrue(evidence.ready) self.assertEqual(evidence.provider_id, "mais_humana") self.assertEqual(evidence.owner_platform_id, "tudo-para-ia-mais-humana-platform") self.assertEqual(set(evidence.tool_ids), set(DEFAULT_GATEWAY_TOOLS)) self.assertTrue(evidence.source_hash) def test_wrangler_evidence_splits_auth_from_runner_eperm(self) -> None: raw = """ wrangler 4.67.1 Error: spawn EPERM Getting User settings... You are logged in with an Account API Token, associated with the account PARA-IA. Account ID 8dd48b8e7c2757b178283ce19872482d """ evidence = build_wrangler_runner_evidence(raw) self.assertTrue(evidence.authenticated) self.assertFalse(evidence.deploy_dry_run_ok) self.assertEqual(evidence.status, GateStatus.PARTIAL) self.assertIn("runner_node_esbuild_spawn_eperm", evidence.blockers) def test_wrangler_not_attempted_is_observation_not_auth_blocker(self) -> None: evidence = build_wrangler_runner_evidence("", attempted=False) self.assertEqual(evidence.status, GateStatus.NOT_RUN) self.assertFalse(evidence.authenticated) self.assertEqual(evidence.blockers, ()) def test_live_probe_requires_complete_mcp_transit_envelope(self) -> None: probe = LiveToolProbe( tool_id="mais_humana.admin_routes.acceptance", endpoint="https://mcps-gateway.ami-app.workers.dev/v1/execute", http_status=200, status=ProbeStatus.OK, ok=True, error_code="", trace_id="trace-live", audit_id="audit-live", evidence_id="evidence-live", source_payload_hash="a" * 64, source_records_hash="b" * 64, transit_fields_present=("origin", "destination", "tool", "actor", "permission", "traceId", "auditId", "timestamp"), missing_transit_fields=("payload", "result"), response_excerpt={"ok": "True"}, observed_at="2026-05-02T00:00:00+00:00", ) self.assertFalse(probe.live_ready) def test_order_decisions_keep_live_and_git_blockers_explicit(self) -> None: repo = self.make_mcp_repo(make_tmp()) gateway = scan_gateway_source(repo) wrangler = build_wrangler_runner_evidence("wrangler 4.67.1\nError: spawn EPERM\nPARA-IA 8dd48b8e7c2757b178283ce19872482d\n") probes = tuple(build_not_run_probe(tool, "unit test") for tool in DEFAULT_GATEWAY_TOOLS) alias = build_alias_policy(repo_remote="https://git.ami.app.br/admin/tudo-para-ia-mais-humana.git") decisions = decide_publication_orders( gateway, wrangler, probes, alias, git_sync_status="fetch/push falhou por SEC_E_NO_CREDENTIALS", central_write_ok=True, ) by_id = {item.order_id: item for item in decisions} self.assertEqual( by_id["0033_EXECUTIVA__sincronizar-git-mais-humana-mcps-central-com-credenciais"].status, GateStatus.BLOCKED, ) self.assertEqual( by_id["0034_EXECUTIVA__corrigir-acl-escrita-central-e-sql-semantico-plataforma-15"].status, GateStatus.PASSED, ) self.assertEqual( by_id["0047_GERENCIAL__decidir-nome-canonico-e-politica-alias-mais-humana"].status, GateStatus.PASSED, ) def test_publication_gate_report_serializes_markdown_and_csv(self) -> None: tmp = make_tmp() repo = self.make_mcp_repo(tmp) project = tmp / "tudo-para-ia-mais-humana" probes = tuple(build_not_run_probe(tool, "unit test") for tool in DEFAULT_GATEWAY_TOOLS) report = build_publication_gate_report( project_root=project, mcp_repo_root=repo, wrangler_raw_summary="wrangler 4.67.1\nError: spawn EPERM\nPARA-IA 8dd48b8e7c2757b178283ce19872482d\n", git_sync_status="push falhou por SEC_E_NO_CREDENTIALS", repo_remote="https://git.ami.app.br/admin/tudo-para-ia-mais-humana.git", live_probe_payloads=probes, central_write_ok=True, ) payload = report.to_dict() self.assertEqual(payload["provider_id"], "mais_humana") self.assertEqual(payload["localReady"], True) self.assertIn("runner_node_esbuild_spawn_eperm", payload["blockers"]) markdown = publication_gate_markdown(report) csv_text = publication_gate_csv(report) self.assertIn("Gate de publicacao MCP Mais Humana", markdown) self.assertIn("0031_EXECUTIVA", csv_text) self.assertNotIn("cfat_", markdown.lower()) def test_run_publication_gate_writes_project_and_central_artifacts(self) -> None: tmp = make_tmp() repo = self.make_mcp_repo(tmp) project = tmp / "tudo-para-ia-mais-humana" central = tmp / "central" / "projects" / "15_repo_tudo-para-ia-mais-humana-platform" report, records = run_publication_gate( project_root=project, mcp_repo_root=repo, central_platform_folder=central, wrangler_raw_summary="wrangler 4.67.1\nError: spawn EPERM\nPARA-IA 8dd48b8e7c2757b178283ce19872482d\n", git_sync_status="fetch/push falhou por SEC_E_NO_CREDENTIALS", repo_remote="https://git.ami.app.br/admin/tudo-para-ia-mais-humana.git", live_probe=False, ) self.assertTrue((project / "dados" / "mcp-publication-gate-mais-humana.json").exists()) self.assertTrue((project / "matrizes" / "mcp-publication-gate-decisions.csv").exists()) self.assertTrue((central / "reports" / "executivos" / "MCP-PUBLICATION-GATE-MAIS-HUMANA__RODADA015.md").exists()) self.assertGreaterEqual(len(records), 4) self.assertTrue(all(probe.status == ProbeStatus.NOT_RUN for probe in report.live_probes)) def test_cli_publication_gate_writes_json_payload(self) -> None: tmp = make_tmp() repo = self.make_mcp_repo(tmp) project = tmp / "tudo-para-ia-mais-humana" code = main( [ "mcp-publication-gate", "--project-root", str(project), "--mcp-repo-root", str(repo), "--wrangler-summary", "wrangler 4.67.1 Error: spawn EPERM PARA-IA 8dd48b8e7c2757b178283ce19872482d", "--git-sync-status", "push falhou por SEC_E_NO_CREDENTIALS", "--repo-remote", "https://git.ami.app.br/admin/tudo-para-ia-mais-humana.git", ] ) self.assertEqual(code, 0) payload = json.loads((project / "dados" / "mcp-publication-gate-mais-humana.json").read_text(encoding="utf-8")) self.assertEqual(payload["provider_id"], "mais_humana") self.assertEqual(payload["localReady"], True) if __name__ == "__main__": unittest.main()