| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859 |
- 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)
|