| 1 | from __future__ import annotations |
| 2 | |
| 3 | import asyncio |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from app.contracts import ( |
| 8 | McpAuthContext, |
| 9 | McpAuthenticationType, |
| 10 | McpRiskClass, |
| 11 | McpToolDefinition, |
| 12 | McpToolExecution, |
| 13 | McpToolKind, |
| 14 | StrictContract, |
| 15 | ) |
| 16 | from app.invocation import McpInvocationService |
| 17 | from app.policy import McpToolPolicy |
| 18 | from app.registry import McpToolRegistry |
| 19 | from conftest import INSTRUMENT_ID, WATCHLIST_ID |
| 20 | |
| 21 | |
| 22 | async def invoke(container, tool, arguments, request_id="request-5a"): |
| 23 | return await container.invocation.invoke( |
| 24 | tool, arguments, container.auth, request_id=request_id |
| 25 | ) |
| 26 | |
| 27 | |
| 28 | @pytest.mark.asyncio |
| 29 | async def test_unknown_tool_is_rejected_and_audited(container, audit) -> None: |
| 30 | result = await invoke(container, "does_not_exist", {}) |
| 31 | assert result["error"]["code"] == "MCP_TOOL_NOT_FOUND" |
| 32 | assert [event["event"] for event in audit.events] == ["MCP_TOOL_INVOKED", "MCP_TOOL_FAILED"] |
| 33 | |
| 34 | |
| 35 | @pytest.mark.asyncio |
| 36 | async def test_global_instrument_id_is_required_and_fuzzy_identity_is_rejected(container, reader) -> None: |
| 37 | missing = await invoke(container, "get_research_readiness", {}) |
| 38 | fuzzy = await invoke( |
| 39 | container, "get_research_readiness", {"globalInstrumentId": "Reliance Industries"} |
| 40 | ) |
| 41 | assert missing["error"]["code"] == "INVALID_ARGUMENT" |
| 42 | assert fuzzy["error"]["code"] == "INVALID_ARGUMENT" |
| 43 | assert reader.calls == [] |
| 44 | |
| 45 | |
| 46 | @pytest.mark.asyncio |
| 47 | async def test_non_held_company_is_supported_without_portfolio_fields(container, reader) -> None: |
| 48 | result = await invoke( |
| 49 | container, |
| 50 | "get_financial_facts", |
| 51 | {"globalInstrumentId": str(INSTRUMENT_ID)}, |
| 52 | ) |
| 53 | assert result["ok"] is True |
| 54 | assert result["data"]["globalInstrumentId"] == str(INSTRUMENT_ID) |
| 55 | assert not {"quantity", "averageCost", "investedAmount", "pnl", "allocation"}.intersection( |
| 56 | result["data"] |
| 57 | ) |
| 58 | assert reader.provider_calls == 0 |
| 59 | |
| 60 | |
| 61 | @pytest.mark.asyncio |
| 62 | async def test_readiness_matches_existing_capability_and_does_not_acquire(container, reader) -> None: |
| 63 | expected = await reader.invoke( |
| 64 | "get_research_readiness", |
| 65 | {"globalInstrumentId": str(INSTRUMENT_ID)}, |
| 66 | container.auth, |
| 67 | "direct", |
| 68 | ) |
| 69 | reader.calls.clear() |
| 70 | result = await invoke( |
| 71 | container, "get_research_readiness", {"globalInstrumentId": str(INSTRUMENT_ID)} |
| 72 | ) |
| 73 | assert result["data"] == expected |
| 74 | assert reader.provider_calls == 0 |
| 75 | assert reader.calls[0][0] == "get_research_readiness" |
| 76 | |
| 77 | |
| 78 | @pytest.mark.asyncio |
| 79 | async def test_analysis_has_rule_engine_parity_and_zero_provider_calls(container, reader) -> None: |
| 80 | direct = await reader.invoke( |
| 81 | "get_company_analysis", |
| 82 | {"globalInstrumentId": str(INSTRUMENT_ID), "allowPartial": False}, |
| 83 | container.auth, |
| 84 | "direct", |
| 85 | ) |
| 86 | result = await invoke( |
| 87 | container, |
| 88 | "get_company_analysis", |
| 89 | {"globalInstrumentId": str(INSTRUMENT_ID)}, |
| 90 | ) |
| 91 | assert result["data"] == direct |
| 92 | assert result["provenance"]["ruleEngineVersion"] == "STOCK_RULE_ENGINE_V1" |
| 93 | assert reader.provider_calls == 0 |
| 94 | |
| 95 | |
| 96 | @pytest.mark.asyncio |
| 97 | async def test_recent_news_default_and_maximum_contract(container, reader) -> None: |
| 98 | default_result = await invoke( |
| 99 | container, "get_recent_news", {"globalInstrumentId": str(INSTRUMENT_ID)} |
| 100 | ) |
| 101 | too_large = await invoke( |
| 102 | container, |
| 103 | "get_recent_news", |
| 104 | {"globalInstrumentId": str(INSTRUMENT_ID), "days": 31}, |
| 105 | ) |
| 106 | item = default_result["data"]["news"][0] |
| 107 | assert default_result["data"]["days"] == 30 |
| 108 | assert too_large["error"]["code"] == "INVALID_ARGUMENT" |
| 109 | assert item["publicationDate"] |
| 110 | assert item["source"]["type"] == "EXCHANGE_FILING" |
| 111 | assert reader.provider_calls == 0 |
| 112 | |
| 113 | |
| 114 | @pytest.mark.asyncio |
| 115 | async def test_sector_performance_uses_durable_read_only_capability(container, reader) -> None: |
| 116 | result = await invoke( |
| 117 | container, |
| 118 | "get_sector_performance", |
| 119 | {"region": "INDIA", "sector": "FINANCIAL_SERVICES", "period": "MONTH"}, |
| 120 | ) |
| 121 | assert result["ok"] is True |
| 122 | assert result["data"]["sector"] == "FINANCIAL_SERVICES" |
| 123 | assert reader.calls[-1][0] == "get_sector_performance" |
| 124 | assert reader.provider_calls == 0 |
| 125 | |
| 126 | |
| 127 | @pytest.mark.asyncio |
| 128 | async def test_private_financial_fields_and_tokens_are_removed(container, reader) -> None: |
| 129 | async def private_response(*_args, **_kwargs): |
| 130 | return { |
| 131 | "globalInstrumentId": str(INSTRUMENT_ID), |
| 132 | "quantity": 10, |
| 133 | "averageCost": 99, |
| 134 | "pnl": 12, |
| 135 | "P&L": 12, |
| 136 | "cost basis": 99, |
| 137 | "nested": {"authorization": "Bearer abc.def.ghi", "safe": "kept"}, |
| 138 | "apiKey": "secret-value", |
| 139 | "message": "upstream failed with api_key=must-not-leak", |
| 140 | } |
| 141 | |
| 142 | reader.invoke = private_response |
| 143 | result = await invoke( |
| 144 | container, "get_financial_facts", {"globalInstrumentId": str(INSTRUMENT_ID)} |
| 145 | ) |
| 146 | serialized = str(result) |
| 147 | assert result["data"] == { |
| 148 | "globalInstrumentId": str(INSTRUMENT_ID), |
| 149 | "nested": {"safe": "kept"}, |
| 150 | "message": "upstream failed with api_key=<redacted>", |
| 151 | } |
| 152 | assert "secret-value" not in serialized and "must-not-leak" not in serialized |
| 153 | assert "Bearer" not in serialized |
| 154 | |
| 155 | |
| 156 | @pytest.mark.asyncio |
| 157 | async def test_invalid_arguments_are_deterministic(container) -> None: |
| 158 | result = await invoke( |
| 159 | container, |
| 160 | "get_sector_performance", |
| 161 | {"region": "MARS", "sector": "", "period": "WEEK", "unexpected": True}, |
| 162 | ) |
| 163 | assert result["error"] == { |
| 164 | "code": "INVALID_ARGUMENT", |
| 165 | "message": "The tool arguments are invalid.", |
| 166 | } |
| 167 | |
| 168 | |
| 169 | class EmptyInput(StrictContract): |
| 170 | pass |
| 171 | |
| 172 | |
| 173 | def service_for(handler, auth, audit, timeout=0.01, risk=McpRiskClass.SAFE_READ): |
| 174 | registry = McpToolRegistry() |
| 175 | registry.register( |
| 176 | McpToolDefinition( |
| 177 | name="test_tool", |
| 178 | description="test", |
| 179 | input_model=EmptyInput, |
| 180 | risk_class=risk, |
| 181 | kind=McpToolKind.INTERNAL, |
| 182 | handler=handler, |
| 183 | ) |
| 184 | ) |
| 185 | return McpInvocationService(registry, McpToolPolicy(), audit, timeout_seconds=timeout) |
| 186 | |
| 187 | |
| 188 | @pytest.mark.asyncio |
| 189 | async def test_timeout_is_deterministic(auth, audit) -> None: |
| 190 | async def slow(*_args): |
| 191 | await asyncio.sleep(0.1) |
| 192 | return McpToolExecution(data={}) |
| 193 | |
| 194 | result = await service_for(slow, auth, audit).invoke("test_tool", {}, auth) |
| 195 | assert result["error"]["code"] == "DOWNSTREAM_TIMEOUT" |
| 196 | |
| 197 | |
| 198 | @pytest.mark.asyncio |
| 199 | async def test_downstream_exception_does_not_leak(auth, audit) -> None: |
| 200 | async def explode(*_args): |
| 201 | raise RuntimeError("database password leaked-stack-marker") |
| 202 | |
| 203 | result = await service_for(explode, auth, audit).invoke("test_tool", {}, auth) |
| 204 | assert result["error"]["code"] == "DOWNSTREAM_UNAVAILABLE" |
| 205 | assert "leaked-stack-marker" not in str(result) |
| 206 | |
| 207 | |
| 208 | @pytest.mark.asyncio |
| 209 | async def test_policy_denial_emits_denied_audit_event(auth, audit) -> None: |
| 210 | async def unused(*_args): |
| 211 | raise AssertionError("denied handler must never run") |
| 212 | |
| 213 | result = await service_for( |
| 214 | unused, auth, audit, risk=McpRiskClass.FINANCIAL_ACTION |
| 215 | ).invoke("test_tool", {}, auth, request_id="denied-5a") |
| 216 | assert result["error"]["code"] == "MCP_TOOL_DENIED" |
| 217 | assert [event["event"] for event in audit.events] == [ |
| 218 | "MCP_TOOL_INVOKED", |
| 219 | "MCP_TOOL_DENIED", |
| 220 | ] |
| 221 | |
| 222 | |
| 223 | @pytest.mark.asyncio |
| 224 | async def test_invoked_success_and_request_correlation_are_retained(container, audit) -> None: |
| 225 | result = await invoke( |
| 226 | container, |
| 227 | "get_research_readiness", |
| 228 | {"globalInstrumentId": str(INSTRUMENT_ID)}, |
| 229 | request_id="correlation-5a", |
| 230 | ) |
| 231 | assert result["requestId"] == "correlation-5a" |
| 232 | assert [item["event"] for item in audit.events] == ["MCP_TOOL_INVOKED", "MCP_TOOL_SUCCEEDED"] |
| 233 | assert all(item["request_id"] == "correlation-5a" for item in audit.events) |
| 234 | |
| 235 | |
| 236 | @pytest.mark.asyncio |
| 237 | async def test_watchlist_requires_user_scope_and_preserves_user_isolation(container, reader) -> None: |
| 238 | result = await invoke(container, "get_watchlist", {"watchlistId": str(WATCHLIST_ID)}) |
| 239 | assert result["data"]["userId"] == str(container.auth.user_id) |
| 240 | assert reader.calls[-1][2].user_id == container.auth.user_id |
| 241 | |
| 242 | |
| 243 | @pytest.mark.asyncio |
| 244 | async def test_watchlist_fails_closed_without_user_or_scope(container, reader) -> None: |
| 245 | service_only = McpAuthContext( |
| 246 | serviceIdentity="mcp-test-service", |
| 247 | scopes=("mcp:read", "watchlist:read"), |
| 248 | authenticationType=McpAuthenticationType.TEST, |
| 249 | ) |
| 250 | no_user = await container.invocation.invoke( |
| 251 | "get_watchlist", {"watchlistId": str(WATCHLIST_ID)}, service_only |
| 252 | ) |
| 253 | missing_scope = container.auth.model_copy(update={"scopes": ("mcp:read",)}) |
| 254 | no_scope = await container.invocation.invoke( |
| 255 | "get_watchlist", {"watchlistId": str(WATCHLIST_ID)}, missing_scope |
| 256 | ) |
| 257 | assert no_user["error"]["code"] == "UNAUTHORIZED" |
| 258 | assert no_scope["error"]["code"] == "FORBIDDEN" |
| 259 | assert reader.calls == [] |