from __future__ import annotations import pytest from app.core.mcp.context import ContextService from app.core.mcp.identity import AgentIdentity class Repository: def list_semantic_property_context(self, property_uid): assert property_uid == "property-1" return [ { "path_uid": "path-1", "property_uid": property_uid, "ontology_uid": "ontology-1", "ontology_status": "published", "business_domain_uid": "sales", "data_element_uid": "element-1", "field_uid": "field-1", "evidence_uid": "evidence-1", "token": "must-not-leak", }, { "path_uid": "path-2", "property_uid": property_uid, "ontology_uid": "ontology-2", "ontology_status": "draft", "business_domain_uid": "sales", }, ] def identity(domains=frozenset({"sales"})): return AgentIdentity( subject="agent-viewer", roles=frozenset({"viewer"}), business_domains=domains, environments=frozenset({"test"}), correlation_id="correlation-1", ) def test_mcp_context_returns_only_published_secret_free_scoped_semantics(): result = ContextService(Repository()).get_semantic_property_context( identity(), "property-1", business_domain="sales", limit=10 ) assert result["count"] == 1 assert result["items"][0]["evidence_uid"] == "evidence-1" assert "must-not-leak" not in repr(result) assert "token" not in repr(result).lower() def test_mcp_context_rejects_cross_domain_and_unbounded_limit(): service = ContextService(Repository()) with pytest.raises(PermissionError): service.get_semantic_property_context(identity(), "property-1", business_domain="hr") with pytest.raises(ValueError, match="limit"): service.get_semantic_property_context(identity(), "property-1", business_domain="sales", limit=51)