69 lines
2.7 KiB
Python
69 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from dataclasses import replace
|
|
|
|
from mais_humana.cli import main
|
|
from mais_humana.mcp_contract import build_mcp_contract_report
|
|
from mais_humana.mcp_transit_ledger import (
|
|
FindingSeverity,
|
|
build_mcp_transit_ledger,
|
|
mcp_transit_csv,
|
|
mcp_transit_ledger_compact_json,
|
|
mcp_transit_markdown,
|
|
validate_transit_record,
|
|
)
|
|
from tests.helpers import make_tmp
|
|
|
|
|
|
class McpTransitLedgerTests(unittest.TestCase):
|
|
def test_ledger_builds_complete_mcp_envelopes_from_contracts(self) -> None:
|
|
contract_report = build_mcp_contract_report(limit=12)
|
|
ledger = build_mcp_transit_ledger(contract_report)
|
|
|
|
self.assertEqual(ledger.records_count, 12)
|
|
self.assertGreaterEqual(ledger.ready_count, 1)
|
|
self.assertTrue(ledger.passed)
|
|
for record in ledger.records:
|
|
self.assertEqual(record.destination, contract_report.control_plane_id)
|
|
self.assertEqual(tuple(record.envelope.keys()), ledger.required_fields)
|
|
self.assertTrue(record.traceId.startswith("trace-"))
|
|
self.assertTrue(record.auditId.startswith("audit-"))
|
|
self.assertTrue(record.evidenceId.startswith("evidence-"))
|
|
|
|
def test_validation_rejects_direct_route_and_secret_shaped_payload(self) -> None:
|
|
ledger = build_mcp_transit_ledger(build_mcp_contract_report(limit=1))
|
|
original = ledger.records[0]
|
|
unsafe = replace(
|
|
original,
|
|
destination="tudo-para-ia-business-platform",
|
|
payload={**original.payload, "cloudflare_token": "cfat_abcdefghijklmnopqrstuvwxyz1234567890"},
|
|
)
|
|
|
|
findings = validate_transit_record(unsafe, control_plane_id=ledger.control_plane_id)
|
|
blocker_fields = {item.field for item in findings if item.severity == FindingSeverity.BLOCKER}
|
|
|
|
self.assertIn("destination", blocker_fields)
|
|
self.assertIn("payload", blocker_fields)
|
|
|
|
def test_markdown_csv_and_compact_payload_are_useful(self) -> None:
|
|
ledger = build_mcp_transit_ledger(build_mcp_contract_report(limit=5))
|
|
|
|
compact = mcp_transit_ledger_compact_json(ledger, limit=3)
|
|
markdown = mcp_transit_markdown(ledger)
|
|
csv_text = mcp_transit_csv(ledger)
|
|
|
|
self.assertEqual(compact["recordsCount"], 5)
|
|
self.assertEqual(len(compact["records"]), 3)
|
|
self.assertIn("MCP Transit Ledger", markdown)
|
|
self.assertIn("record_id,contract_id,platform_id", csv_text)
|
|
|
|
def test_cli_mcp_transit_ledger_prints_json(self) -> None:
|
|
tmp = make_tmp()
|
|
code = main(["mcp-transit-ledger", "--ecosystem-root", str(tmp), "--limit", "2"])
|
|
self.assertEqual(code, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|