| 1 | from fastapi.testclient import TestClient |
| 2 | import asyncio |
| 3 | from uuid import UUID, uuid4 |
| 4 | |
| 5 | import httpx |
| 6 | import pytest |
| 7 | |
| 8 | import app.main as main |
| 9 | from app.main import app |
| 10 | from app.portfolio_orchestration import PortfolioResearchOrchestrator, PortfolioServiceUnavailableError |
| 11 | from app.repository import ResearchRepository |
| 12 | from app.settings import Settings |
| 13 | from app.models import ( |
| 14 | PortfolioResearchCompany, |
| 15 | PortfolioResearchSummary, |
| 16 | ReliabilityLevel, |
| 17 | ShareholdingCategory, |
| 18 | ShareholdingSnapshot, |
| 19 | ShareholdingSnapshotValue, |
| 20 | SourceMode, |
| 21 | ProvenancedValue, |
| 22 | ) |
| 23 | from app.persistence import SqliteResearchPersistence |
| 24 | from app.fact_precedence import FactSourceTier, FinancialFact, FinancialFactKey |
| 25 | from datetime import datetime, timezone |
| 26 | from decimal import Decimal |
| 27 | |
| 28 | |
| 29 | def test_research_company_summary_api_returns_demo_evidence_without_raw_bodies() -> None: |
| 30 | client = TestClient(app) |
| 31 | |
| 32 | companies = client.get("/api/v1/research/companies").json() |
| 33 | assert companies |
| 34 | instrument_id = companies[0]["instrumentId"] |
| 35 | |
| 36 | summary = client.get(f"/api/v1/research/companies/{instrument_id}/summary") |
| 37 | |
| 38 | assert summary.status_code == 200 |
| 39 | body = summary.json() |
| 40 | assert body["demo"] is True |
| 41 | assert "overallScore" in body["catalystScore"] |
| 42 | assert body["recentEvents"] |
| 43 | assert "rawText" not in str(body) |
| 44 | assert "normalizedText" not in str(body) |
| 45 | |
| 46 | |
| 47 | def test_research_company_summary_api_is_object_and_normalizes_casefolded_bucket_aliases() -> None: |
| 48 | client = TestClient(app) |
| 49 | instrument_id = client.get("/api/v1/research/companies").json()[0]["instrumentId"] |
| 50 | expected = main.repository.summary(UUID(instrument_id), allow_demo=True) |
| 51 | |
| 52 | response = client.get(f"/api/v1/research/companies/{instrument_id}/summary") |
| 53 | |
| 54 | assert response.status_code == 200 |
| 55 | assert response.headers["content-type"].startswith("application/json") |
| 56 | assert response.text.lstrip().startswith("{") |
| 57 | body = response.json() |
| 58 | assert isinstance(body, dict) |
| 59 | assert isinstance(body["shareholdingSnapshots"], list) |
| 60 | assert body["shareholdingFreshness"] in {"REAL", "UNAVAILABLE"} |
| 61 | buckets = body["catalystScore"]["buckets"] |
| 62 | assert len({key.casefold() for key in buckets}) == len(buckets) |
| 63 | assert not ({"Guidance", "GUIDANCE"} <= set(buckets)) |
| 64 | category_evidence = body["catalystScore"]["categoryEvidence"] |
| 65 | assert "CAPEX & Capacity" not in category_evidence |
| 66 | for evidence in category_evidence.values(): |
| 67 | supporting_events = evidence["supportingEvents"] |
| 68 | if evidence["status"] == "NO_EVIDENCE": |
| 69 | assert supporting_events == [] |
| 70 | continue |
| 71 | assert supporting_events |
| 72 | identities = { |
| 73 | event.get("independenceKey") or event["sourceDocumentId"] |
| 74 | for event in supporting_events |
| 75 | } |
| 76 | assert evidence["sourceCount"] == len(identities) |
| 77 | assert body["catalystScore"]["overallScore"] == expected.catalyst_score.overall_score |
| 78 | assert body["shareholdingSnapshots"] == [snapshot.model_dump(mode="json", by_alias=True) for snapshot in expected.shareholding_snapshots] |
| 79 | |
| 80 | |
| 81 | def test_company_summary_serializes_unknown_basis_financial_history_from_persisted_facts(monkeypatch) -> None: |
| 82 | repository = ResearchRepository(persistence=SqliteResearchPersistence()) |
| 83 | instrument_id = repository.list_profiles()[0].instrument_id |
| 84 | retrieved = datetime(2026, 8, 1, tzinfo=timezone.utc) |
| 85 | facts = [] |
| 86 | def add(period: str, period_type: str, metric: str, value: str) -> None: |
| 87 | facts.append(FinancialFact( |
| 88 | FinancialFactKey(instrument_id, metric, period, period_type, None), |
| 89 | ProvenancedValue(value=Decimal(value), unit="INR lakh" if metric != "eps" else "INR per share", |
| 90 | source_url="https://nsearchives.nseindia.com/result.pdf", source_name="NSE", source_type="EXCHANGE", retrieved_at=retrieved), |
| 91 | FactSourceTier.OFFICIAL_NSE, "NSE", f"nse:{period}:{metric}", SourceMode.REAL, |
| 92 | )) |
| 93 | for period in ("2026-06-30", "2026-03-31", "2025-12-31", "2025-06-30", "2025-03-31"): |
| 94 | for metric, value in (("revenue", "228093"), ("pat", "31654"), ("eps", "1.63")): |
| 95 | add(period, "QUARTERLY", metric, value) |
| 96 | for period in ("2026-03-31", "2025-03-31", "2024-03-31", "2023-03-31", "2022-03-31"): |
| 97 | for metric, value in (("revenue", "803897"), ("pat", "69263"), ("eps", "3.57")): |
| 98 | add(period, "ANNUAL", metric, value) |
| 99 | monkeypatch.setattr(main, "repository", repository) |
| 100 | monkeypatch.setattr(repository, "financial_facts_for", lambda _instrument_id: facts) |
| 101 | |
| 102 | body = TestClient(app).get(f"/api/v1/research/companies/{instrument_id}/summary").json() |
| 103 | |
| 104 | assert body["latestQuarterlyResult"]["period"] == "2026-06-30" |
| 105 | assert [item["period"] for item in body["financialResultHistory"] if item["periodType"] == "QUARTERLY"] == ["2026-06-30", "2026-03-31", "2025-12-31", "2025-06-30"] |
| 106 | assert [item["period"] for item in body["financialResultHistory"] if item["periodType"] == "ANNUAL"] == ["2026-03-31", "2025-03-31", "2024-03-31", "2023-03-31"] |
| 107 | assert {item["reportingBasis"] for item in body["financialResultHistory"]} == {None} |
| 108 | |
| 109 | |
| 110 | @pytest.mark.parametrize( |
| 111 | ("method", "path"), |
| 112 | ( |
| 113 | ("post", "/api/v1/research/prefetch"), |
| 114 | ("get", "/api/v1/research/prefetch/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa"), |
| 115 | ("post", "/api/v1/research/companies/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa/refresh"), |
| 116 | ("post", "/api/v1/research/companies/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa/backfill"), |
| 117 | ("post", "/api/v1/research/portfolios/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa/refresh"), |
| 118 | ("get", "/api/v1/research/refresh-jobs/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa"), |
| 119 | ("get", "/api/v1/research/portfolios/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa/refresh-job/active"), |
| 120 | ), |
| 121 | ) |
| 122 | def test_retired_legacy_research_routes_return_404(method: str, path: str) -> None: |
| 123 | response = getattr(TestClient(app), method)(path) |
| 124 | |
| 125 | assert response.status_code == 404 |
| 126 | |
| 127 | |
| 128 | def test_openapi_exposes_only_the_readiness_and_analysis_cutover_routes() -> None: |
| 129 | paths = TestClient(app).get("/openapi.json").json()["paths"] |
| 130 | |
| 131 | assert "/api/v1/research/readiness/{global_instrument_id}" in paths |
| 132 | assert "/api/v1/research/readiness/{global_instrument_id}/ensure" in paths |
| 133 | assert "/api/v1/research/analysis/{global_instrument_id}" in paths |
| 134 | assert "/api/v1/research/prefetch" not in paths |
| 135 | assert "/api/v1/research/prefetch/{instrument_id}" not in paths |
| 136 | assert "/api/v1/research/companies/{instrument_id}/refresh" not in paths |
| 137 | assert "/api/v1/research/companies/{instrument_id}/backfill" not in paths |
| 138 | assert "/api/v1/research/portfolios/{portfolio_id}/refresh" not in paths |
| 139 | assert "/api/v1/research/refresh-jobs/{job_id}" not in paths |
| 140 | assert "/api/v1/research/portfolios/{portfolio_id}/refresh-job/active" not in paths |
| 141 | |
| 142 | |
| 143 | def test_portfolio_research_summary_api_is_read_only(monkeypatch) -> None: |
| 144 | class FakePortfolioOrchestrator: |
| 145 | def __init__(self) -> None: |
| 146 | self.read_called = False |
| 147 | self.refresh_called = False |
| 148 | |
| 149 | async def read_portfolio_summary(self, portfolio_id, correlation_id=None, identity_headers=None): |
| 150 | self.read_called = True |
| 151 | return PortfolioResearchSummary(portfolio_id=portfolio_id) |
| 152 | |
| 153 | async def refresh_portfolio(self, portfolio_id, correlation_id=None, identity_headers=None): |
| 154 | self.refresh_called = True |
| 155 | return PortfolioResearchSummary(portfolio_id=portfolio_id) |
| 156 | |
| 157 | fake = FakePortfolioOrchestrator() |
| 158 | monkeypatch.setattr(main, "portfolio_orchestrator", fake) |
| 159 | client = TestClient(app) |
| 160 | |
| 161 | response = client.get("/api/v1/research/portfolios/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa/summary") |
| 162 | |
| 163 | assert response.status_code == 200 |
| 164 | assert response.json()["portfolioId"] == "aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa" |
| 165 | assert fake.read_called is True |
| 166 | assert fake.refresh_called is False |
| 167 | |
| 168 | |
| 169 | def test_portfolio_summary_api_is_object_and_normalizes_casefolded_company_evidence_without_losing_shareholding(monkeypatch) -> None: |
| 170 | instrument_id = uuid4() |
| 171 | periods = ( |
| 172 | datetime(2026, 6, 30, tzinfo=timezone.utc), |
| 173 | datetime(2026, 3, 31, tzinfo=timezone.utc), |
| 174 | datetime(2025, 12, 31, tzinfo=timezone.utc), |
| 175 | datetime(2025, 9, 30, tzinfo=timezone.utc), |
| 176 | ) |
| 177 | snapshots = [ |
| 178 | ShareholdingSnapshot( |
| 179 | instrument_id=instrument_id, |
| 180 | period_end=period, |
| 181 | source_provider="NSE", |
| 182 | source_type="NSE_SHAREHOLDING_XBRL", |
| 183 | source_identity_key=f"NSE_SHAREHOLDING:{index}", |
| 184 | source_url=f"https://nsearchives.nseindia.com/corporate/xbrl/{index}.xml", |
| 185 | confidence=Decimal("0.95"), |
| 186 | reliability_level=ReliabilityLevel.LEVEL_A, |
| 187 | source_mode=SourceMode.REAL, |
| 188 | values=[ShareholdingSnapshotValue( |
| 189 | category=ShareholdingCategory.PROMOTER, |
| 190 | percentage=Decimal("49.40"), |
| 191 | raw_source_label="Promoter and Promoter Group", |
| 192 | source_locator="nse-xbrl:fixture", |
| 193 | evidence_text="Official NSE XBRL fixture", |
| 194 | )], |
| 195 | ) |
| 196 | for index, period in enumerate(periods, start=1) |
| 197 | ] |
| 198 | |
| 199 | class FakePortfolioOrchestrator: |
| 200 | async def read_portfolio_summary(self, portfolio_id, correlation_id=None, identity_headers=None): |
| 201 | return PortfolioResearchSummary( |
| 202 | portfolio_id=portfolio_id, |
| 203 | companies=[PortfolioResearchCompany( |
| 204 | instrument_id=instrument_id, |
| 205 | company_name="Generic Global Equity", |
| 206 | status="RESOLVED_RESEARCH_AVAILABLE", |
| 207 | evidence_coverage={ |
| 208 | "Guidance": "POSITIVE_EVIDENCE", |
| 209 | "GUIDANCE": "", |
| 210 | "Growth": "NO_EVIDENCE", |
| 211 | }, |
| 212 | shareholding_snapshots=snapshots, |
| 213 | shareholding_freshness="REAL", |
| 214 | )], |
| 215 | ) |
| 216 | |
| 217 | monkeypatch.setattr(main, "portfolio_orchestrator", FakePortfolioOrchestrator()) |
| 218 | client = TestClient(app) |
| 219 | response = client.get("/api/v1/research/portfolios/aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa/summary") |
| 220 | |
| 221 | assert response.status_code == 200 |
| 222 | assert response.headers["content-type"].startswith("application/json") |
| 223 | assert response.text.lstrip().startswith("{") |
| 224 | body = response.json() |
| 225 | assert isinstance(body, dict) |
| 226 | company = body["companies"][0] |
| 227 | assert company["evidenceCoverage"]["GUIDANCE"] == "POSITIVE_EVIDENCE" |
| 228 | assert len({key.casefold() for key in company["evidenceCoverage"]}) == len(company["evidenceCoverage"]) |
| 229 | assert len(company["shareholdingSnapshots"]) == 4 |
| 230 | assert company["shareholdingFreshness"] == "REAL" |
| 231 | assert [snapshot["periodEnd"][:10] for snapshot in company["shareholdingSnapshots"]] == [period.date().isoformat() for period in periods] |
| 232 | |
| 233 | |
| 234 | class _GlobalInstrumentClient: |
| 235 | def __init__(self, response: httpx.Response | Exception) -> None: |
| 236 | self.response = response |
| 237 | self.calls: list[dict] = [] |
| 238 | |
| 239 | async def get(self, url: str, headers: dict | None = None) -> httpx.Response: |
| 240 | self.calls.append({"method": "GET", "url": url, "headers": headers or {}}) |
| 241 | if isinstance(self.response, Exception): |
| 242 | raise self.response |
| 243 | return self.response |
| 244 | |
| 245 | async def post(self, url: str, headers: dict | None = None) -> httpx.Response: |
| 246 | self.calls.append({"method": "POST", "url": url, "headers": headers or {}}) |
| 247 | if isinstance(self.response, Exception): |
| 248 | raise self.response |
| 249 | return self.response |
| 250 | |
| 251 | |
| 252 | def _global_master_payload(global_id: UUID) -> dict: |
| 253 | return { |
| 254 | "globalInstrumentId": str(global_id), "canonicalName": "Generic Components Limited", |
| 255 | "isin": "INE000A01010", "assetType": "EQUITY", "country": "IN", "currency": "INR", |
| 256 | "primaryExchange": "NSE", "primarySymbol": "GENERIC", |
| 257 | "providerMappings": [ |
| 258 | {"provider": "NSE", "providerSymbol": "GENERIC", "status": "VERIFIED", "exchange": "NSE"}, |
| 259 | {"provider": "YAHOO_FINANCE", "providerSymbol": "GENERIC.NS", "status": "VERIFIED", "exchange": "NSE"}, |
| 260 | {"provider": "NSE", "providerSymbol": "BAD_ALIAS", "status": "INVALID", "exchange": "NSE"}, |
| 261 | ], |
| 262 | } |
| 263 | |
| 264 | |
| 265 | def test_global_master_api_restores_unknown_profile_with_verified_mappings() -> None: |
| 266 | global_id = uuid4() |
| 267 | url = f"http://portfolio-service/api/v1/instruments/{global_id}" |
| 268 | client = _GlobalInstrumentClient(httpx.Response(200, json=_global_master_payload(global_id), request=httpx.Request("GET", url))) |
| 269 | repo = ResearchRepository(settings=Settings(research_demo_enabled=False)) |
| 270 | orchestrator = PortfolioResearchOrchestrator(repo, Settings(portfolio_service_base_url="http://portfolio-service"), client=client) |
| 271 | |
| 272 | assert asyncio.run(orchestrator.restore_global_profile(global_id, identity_headers={"X-AIP-User-Id": "user"})) is True |
| 273 | profile = repo.profile(global_id) |
| 274 | assert profile.instrument_id == global_id |
| 275 | assert profile.provider_instrument_ids["NSE"] == "GENERIC" |
| 276 | assert profile.provider_instrument_ids["YAHOO_FINANCE"] == "GENERIC.NS" |
| 277 | assert "BAD_ALIAS" not in profile.provider_instrument_ids.values() |
| 278 | assert client.calls == [{"method": "GET", "url": url, "headers": {"X-AIP-User-Id": "user"}}] |
| 279 | |
| 280 | |
| 281 | def test_global_master_restore_preserves_verified_international_mapping_for_existing_drawer_path() -> None: |
| 282 | global_id = uuid4(); payload = _global_master_payload(global_id) |
| 283 | payload.update({"canonicalName": "Aalberts N.V.", "isin": "NL0000852564", "country": "NL", "currency": "EUR", "primaryExchange": "XAMS", "primarySymbol": "AALB"}) |
| 284 | payload["providerMappings"] = [{"provider": "EODHD", "providerSymbol": "AALB.AS", "providerInstrumentId": "AALB.AS", "status": "VERIFIED", "exchange": "XAMS", "currency": "EUR"}] |
| 285 | url = f"http://portfolio-service/api/v1/instruments/{global_id}" |
| 286 | client = _GlobalInstrumentClient(httpx.Response(200, json=payload, request=httpx.Request("GET", url))) |
| 287 | repo = ResearchRepository(settings=Settings(research_demo_enabled=False)) |
| 288 | assert asyncio.run(PortfolioResearchOrchestrator(repo, Settings(portfolio_service_base_url="http://portfolio-service"), client=client).restore_global_profile(global_id)) |
| 289 | assert repo.profile(global_id).provider_instrument_ids == {"EODHD": "AALB.AS"} |
| 290 | |
| 291 | |
| 292 | def test_direct_summary_restores_from_global_master_without_portfolio_hydration(monkeypatch) -> None: |
| 293 | global_id = uuid4() |
| 294 | url = f"http://portfolio-service/api/v1/instruments/{global_id}" |
| 295 | lookup_client = _GlobalInstrumentClient(httpx.Response(200, json=_global_master_payload(global_id), request=httpx.Request("GET", url))) |
| 296 | repo = ResearchRepository(settings=Settings(research_demo_enabled=False)) |
| 297 | orchestrator = PortfolioResearchOrchestrator(repo, Settings(portfolio_service_base_url="http://portfolio-service"), client=lookup_client) |
| 298 | monkeypatch.setattr(main, "repository", repo) |
| 299 | monkeypatch.setattr(main, "portfolio_orchestrator", orchestrator) |
| 300 | |
| 301 | response = TestClient(app).get(f"/api/v1/research/companies/{global_id}/summary", headers={"X-AIP-User-Id": "user"}) |
| 302 | |
| 303 | assert response.status_code == 200 |
| 304 | assert response.json()["profile"]["instrumentId"] == str(global_id) |
| 305 | assert len(lookup_client.calls) == 1 |
| 306 | assert lookup_client.calls[0]["method"] == "GET" |
| 307 | |
| 308 | |
| 309 | def test_global_master_404_is_unresolved_and_transient_failure_is_not_cached() -> None: |
| 310 | global_id = uuid4() |
| 311 | url = f"http://portfolio-service/api/v1/instruments/{global_id}" |
| 312 | repo = ResearchRepository(settings=Settings(research_demo_enabled=False)) |
| 313 | missing = PortfolioResearchOrchestrator(repo, Settings(portfolio_service_base_url="http://portfolio-service"), client=_GlobalInstrumentClient(httpx.Response(404, request=httpx.Request("GET", url)))) |
| 314 | # The orchestrator's 404 is explicit and does not register an alias profile. |
| 315 | from app.portfolio_orchestration import GlobalInstrumentNotFoundError |
| 316 | try: |
| 317 | asyncio.run(missing.restore_global_profile(global_id)) |
| 318 | assert False, "expected global master 404" |
| 319 | except GlobalInstrumentNotFoundError: |
| 320 | pass |
| 321 | failing = PortfolioResearchOrchestrator(repo, Settings(portfolio_service_base_url="http://portfolio-service"), client=_GlobalInstrumentClient(httpx.ConnectError("down"))) |
| 322 | try: |
| 323 | asyncio.run(failing.restore_global_profile(global_id)) |
| 324 | assert False, "expected service-unavailable outcome" |
| 325 | except PortfolioServiceUnavailableError: |
| 326 | pass |
| 327 | assert all(profile.instrument_id != global_id for profile in repo.list_profiles()) |