163 lines
7.7 KiB
Python
163 lines
7.7 KiB
Python
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_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 <redacted>")
|
|
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-"))
|
|
|
|
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_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()
|