| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | import os |
| 5 | import sys |
| 6 | from types import SimpleNamespace |
| 7 | from uuid import UUID |
| 8 | |
| 9 | import pytest |
| 10 | from mcp import Client, StdioServerParameters |
| 11 | from pydantic import ValidationError |
| 12 | from starlette.testclient import TestClient |
| 13 | |
| 14 | from fake_yahoo import factory |
| 15 | from yahoo_mcp_server.acquisition import YahooAcquisitionService |
| 16 | from yahoo_mcp_server.contracts import IdentityInput, NewsInput, RawFact, ToolPayload |
| 17 | from yahoo_mcp_server.server import REGISTERED_TOOLS, create_yahoo_mcp_server |
| 18 | from yahoo_mcp_server.settings import YahooMcpSettings |
| 19 | |
| 20 | |
| 21 | INSTRUMENT_ID = UUID("11111111-1111-4111-8111-111111111111") |
| 22 | |
| 23 | |
| 24 | class RecordingAudit: |
| 25 | def __init__(self) -> None: |
| 26 | self.events = [] |
| 27 | |
| 28 | def emit(self, event, **values) -> None: |
| 29 | self.events.append({"event": event, **values}) |
| 30 | |
| 31 | |
| 32 | def settings(**overrides) -> YahooMcpSettings: |
| 33 | values = { |
| 34 | "AIP_ENVIRONMENT": "TEST", |
| 35 | "AIP_YAHOO_MCP_UPSTREAM_TIMEOUT_SECONDS": "1", |
| 36 | "AIP_YAHOO_MCP_MAX_CONCURRENCY": "2", |
| 37 | } |
| 38 | values.update(overrides) |
| 39 | return YahooMcpSettings(**values) |
| 40 | |
| 41 | |
| 42 | def server(scenario="SUCCESS", *, audit=None, timeout="1"): |
| 43 | config = settings(AIP_YAHOO_MCP_UPSTREAM_TIMEOUT_SECONDS=timeout) |
| 44 | acquisition = YahooAcquisitionService(config, ticker_factory=factory(scenario)) |
| 45 | return create_yahoo_mcp_server(config, acquisition=acquisition, audit=audit) |
| 46 | |
| 47 | |
| 48 | def arguments(symbol="HAL.NS", **overrides): |
| 49 | exchange = "XNSE" if symbol.endswith(".NS") else "XAMS" if symbol.endswith(".AS") else "XNAS" |
| 50 | currency = "INR" if symbol.endswith(".NS") else "EUR" if symbol.endswith(".AS") else "USD" |
| 51 | value = { |
| 52 | "globalInstrumentId": str(INSTRUMENT_ID), |
| 53 | "verifiedYahooSymbol": symbol, |
| 54 | "region": "INDIA" if symbol.endswith(".NS") else "EUROPE" if symbol.endswith(".AS") else "USA", |
| 55 | "exchange": exchange, |
| 56 | "currency": currency, |
| 57 | } |
| 58 | value.update(overrides) |
| 59 | return value |
| 60 | |
| 61 | |
| 62 | async def call_tool(instance, name, values=None): |
| 63 | async with Client(instance) as client: |
| 64 | return await client.call_tool(name, values or arguments()) |
| 65 | |
| 66 | |
| 67 | @pytest.mark.asyncio |
| 68 | async def test_registers_only_the_eight_safe_read_tools_with_discoverable_strict_schemas(): |
| 69 | async with Client(server()) as client: |
| 70 | listed = await client.list_tools() |
| 71 | tools = {item.name: item for item in listed.tools} |
| 72 | assert set(tools) == set(REGISTERED_TOOLS) |
| 73 | assert all(item.annotations.read_only_hint is True for item in tools.values()) |
| 74 | assert all(item.annotations.destructive_hint is False for item in tools.values()) |
| 75 | quote_schema = tools["get_quote"].input_schema |
| 76 | assert {"globalInstrumentId", "verifiedYahooSymbol", "region"}.issubset(quote_schema["required"]) |
| 77 | assert quote_schema["additionalProperties"] is False |
| 78 | serialized = json.dumps(quote_schema) |
| 79 | assert "companyName" not in serialized |
| 80 | assert not {"trade", "order", "search_symbol", "fetch_url"}.intersection(tools) |
| 81 | |
| 82 | |
| 83 | @pytest.mark.asyncio |
| 84 | async def test_real_stdio_entrypoint_initializes_lists_tools_and_shuts_down_cleanly(): |
| 85 | environment = { |
| 86 | key: value |
| 87 | for key in ("PATH", "SYSTEMROOT", "WINDIR", "HOME", "TMP", "TEMP") |
| 88 | if (value := os.environ.get(key)) |
| 89 | } |
| 90 | async with Client( |
| 91 | StdioServerParameters( |
| 92 | command=sys.executable, |
| 93 | args=["-m", "yahoo_mcp_server.main", "--transport", "stdio"], |
| 94 | env=environment, |
| 95 | ), |
| 96 | read_timeout_seconds=5, |
| 97 | cache=None, |
| 98 | ) as client: |
| 99 | listed = await client.list_tools() |
| 100 | assert {item.name for item in listed.tools} == set(REGISTERED_TOOLS) |
| 101 | |
| 102 | |
| 103 | @pytest.mark.asyncio |
| 104 | async def test_quote_contract_preserves_canonical_identity_and_normalized_price(): |
| 105 | result = await call_tool(server(), "get_quote") |
| 106 | assert result.is_error is False |
| 107 | payload = ToolPayload.model_validate(result.structured_content) |
| 108 | assert payload.global_instrument_id == INSTRUMENT_ID |
| 109 | assert payload.symbol == "HAL.NS" |
| 110 | assert str(payload.price) == "250.5" |
| 111 | assert payload.source == "YAHOO_FINANCE" |
| 112 | assert "quantity" not in result.structured_content |
| 113 | |
| 114 | |
| 115 | @pytest.mark.asyncio |
| 116 | @pytest.mark.parametrize("symbol", ["HAL.NS", "RBLBANK.NS", "AAPL", "MSFT", "BESI.AS"]) |
| 117 | async def test_representative_verified_provider_symbols_are_forwarded_exactly(symbol): |
| 118 | result = await call_tool(server(), "get_quote", arguments(symbol)) |
| 119 | assert result.is_error is False |
| 120 | assert result.structured_content["symbol"] == symbol |
| 121 | |
| 122 | |
| 123 | @pytest.mark.asyncio |
| 124 | async def test_history_reuses_close_semantics_and_filters_non_finite_rows(): |
| 125 | result = await call_tool(server(), "get_price_history", {**arguments(), "lookbackDays": 400}) |
| 126 | assert result.is_error is False |
| 127 | assert [item["close"] for item in result.structured_content["prices"]] == ["248.0", "250.5"] |
| 128 | |
| 129 | |
| 130 | @pytest.mark.asyncio |
| 131 | async def test_profile_sector_and_industry_are_normalized_without_raw_response(): |
| 132 | profile = await call_tool(server(), "get_company_profile") |
| 133 | sector = await call_tool(server(), "get_sector_industry") |
| 134 | assert profile.structured_content["profile"]["companyName"] == "HAL.NS Company" |
| 135 | assert sector.structured_content["profile"] == { |
| 136 | "companyName": "HAL.NS Company", |
| 137 | "sector": "Industrials", |
| 138 | "industry": "Aerospace & Defense", |
| 139 | } |
| 140 | assert "longBusinessSummary" not in str(profile.structured_content) |
| 141 | |
| 142 | |
| 143 | @pytest.mark.asyncio |
| 144 | async def test_annual_and_quarterly_tools_keep_periods_separate_and_raw_origin(): |
| 145 | annual = await call_tool(server(), "get_financials") |
| 146 | quarterly = await call_tool(server(), "get_quarterly_financials") |
| 147 | annual_facts = annual.structured_content["facts"] |
| 148 | quarterly_facts = quarterly.structured_content["facts"] |
| 149 | assert {item.get("periodType") for item in annual_facts if item.get("periodType")} == {"ANNUAL"} |
| 150 | assert {item.get("periodType") for item in quarterly_facts} == {"QUARTERLY"} |
| 151 | assert any(item.get("rawFieldOrigin") == "Total Revenue" for item in annual_facts) |
| 152 | assert len({item["periodEnd"] for item in quarterly_facts}) == 2 |
| 153 | |
| 154 | |
| 155 | @pytest.mark.asyncio |
| 156 | async def test_news_is_current_deduplicated_and_issuer_scoped(): |
| 157 | result = await call_tool(server(), "get_news", {**arguments(), "days": 30}) |
| 158 | assert result.is_error is False |
| 159 | assert len(result.structured_content["news"]) == 1 |
| 160 | article = result.structured_content["news"][0] |
| 161 | assert article["issuerSymbol"] == "HAL.NS" |
| 162 | assert article["publishedAt"] |
| 163 | assert "materiality" not in article |
| 164 | |
| 165 | |
| 166 | @pytest.mark.asyncio |
| 167 | async def test_analyst_tool_exposes_only_present_supporting_fields(): |
| 168 | result = await call_tool(server(), "get_analyst_data") |
| 169 | metrics = {item["metric"] for item in result.structured_content["facts"]} |
| 170 | assert { |
| 171 | "publicAnalystTargetMeanPrice", |
| 172 | "publicAnalystCount", |
| 173 | "publicAnalystConsensus", |
| 174 | }.issubset(metrics) |
| 175 | assert "investmentRecommendation" not in metrics |
| 176 | |
| 177 | |
| 178 | @pytest.mark.asyncio |
| 179 | @pytest.mark.parametrize( |
| 180 | ("scenario", "expected"), |
| 181 | [ |
| 182 | ("WRONG_SYMBOL", "YAHOO_MCP_IDENTITY_MISMATCH"), |
| 183 | ("WRONG_EXCHANGE", "YAHOO_MCP_IDENTITY_MISMATCH"), |
| 184 | ("WRONG_CURRENCY", "YAHOO_MCP_IDENTITY_MISMATCH"), |
| 185 | ("EMPTY", "YAHOO_MCP_INCOMPLETE"), |
| 186 | ("INVALID_INFO", "YAHOO_MCP_UPSTREAM_UNAVAILABLE"), |
| 187 | ("UPSTREAM_FAILURE", "YAHOO_MCP_UPSTREAM_UNAVAILABLE"), |
| 188 | ], |
| 189 | ) |
| 190 | async def test_failures_are_deterministic_and_do_not_leak_upstream_details(scenario, expected): |
| 191 | result = await call_tool(server(scenario), "get_quote") |
| 192 | assert result.is_error is True |
| 193 | message = str(result.content) |
| 194 | assert expected in message |
| 195 | assert "sensitive" not in message |
| 196 | assert "Traceback" not in message |
| 197 | |
| 198 | |
| 199 | @pytest.mark.asyncio |
| 200 | async def test_timeout_is_bounded_and_deterministic(): |
| 201 | result = await call_tool(server("TIMEOUT", timeout="0.05"), "get_quote") |
| 202 | assert result.is_error is True |
| 203 | assert "YAHOO_MCP_UPSTREAM_TIMEOUT" in str(result.content) |
| 204 | |
| 205 | |
| 206 | @pytest.mark.asyncio |
| 207 | async def test_nonfinite_financial_cell_is_omitted_without_rejecting_valid_facts(): |
| 208 | result = await call_tool(server("NAN_FACT"), "get_financials") |
| 209 | assert result.is_error is False |
| 210 | payload = ToolPayload.model_validate(result.structured_content) |
| 211 | assert not any(fact.metric == "marketCap" for fact in payload.facts) |
| 212 | assert any(fact.metric == "pat" for fact in payload.facts) |
| 213 | |
| 214 | |
| 215 | @pytest.mark.asyncio |
| 216 | async def test_unknown_or_unsafe_tool_and_extra_arguments_fail_closed(): |
| 217 | unknown = await call_tool(server(), "place_order") |
| 218 | extra = await call_tool(server(), "get_quote", {**arguments(), "companyName": "fuzzy"}) |
| 219 | assert unknown.is_error is True |
| 220 | assert extra.is_error is True |
| 221 | |
| 222 | |
| 223 | def test_input_contract_rejects_fuzzy_identity_bad_symbol_and_more_than_30_news_days(): |
| 224 | with pytest.raises(ValidationError): |
| 225 | IdentityInput(**{**arguments(), "companyName": "Hindustan Aeronautics"}) |
| 226 | with pytest.raises(ValidationError): |
| 227 | IdentityInput(**arguments("../../etc/passwd")) |
| 228 | with pytest.raises(ValidationError): |
| 229 | NewsInput(**{**arguments(), "days": 31}) |
| 230 | |
| 231 | |
| 232 | def test_output_fact_contract_rejects_malformed_period_identity(): |
| 233 | with pytest.raises(ValidationError): |
| 234 | RawFact(metric="revenue", value="1", periodEnd="FY2026", periodType="ANNUAL") |
| 235 | with pytest.raises(ValidationError): |
| 236 | RawFact(metric="revenue", value="1", periodEnd="2026-03-31", periodType="TRAILING") |
| 237 | |
| 238 | |
| 239 | def test_health_and_readiness_do_not_call_yahoo_upstream(): |
| 240 | app = server("UPSTREAM_FAILURE").streamable_http_app(stateless_http=True, json_response=True) |
| 241 | with TestClient(app) as client: |
| 242 | health = client.get("/health") |
| 243 | ready = client.get("/health/ready") |
| 244 | assert health.status_code == 200 and health.json()["registeredTools"] == 8 |
| 245 | assert ready.status_code == 200 and ready.json()["upstreamRequired"] is False |
| 246 | |
| 247 | |
| 248 | @pytest.mark.asyncio |
| 249 | async def test_audit_records_safe_metadata_without_arguments_results_or_private_fields(): |
| 250 | audit = RecordingAudit() |
| 251 | result = await call_tool(server(audit=audit), "get_quote") |
| 252 | assert result.is_error is False |
| 253 | assert [item["event"] for item in audit.events] == ["YAHOO_MCP_REQUEST", "YAHOO_MCP_SUCCESS"] |
| 254 | serialized = json.dumps(audit.events, default=str) |
| 255 | for forbidden in ("quantity", "averageCost", "authorization", "currentPrice", "facts", "news"): |
| 256 | assert forbidden not in serialized |
| 257 | |
| 258 | |
| 259 | def test_local_and_azure_profiles_parse_without_secrets_or_cloud_dependency(): |
| 260 | local = YahooMcpSettings(AIP_ENVIRONMENT="LOCAL") |
| 261 | azure = YahooMcpSettings(AIP_ENVIRONMENT="AZURE") |
| 262 | assert local.transport == "stdio" |
| 263 | assert azure.environment == "AZURE" |
| 264 | assert "secret" not in json.dumps(azure.model_dump()).lower() |
| 265 | |
| 266 | @pytest.mark.asyncio |
| 267 | async def test_financial_issuer_quarters_with_missing_cells_keep_valid_reported_values(): |
| 268 | from fake_yahoo import FakeYahooTicker |
| 269 | class FinancialTicker(FakeYahooTicker): |
| 270 | @property |
| 271 | def info(self): |
| 272 | return {**super().info, "sector": "Financial Services", "industry": "Banks - Regional"} |
| 273 | def __init__(self, symbol): |
| 274 | super().__init__(symbol) |
| 275 | self.quarterly_income_stmt.iloc[0, 0] = float("nan") |
| 276 | config = settings() |
| 277 | acquisition = YahooAcquisitionService(config, ticker_factory=FinancialTicker) |
| 278 | instance = create_yahoo_mcp_server(config, acquisition=acquisition) |
| 279 | result = await call_tool(instance, "get_quarterly_financials") |
| 280 | assert not result.is_error |
| 281 | payload = ToolPayload.model_validate(result.structured_content) |
| 282 | assert any(fact.metric == "pat" for fact in payload.facts) |
| 283 | assert sum(fact.metric == "revenue" for fact in payload.facts) == 1 |
| 284 | assert not any(fact.metric == "roce" for fact in payload.facts) |