| 1 | from __future__ import annotations |
| 2 | |
| 3 | import asyncio |
| 4 | from datetime import datetime, timedelta, timezone |
| 5 | from decimal import Decimal |
| 6 | from types import SimpleNamespace |
| 7 | from uuid import UUID, uuid4 |
| 8 | |
| 9 | import httpx |
| 10 | import pytest |
| 11 | from fastapi.testclient import TestClient |
| 12 | |
| 13 | import app.main as main |
| 14 | from app.fact_precedence import FactSourceTier, FinancialFact, FinancialFactKey |
| 15 | from app.models import ( |
| 16 | CompanyResearchProfile, |
| 17 | DocumentStatus, |
| 18 | EventImpact, |
| 19 | MarketPriceObservation, |
| 20 | ProvenancedValue, |
| 21 | ReliabilityLevel, |
| 22 | ResearchEvent, |
| 23 | ResearchEventType, |
| 24 | ResearchLifecycleStatus, |
| 25 | ShareholdingCategory, |
| 26 | ShareholdingSnapshot, |
| 27 | ShareholdingSnapshotValue, |
| 28 | SourceClassification, |
| 29 | SourceMode, |
| 30 | SourceType, |
| 31 | StructuredInstrumentResolution, |
| 32 | StructuredMarketSnapshot, |
| 33 | StructuredMarketSnapshotRecord, |
| 34 | TimeHorizon, |
| 35 | ) |
| 36 | from app.persistence import SqliteResearchPersistence |
| 37 | from app.portfolio_orchestration import PortfolioResearchOrchestrator |
| 38 | from app.repository import ResearchRepository, _TRUSTED_NSE_PROFILE_IDENTITY |
| 39 | from app.research_readiness import ( |
| 40 | DurableResearchSnapshot, |
| 41 | ProviderAuthorityRegistry, |
| 42 | ResearchEvidence, |
| 43 | ResearchReadinessService, |
| 44 | ResearchRefreshTarget, |
| 45 | ResearchRequirementRegistry, |
| 46 | ResearchRequirementStatus, |
| 47 | ResearchSourceTier, |
| 48 | ) |
| 49 | from app.research_readiness_runtime import ( |
| 50 | CapabilityExecutionResult, |
| 51 | ExistingResearchCapabilityExecutor, |
| 52 | RepositoryResearchReadinessAdapter, |
| 53 | ResearchReadinessRuntime, |
| 54 | jurisdiction_for_profile, |
| 55 | readiness_response, |
| 56 | ) |
| 57 | from app.settings import Settings |
| 58 | |
| 59 | |
| 60 | NOW = datetime(2026, 9, 10, 12, 0, tzinfo=timezone.utc) |
| 61 | INSTRUMENT_ID = UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb") |
| 62 | SOURCE_URL = "https://nsearchives.nseindia.com/corporate/quarterly-result.pdf" |
| 63 | |
| 64 | |
| 65 | def _profile(instrument_id: UUID = INSTRUMENT_ID) -> CompanyResearchProfile: |
| 66 | return CompanyResearchProfile( |
| 67 | instrument_id=instrument_id, |
| 68 | company_id=uuid4(), |
| 69 | company_name="Readiness India Limited", |
| 70 | ticker="READY", |
| 71 | exchange="NSE", |
| 72 | mic="XNSE", |
| 73 | country="IN", |
| 74 | currency="INR", |
| 75 | isin="INE000A01010", |
| 76 | provider_instrument_ids={"NSE": "READY", "YAHOO_FINANCE": "READY.NS"}, |
| 77 | ) |
| 78 | |
| 79 | |
| 80 | def test_global_metadata_registration_keeps_canonical_id_when_demo_isin_matches() -> None: |
| 81 | repository = ResearchRepository( |
| 82 | settings=Settings(research_live_enabled=False, research_demo_enabled=True) |
| 83 | ) |
| 84 | orchestrator = PortfolioResearchOrchestrator(repository, repository.settings) |
| 85 | canonical_id = UUID("6c6e3c9f-9d08-421b-a3ce-72589f57e23a") |
| 86 | |
| 87 | registered = orchestrator.register_global_profile_metadata( |
| 88 | canonical_id, |
| 89 | { |
| 90 | "globalInstrumentId": str(canonical_id), |
| 91 | "canonicalName": "Reliance Industries Ltd.", |
| 92 | "isin": "INE002A01018", |
| 93 | "assetType": "EQUITY", |
| 94 | "currency": "INR", |
| 95 | "country": "IN", |
| 96 | "primaryExchange": "NSE", |
| 97 | "primarySymbol": "RELIANCE", |
| 98 | "status": "ACTIVE", |
| 99 | "providerMappings": [ |
| 100 | { |
| 101 | "provider": "NSE", |
| 102 | "providerSymbol": "RELIANCE", |
| 103 | "providerInstrumentId": "INE002A01018", |
| 104 | "exchange": "NSE", |
| 105 | "currency": "INR", |
| 106 | "status": "VERIFIED", |
| 107 | } |
| 108 | ], |
| 109 | }, |
| 110 | ) |
| 111 | |
| 112 | assert registered is True |
| 113 | assert repository.profile(canonical_id).instrument_id == canonical_id |
| 114 | assert repository.profile(canonical_id).provider_instrument_ids["NSE"] == "RELIANCE" |
| 115 | |
| 116 | |
| 117 | def _fact( |
| 118 | profile: CompanyResearchProfile, |
| 119 | metric: str, |
| 120 | value: str, |
| 121 | period_end: str, |
| 122 | period_type: str, |
| 123 | ) -> FinancialFact: |
| 124 | return FinancialFact( |
| 125 | FinancialFactKey( |
| 126 | profile.instrument_id, metric, period_end, period_type, "CONSOLIDATED" |
| 127 | ), |
| 128 | ProvenancedValue( |
| 129 | value=Decimal(value), |
| 130 | unit="INR crore", |
| 131 | as_of_date=datetime.fromisoformat(period_end).replace(tzinfo=timezone.utc), |
| 132 | source_url=SOURCE_URL, |
| 133 | source_name="NSE", |
| 134 | source_type="EXCHANGE_ANNOUNCEMENT", |
| 135 | published_at=NOW - timedelta(days=2), |
| 136 | retrieved_at=NOW - timedelta(hours=1), |
| 137 | confidence=0.98, |
| 138 | ), |
| 139 | FactSourceTier.OFFICIAL_NSE, |
| 140 | "NSE", |
| 141 | f"nse-{metric}-{period_end}-{period_type}", |
| 142 | SourceMode.REAL, |
| 143 | ) |
| 144 | |
| 145 | |
| 146 | def _structured_record(profile: CompanyResearchProfile) -> StructuredMarketSnapshotRecord: |
| 147 | fact_values = { |
| 148 | "latestPrice": "250", |
| 149 | "marketCap": "100000", |
| 150 | "trailingEps": "12", |
| 151 | "trailingPE": "20", |
| 152 | "priceToBook": "3", |
| 153 | "evToEbitda": "11", |
| 154 | "freeCashFlow": "450", |
| 155 | "roe": "18", |
| 156 | "profitMargin": "14", |
| 157 | "operatingCashFlow": "700", |
| 158 | "revenueGrowth": "12", |
| 159 | "earningsGrowth": "9", |
| 160 | "totalDebt": "1000", |
| 161 | "bookValue": "80", |
| 162 | "totalCash": "500", |
| 163 | "currentRatio": "1.5", |
| 164 | } |
| 165 | facts = { |
| 166 | key: ProvenancedValue( |
| 167 | value=Decimal(value), |
| 168 | source_url="https://finance.yahoo.com/quote/READY.NS", |
| 169 | source_name="Yahoo Finance", |
| 170 | source_type="STRUCTURED_MARKET_PROVIDER", |
| 171 | as_of_date=NOW - timedelta(minutes=5), |
| 172 | retrieved_at=NOW - timedelta(minutes=4), |
| 173 | confidence=0.9, |
| 174 | ) |
| 175 | for key, value in fact_values.items() |
| 176 | } |
| 177 | facts["sector"] = ProvenancedValue( |
| 178 | value="Industrials", |
| 179 | source_url="https://finance.yahoo.com/quote/READY.NS", |
| 180 | source_name="Yahoo Finance", |
| 181 | retrieved_at=NOW - timedelta(hours=1), |
| 182 | ) |
| 183 | resolution = StructuredInstrumentResolution( |
| 184 | instrument_id=profile.instrument_id, |
| 185 | provider="YAHOO_FINANCE", |
| 186 | provider_ticker="READY.NS", |
| 187 | company_name=profile.company_name, |
| 188 | exchange="NSE", |
| 189 | currency="INR", |
| 190 | confidence=0.99, |
| 191 | resolved_at=NOW - timedelta(hours=1), |
| 192 | ) |
| 193 | snapshot = StructuredMarketSnapshot( |
| 194 | resolution=resolution, |
| 195 | status="SUCCESS", |
| 196 | retrieved_at=NOW - timedelta(minutes=4), |
| 197 | market_as_of=NOW - timedelta(minutes=5), |
| 198 | source_url="https://finance.yahoo.com/quote/READY.NS", |
| 199 | facts=facts, |
| 200 | ) |
| 201 | return StructuredMarketSnapshotRecord( |
| 202 | instrument_id=profile.instrument_id, |
| 203 | provider="YAHOO_FINANCE", |
| 204 | provider_instrument_id="READY.NS", |
| 205 | exchange="NSE", |
| 206 | currency="INR", |
| 207 | source_url=snapshot.source_url, |
| 208 | retrieved_at=snapshot.retrieved_at, |
| 209 | persisted_at=snapshot.retrieved_at, |
| 210 | last_price_at=snapshot.retrieved_at, |
| 211 | last_valuation_at=snapshot.retrieved_at, |
| 212 | last_fundamentals_at=snapshot.retrieved_at, |
| 213 | last_success_at=snapshot.retrieved_at, |
| 214 | snapshot=snapshot, |
| 215 | ) |
| 216 | |
| 217 | |
| 218 | def _event( |
| 219 | profile: CompanyResearchProfile, |
| 220 | event_type: ResearchEventType, |
| 221 | *, |
| 222 | age_days: int, |
| 223 | title: str, |
| 224 | impact: EventImpact = EventImpact.POSITIVE, |
| 225 | source_url: str | None = None, |
| 226 | ) -> ResearchEvent: |
| 227 | event_at = NOW - timedelta(days=age_days) |
| 228 | return ResearchEvent( |
| 229 | instrument_id=profile.instrument_id, |
| 230 | company_id=profile.company_id, |
| 231 | event_type=event_type, |
| 232 | event_date=event_at, |
| 233 | detected_at=event_at, |
| 234 | title=title, |
| 235 | summary=title, |
| 236 | source_document_id=uuid4(), |
| 237 | source_url=source_url or f"https://news.example/{title.replace(' ', '-').lower()}", |
| 238 | source_type=SourceType.NEWS, |
| 239 | source_classification=SourceClassification.REPUTABLE_NEWS, |
| 240 | reliability=ReliabilityLevel.LEVEL_B, |
| 241 | source_mode=SourceMode.REAL, |
| 242 | confidence=0.85, |
| 243 | impact=impact, |
| 244 | time_horizon=TimeHorizon.SHORT_TERM, |
| 245 | status=ResearchLifecycleStatus.VALIDATED, |
| 246 | raw_evidence_reference=title, |
| 247 | published_at=event_at, |
| 248 | retrieved_at=event_at, |
| 249 | ) |
| 250 | |
| 251 | |
| 252 | def _shareholding(profile: CompanyResearchProfile) -> ShareholdingSnapshot: |
| 253 | return ShareholdingSnapshot( |
| 254 | instrument_id=profile.instrument_id, |
| 255 | period_end=NOW - timedelta(days=45), |
| 256 | source_provider="NSE", |
| 257 | source_type="NSE_SHAREHOLDING_XBRL", |
| 258 | source_identity_key="NSE_SHAREHOLDING:READY:2026Q2", |
| 259 | source_url="https://nsearchives.nseindia.com/shareholding/ready.xml", |
| 260 | published_at=NOW - timedelta(days=35), |
| 261 | retrieved_at=NOW - timedelta(days=34), |
| 262 | confidence=Decimal("0.99"), |
| 263 | reliability_level=ReliabilityLevel.LEVEL_A, |
| 264 | source_mode=SourceMode.REAL, |
| 265 | values=[ |
| 266 | ShareholdingSnapshotValue( |
| 267 | category=ShareholdingCategory.PROMOTER, percentage=Decimal("51.2") |
| 268 | ), |
| 269 | ShareholdingSnapshotValue( |
| 270 | category=ShareholdingCategory.FII_FPI, percentage=Decimal("12.4") |
| 271 | ), |
| 272 | ], |
| 273 | ) |
| 274 | |
| 275 | |
| 276 | class DurableRepositoryFixture: |
| 277 | def __init__(self, profile: CompanyResearchProfile, *, complete: bool = True) -> None: |
| 278 | self.profiles = {profile.instrument_id: profile} |
| 279 | self.provider_calls = 0 |
| 280 | self.portfolio_mutations = 0 |
| 281 | self.watchlist_mutations = 0 |
| 282 | self.facts = [] |
| 283 | self.structured = [] |
| 284 | self.observations = [] |
| 285 | self.documents = [] |
| 286 | self.events = [] |
| 287 | self.shareholding = [] |
| 288 | if complete: |
| 289 | self.facts = [ |
| 290 | _fact(profile, "revenue", "100", "2026-08-31", "QUARTERLY"), |
| 291 | _fact(profile, "pat", "12", "2026-08-31", "QUARTERLY"), |
| 292 | _fact(profile, "eps", "5", "2026-08-31", "QUARTERLY"), |
| 293 | _fact(profile, "revenue", "90", "2026-05-31", "QUARTERLY"), |
| 294 | _fact(profile, "pat", "10", "2026-05-31", "QUARTERLY"), |
| 295 | _fact(profile, "revenue", "360", "2026-03-31", "ANNUAL"), |
| 296 | _fact(profile, "pat", "40", "2026-03-31", "ANNUAL"), |
| 297 | _fact(profile, "revenue", "320", "2025-03-31", "ANNUAL"), |
| 298 | _fact(profile, "pat", "35", "2025-03-31", "ANNUAL"), |
| 299 | _fact(profile, "total_debt", "1000", "2026-03-31", "ANNUAL"), |
| 300 | _fact(profile, "total_equity", "2500", "2026-03-31", "ANNUAL"), |
| 301 | _fact(profile, "cash_and_cash_equivalents", "500", "2026-03-31", "ANNUAL"), |
| 302 | ] |
| 303 | self.structured = [_structured_record(profile)] |
| 304 | self.observations = [ |
| 305 | MarketPriceObservation( |
| 306 | instrument_id=profile.instrument_id, |
| 307 | observed_at=NOW - timedelta(days=149 - index, hours=1), |
| 308 | price=Decimal(100 + index), |
| 309 | currency="INR", |
| 310 | provider="YAHOO_FINANCE", |
| 311 | source_url="https://finance.yahoo.com/quote/READY.NS/history", |
| 312 | retrieved_at=NOW - timedelta(minutes=3), |
| 313 | ) |
| 314 | for index in range(150) |
| 315 | ] |
| 316 | self.events = [ |
| 317 | _event(profile, ResearchEventType.NEW_ORDER, age_days=2, title="Major order"), |
| 318 | _event( |
| 319 | profile, |
| 320 | ResearchEventType.REGULATORY_EVENT, |
| 321 | age_days=3, |
| 322 | title="Regulatory review", |
| 323 | impact=EventImpact.NEGATIVE, |
| 324 | ), |
| 325 | ] |
| 326 | self.shareholding = [_shareholding(profile)] |
| 327 | |
| 328 | def profile(self, instrument_id): |
| 329 | return self.profiles[instrument_id] |
| 330 | |
| 331 | def financial_facts_for(self, instrument_id): |
| 332 | return [value for value in self.facts if value.key.instrument_id == instrument_id] |
| 333 | |
| 334 | def structured_market_snapshots_for(self, instrument_ids): |
| 335 | return {value: [item for item in self.structured if item.instrument_id == value] for value in instrument_ids} |
| 336 | |
| 337 | def market_price_observations_for(self, instrument_ids): |
| 338 | return {value: [item for item in self.observations if item.instrument_id == value] for value in instrument_ids} |
| 339 | |
| 340 | def documents_for(self, instrument_id, source_mode=None): |
| 341 | return [item for item in self.documents if item.instrument_id == instrument_id] |
| 342 | |
| 343 | def events_for(self, instrument_id, source_mode=None): |
| 344 | return [item for item in self.events if item.instrument_id == instrument_id] |
| 345 | |
| 346 | def shareholding_for(self, instrument_id, limit=4): |
| 347 | return [item for item in self.shareholding if item.instrument_id == instrument_id][:limit] |
| 348 | |
| 349 | async def _run_blocking_persistence(self, operation, *args, **kwargs): |
| 350 | return operation(*args, **kwargs) |
| 351 | |
| 352 | |
| 353 | def test_durable_adapter_maps_all_rule_areas_without_portfolio_context() -> None: |
| 354 | profile = _profile() |
| 355 | repository = DurableRepositoryFixture(profile) |
| 356 | adapter = RepositoryResearchReadinessAdapter(repository) |
| 357 | adapter.remember_canonical_metadata( |
| 358 | profile.instrument_id, |
| 359 | { |
| 360 | "globalInstrumentId": str(profile.instrument_id), |
| 361 | "canonicalSector": "Industrials", |
| 362 | "updatedAt": NOW.isoformat(), |
| 363 | "quantity": 500, |
| 364 | "portfolioId": str(uuid4()), |
| 365 | }, |
| 366 | ) |
| 367 | |
| 368 | result = ResearchReadinessService(adapter).assess( |
| 369 | profile.instrument_id, jurisdiction="INDIA", now=NOW |
| 370 | ) |
| 371 | |
| 372 | assert {item.rule_engine_area for item in result.requirements} == { |
| 373 | item.rule_engine_area for item in ResearchRequirementRegistry.default().requirements |
| 374 | } |
| 375 | assert result.for_requirement("QUARTERLY_FINANCIALS").status == ResearchRequirementStatus.READY_FRESH |
| 376 | assert result.for_requirement("QUARTERLY_FINANCIALS").source_url == SOURCE_URL |
| 377 | assert result.for_requirement("HISTORICAL_PRICE_SERIES").coverage_pct == 100 |
| 378 | assert result.for_requirement("SHAREHOLDING").status == ResearchRequirementStatus.READY_FRESH |
| 379 | assert result.for_requirement("SECTOR_MACRO").status == ResearchRequirementStatus.READY_FRESH |
| 380 | assert repository.provider_calls == 0 |
| 381 | assert repository.portfolio_mutations == 0 |
| 382 | assert repository.watchlist_mutations == 0 |
| 383 | |
| 384 | |
| 385 | def test_held_metadata_fields_do_not_change_public_readiness() -> None: |
| 386 | first, second = _profile(uuid4()), _profile(uuid4()) |
| 387 | repo = DurableRepositoryFixture(first, complete=False) |
| 388 | repo.profiles[second.instrument_id] = second |
| 389 | adapter = RepositoryResearchReadinessAdapter(repo) |
| 390 | base = {"canonicalSector": "Industrials", "updatedAt": NOW.isoformat()} |
| 391 | adapter.remember_canonical_metadata(first.instrument_id, {**base, "quantity": 20, "portfolioId": str(uuid4())}) |
| 392 | adapter.remember_canonical_metadata(second.instrument_id, {**base, "quantity": 0, "held": False}) |
| 393 | service = ResearchReadinessService(adapter) |
| 394 | |
| 395 | held = service.assess(first.instrument_id, jurisdiction="INDIA", now=NOW) |
| 396 | non_held = service.assess(second.instrument_id, jurisdiction="INDIA", now=NOW) |
| 397 | |
| 398 | assert [(item.requirement_id, item.status) for item in held.requirements] == [ |
| 399 | (item.requirement_id, item.status) for item in non_held.requirements |
| 400 | ] |
| 401 | assert repo.portfolio_mutations == repo.watchlist_mutations == 0 |
| 402 | |
| 403 | |
| 404 | def test_news_boundary_deduplication_relevance_and_governance_history() -> None: |
| 405 | profile = _profile() |
| 406 | other = _profile(uuid4()) |
| 407 | repo = DurableRepositoryFixture(profile, complete=False) |
| 408 | boundary = _event( |
| 409 | profile, |
| 410 | ResearchEventType.NEW_ORDER, |
| 411 | age_days=30, |
| 412 | title="Boundary event", |
| 413 | source_url="https://news.example/boundary", |
| 414 | ) |
| 415 | duplicate = boundary.model_copy(update={"event_id": uuid4(), "source_document_id": uuid4()}) |
| 416 | old_news = _event(profile, ResearchEventType.NEW_ORDER, age_days=31, title="Old event") |
| 417 | old_governance = _event( |
| 418 | profile, |
| 419 | ResearchEventType.REGULATORY_EVENT, |
| 420 | age_days=800, |
| 421 | title="Unresolved fraud investigation", |
| 422 | impact=EventImpact.NEGATIVE, |
| 423 | ) |
| 424 | irrelevant = _event(other, ResearchEventType.REGULATORY_EVENT, age_days=1, title="Unrelated war exposure") |
| 425 | repo.events = [boundary, duplicate, old_news, old_governance, irrelevant] |
| 426 | adapter = RepositoryResearchReadinessAdapter(repo) |
| 427 | snapshot = adapter.load_by_global_instrument_id( |
| 428 | profile.instrument_id, ResearchRequirementRegistry.default().requirements |
| 429 | ) |
| 430 | result = ResearchReadinessService(adapter).assess( |
| 431 | profile.instrument_id, jurisdiction="INDIA", now=NOW |
| 432 | ) |
| 433 | |
| 434 | current = result.for_requirement("CURRENT_NEWS") |
| 435 | assert current.status == ResearchRequirementStatus.READY_STALE # Still eligible at 30 days; daily acquisition is stale. |
| 436 | assert len(current.evidence_ids) == 1 |
| 437 | assert "event:" + str(old_news.event_id) not in "|".join(current.evidence_ids) |
| 438 | assert all(str(irrelevant.event_id) not in item.evidence_id for item in snapshot.evidence_for("CURRENT_NEWS")) |
| 439 | governance = snapshot.evidence_for("GOVERNANCE_HISTORY") |
| 440 | assert any(str(old_governance.event_id) in item.evidence_id and item.unresolved for item in governance) |
| 441 | assert result.for_requirement("GOVERNANCE_HISTORY").status == ResearchRequirementStatus.READY_FRESH |
| 442 | |
| 443 | |
| 444 | def _target(requirement_id: str, jurisdiction: str = "INDIA") -> ResearchRefreshTarget: |
| 445 | requirement = ResearchRequirementRegistry.default().get(requirement_id) |
| 446 | return ResearchRefreshTarget( |
| 447 | requirement_id, |
| 448 | requirement.rule_engine_area, |
| 449 | ResearchRequirementStatus.MISSING, |
| 450 | ProviderAuthorityRegistry.default().policy_for(requirement_id, jurisdiction), |
| 451 | (), |
| 452 | ) |
| 453 | |
| 454 | |
| 455 | class RecordingTargetRepository: |
| 456 | def __init__(self, profile: CompanyResearchProfile) -> None: |
| 457 | self._profile = profile |
| 458 | self.category_calls: list[set[str]] = [] |
| 459 | self.settings = Settings(market_data_population_initial_lookback_days=400) |
| 460 | |
| 461 | def profile(self, _instrument_id): |
| 462 | return self._profile |
| 463 | |
| 464 | async def refresh_targeted_categories(self, _instrument_id, categories, **_kwargs): |
| 465 | self.category_calls.append(set(categories)) |
| 466 | |
| 467 | async def market_price_observations_for_instruments(self, ids): |
| 468 | return {value: [] for value in ids} |
| 469 | |
| 470 | |
| 471 | class RecordingOrchestrator: |
| 472 | def __init__(self) -> None: |
| 473 | self.structured_calls: list[set[str]] = [] |
| 474 | self.international_calls = 0 |
| 475 | |
| 476 | async def ensure_structured_market(self, _instrument_id, classes): |
| 477 | self.structured_calls.append(set(classes)) |
| 478 | return SimpleNamespace(error=None) |
| 479 | |
| 480 | async def refresh_international_fundamentals(self, *_args, **_kwargs): |
| 481 | self.international_calls += 1 |
| 482 | return SimpleNamespace(facts=[object()]) |
| 483 | |
| 484 | |
| 485 | class RecordingPopulation: |
| 486 | def __init__(self) -> None: |
| 487 | self.calls = [] |
| 488 | |
| 489 | async def populate(self, instruments, **kwargs): |
| 490 | self.calls.append((instruments, kwargs)) |
| 491 | return 1 |
| 492 | |
| 493 | |
| 494 | def _capability_executor(): |
| 495 | repo = RecordingTargetRepository(_profile()) |
| 496 | orchestrator = RecordingOrchestrator() |
| 497 | jobs = SimpleNamespace( |
| 498 | population=RecordingPopulation(), |
| 499 | settings=repo.settings, |
| 500 | ) |
| 501 | return ExistingResearchCapabilityExecutor(repo, orchestrator, jobs), repo, orchestrator, jobs |
| 502 | |
| 503 | |
| 504 | @pytest.mark.asyncio |
| 505 | async def test_only_news_missing_runs_only_existing_global_search_path() -> None: |
| 506 | executor, repo, orchestrator, jobs = _capability_executor() |
| 507 | result = await executor.execute_primary( |
| 508 | INSTRUMENT_ID, |
| 509 | [_target("CURRENT_NEWS")], |
| 510 | jurisdiction="INDIA", |
| 511 | correlation_id=None, |
| 512 | identity_headers=None, |
| 513 | ) |
| 514 | |
| 515 | assert result.executed_capabilities == ("GLOBAL_NEWS_SEARCH",) |
| 516 | assert repo.category_calls == [{"CATALYSTS", "RISKS", "REGULATORY", "MANAGEMENT", "GUIDANCE"}] |
| 517 | assert orchestrator.structured_calls == [] |
| 518 | assert jobs.population.calls == [] |
| 519 | |
| 520 | |
| 521 | @pytest.mark.asyncio |
| 522 | async def test_only_shareholding_stale_runs_only_nse_shareholding_path() -> None: |
| 523 | executor, repo, orchestrator, jobs = _capability_executor() |
| 524 | result = await executor.execute_primary( |
| 525 | INSTRUMENT_ID, |
| 526 | [_target("SHAREHOLDING")], |
| 527 | jurisdiction="INDIA", |
| 528 | correlation_id=None, |
| 529 | identity_headers=None, |
| 530 | ) |
| 531 | |
| 532 | assert result.executed_capabilities == ("SHAREHOLDING",) |
| 533 | assert repo.category_calls == [{"SHAREHOLDING_PATTERN"}] |
| 534 | assert orchestrator.structured_calls == [] |
| 535 | assert jobs.population.calls == [] |
| 536 | |
| 537 | |
| 538 | @pytest.mark.asyncio |
| 539 | async def test_financials_and_news_group_only_their_existing_capabilities() -> None: |
| 540 | executor, repo, orchestrator, _jobs = _capability_executor() |
| 541 | result = await executor.execute_primary( |
| 542 | INSTRUMENT_ID, |
| 543 | [_target("QUARTERLY_FINANCIALS"), _target("CURRENT_NEWS")], |
| 544 | jurisdiction="INDIA", |
| 545 | correlation_id="targeted", |
| 546 | identity_headers=None, |
| 547 | ) |
| 548 | |
| 549 | assert result.executed_capabilities == ("FINANCIALS", "GLOBAL_NEWS_SEARCH") |
| 550 | assert repo.category_calls == [{ |
| 551 | "FINANCIAL_RESULTS", "CATALYSTS", "RISKS", "REGULATORY", "MANAGEMENT", "GUIDANCE" |
| 552 | }] |
| 553 | assert orchestrator.structured_calls == [] |
| 554 | |
| 555 | |
| 556 | class StateDataSource: |
| 557 | def __init__(self, missing: set[str]) -> None: |
| 558 | self.missing = set(missing) |
| 559 | self.loads = 0 |
| 560 | self.failures: dict[str, str] = {} |
| 561 | |
| 562 | def load_by_global_instrument_id(self, instrument_id, requirements): |
| 563 | self.loads += 1 |
| 564 | now = datetime.now(timezone.utc) |
| 565 | evidence = { |
| 566 | item.requirement_id: ( |
| 567 | ResearchEvidence( |
| 568 | evidence_id=f"ready:{item.requirement_id}", |
| 569 | requirement_id=item.requirement_id, |
| 570 | source="NSE", |
| 571 | source_tier=ResearchSourceTier.OFFICIAL, |
| 572 | retrieved_at=now - timedelta(minutes=1), |
| 573 | as_of=now - timedelta(minutes=1), |
| 574 | event_date=now - timedelta(minutes=1), |
| 575 | published_at=now - timedelta(minutes=1), |
| 576 | ), |
| 577 | ) |
| 578 | for item in requirements |
| 579 | if item.requirement_id not in self.missing |
| 580 | } |
| 581 | return DurableResearchSnapshot( |
| 582 | instrument_id, evidence, failure_reasons=self.failures |
| 583 | ) |
| 584 | |
| 585 | def mark_refreshing(self, _instrument_id, _requirement_ids): |
| 586 | return None |
| 587 | |
| 588 | def finish_refresh(self, _instrument_id, failures=None): |
| 589 | self.failures = dict(failures or {}) |
| 590 | |
| 591 | |
| 592 | class RuntimeRepository: |
| 593 | async def _run_blocking_persistence(self, operation, *args, **kwargs): |
| 594 | return operation(*args, **kwargs) |
| 595 | |
| 596 | |
| 597 | class UpdatingExecutor: |
| 598 | def __init__(self, source: StateDataSource, *, update_primary: bool = True) -> None: |
| 599 | self.source = source |
| 600 | self.update_primary = update_primary |
| 601 | self.primary_calls: list[set[str]] = [] |
| 602 | self.fallback_calls: list[set[str]] = [] |
| 603 | self.started = asyncio.Event() |
| 604 | self.release: asyncio.Event | None = None |
| 605 | |
| 606 | async def execute_primary(self, _instrument_id, targets, **_kwargs): |
| 607 | ids = {target.requirement_id for target in targets} |
| 608 | self.primary_calls.append(ids) |
| 609 | self.started.set() |
| 610 | if self.release is not None: |
| 611 | await self.release.wait() |
| 612 | if self.update_primary: |
| 613 | self.source.missing.difference_update(ids) |
| 614 | return CapabilityExecutionResult(("RECORDED_PRIMARY",), {}) |
| 615 | |
| 616 | async def execute_approved_fallbacks(self, _instrument_id, targets): |
| 617 | ids = {target.requirement_id for target in targets} |
| 618 | self.fallback_calls.append(ids) |
| 619 | self.source.missing.difference_update(ids) |
| 620 | return CapabilityExecutionResult(("RECORDED_FALLBACK",), {}) |
| 621 | |
| 622 | |
| 623 | def _runtime(source: StateDataSource, executor: UpdatingExecutor) -> ResearchReadinessRuntime: |
| 624 | return ResearchReadinessRuntime( |
| 625 | RuntimeRepository(), |
| 626 | source, # type: ignore[arg-type] |
| 627 | executor, # type: ignore[arg-type] |
| 628 | ) |
| 629 | |
| 630 | |
| 631 | @pytest.mark.asyncio |
| 632 | async def test_all_fresh_ensure_executes_zero_provider_capabilities() -> None: |
| 633 | source = StateDataSource(set()) |
| 634 | executor = UpdatingExecutor(source) |
| 635 | result = await _runtime(source, executor).ensure( |
| 636 | INSTRUMENT_ID, |
| 637 | jurisdiction="INDIA", |
| 638 | requirement_ids=None, |
| 639 | ) |
| 640 | |
| 641 | assert result.planned_requirement_ids == () |
| 642 | assert result.executed_capabilities == () |
| 643 | assert executor.primary_calls == executor.fallback_calls == [] |
| 644 | |
| 645 | |
| 646 | @pytest.mark.asyncio |
| 647 | async def test_runtime_targets_only_selected_missing_requirement_and_skips_fresh() -> None: |
| 648 | source = StateDataSource({"CURRENT_NEWS", "SHAREHOLDING"}) |
| 649 | executor = UpdatingExecutor(source) |
| 650 | result = await _runtime(source, executor).ensure( |
| 651 | INSTRUMENT_ID, |
| 652 | jurisdiction="INDIA", |
| 653 | requirement_ids=["NEWS_GEOPOLITICAL_EVENTS"], |
| 654 | ) |
| 655 | |
| 656 | assert result.planned_requirement_ids == ("CURRENT_NEWS",) |
| 657 | assert executor.primary_calls == [{"CURRENT_NEWS"}] |
| 658 | assert "SHAREHOLDING" in source.missing |
| 659 | |
| 660 | |
| 661 | @pytest.mark.asyncio |
| 662 | async def test_primary_missing_uses_only_approved_existing_financial_fallback() -> None: |
| 663 | source = StateDataSource({"QUARTERLY_FINANCIALS"}) |
| 664 | executor = UpdatingExecutor(source, update_primary=False) |
| 665 | result = await _runtime(source, executor).ensure( |
| 666 | INSTRUMENT_ID, |
| 667 | jurisdiction="INDIA", |
| 668 | requirement_ids=["QUARTERLY_FINANCIALS"], |
| 669 | ) |
| 670 | |
| 671 | assert executor.primary_calls == [{"QUARTERLY_FINANCIALS"}] |
| 672 | assert executor.fallback_calls == [{"QUARTERLY_FINANCIALS"}] |
| 673 | assert result.executed_capabilities == ("RECORDED_PRIMARY", "RECORDED_FALLBACK") |
| 674 | assert result.readiness.for_requirement("QUARTERLY_FINANCIALS").status == ResearchRequirementStatus.READY_FRESH |
| 675 | |
| 676 | |
| 677 | @pytest.mark.asyncio |
| 678 | async def test_concurrent_same_instrument_ensure_reuses_single_flight() -> None: |
| 679 | source = StateDataSource({"CURRENT_NEWS"}) |
| 680 | executor = UpdatingExecutor(source) |
| 681 | executor.release = asyncio.Event() |
| 682 | runtime = _runtime(source, executor) |
| 683 | |
| 684 | first = asyncio.create_task(runtime.ensure( |
| 685 | INSTRUMENT_ID, jurisdiction="INDIA", requirement_ids=["CURRENT_NEWS"] |
| 686 | )) |
| 687 | await executor.started.wait() |
| 688 | second = asyncio.create_task(runtime.ensure( |
| 689 | INSTRUMENT_ID, jurisdiction="INDIA", requirement_ids=["CURRENT_NEWS"] |
| 690 | )) |
| 691 | await asyncio.sleep(0) |
| 692 | executor.release.set() |
| 693 | first_result, second_result = await asyncio.gather(first, second) |
| 694 | |
| 695 | assert executor.primary_calls == [{"CURRENT_NEWS"}] |
| 696 | assert first_result.reused_single_flight is False |
| 697 | assert second_result.reused_single_flight is True |
| 698 | |
| 699 | |
| 700 | class BudgetExecutor: |
| 701 | def __init__(self, source: StateDataSource) -> None: |
| 702 | self.source = source |
| 703 | self.primary_calls = 0 |
| 704 | self.fallback_calls = 0 |
| 705 | self.cancelled = False |
| 706 | |
| 707 | async def execute_primary(self, _instrument_id, targets, **_kwargs): |
| 708 | self.primary_calls += 1 |
| 709 | first = targets[0].requirement_id |
| 710 | progress = _kwargs.get("progress") |
| 711 | if progress is not None: |
| 712 | progress.executed(f"YAHOO_FINANCE_MCP:{first}") |
| 713 | for target in targets: |
| 714 | progress.failed(target.requirement_id, "EXTERNAL_RESULT_INCOMPLETE") |
| 715 | self.source.missing.discard(first) |
| 716 | try: |
| 717 | await asyncio.sleep(10) |
| 718 | except asyncio.CancelledError: |
| 719 | self.cancelled = True |
| 720 | raise |
| 721 | return CapabilityExecutionResult(("UNREACHABLE",), {}) |
| 722 | |
| 723 | async def execute_approved_fallbacks(self, _instrument_id, _targets): |
| 724 | self.fallback_calls += 1 |
| 725 | return CapabilityExecutionResult() |
| 726 | |
| 727 | |
| 728 | @pytest.mark.asyncio |
| 729 | async def test_ensure_budget_retains_committed_success_and_reports_unexecuted_targets() -> None: |
| 730 | source = StateDataSource({"CURRENT_NEWS", "SHAREHOLDING"}) |
| 731 | executor = BudgetExecutor(source) |
| 732 | runtime = ResearchReadinessRuntime( |
| 733 | RuntimeRepository(), source, executor, ensure_timeout_seconds=0.02 # type: ignore[arg-type] |
| 734 | ) |
| 735 | started = asyncio.get_running_loop().time() |
| 736 | |
| 737 | result = await runtime.ensure( |
| 738 | INSTRUMENT_ID, |
| 739 | jurisdiction="INDIA", |
| 740 | requirement_ids=["CURRENT_NEWS", "SHAREHOLDING"], |
| 741 | ) |
| 742 | |
| 743 | assert asyncio.get_running_loop().time() - started < 0.5 |
| 744 | assert executor.primary_calls == 1 |
| 745 | assert executor.fallback_calls == 0 |
| 746 | assert executor.cancelled is True |
| 747 | completed = result.planned_requirement_ids[0] |
| 748 | deferred = result.planned_requirement_ids[1] |
| 749 | assert result.readiness.for_requirement(completed).status == ResearchRequirementStatus.READY_FRESH |
| 750 | assert result.readiness.for_requirement(deferred).status == ResearchRequirementStatus.FAILED |
| 751 | assert result.executed_capabilities == (f"YAHOO_FINANCE_MCP:{completed}",) |
| 752 | assert result.failures == { |
| 753 | deferred: "EXTERNAL_RESULT_INCOMPLETE|ACQUISITION_TIMEOUT" |
| 754 | } |
| 755 | |
| 756 | |
| 757 | class EmptyFailureExecutor: |
| 758 | def __init__(self, reason: str) -> None: |
| 759 | self.reason = reason |
| 760 | self.primary_calls = 0 |
| 761 | self.fallback_calls = 0 |
| 762 | |
| 763 | async def execute_primary(self, _instrument_id, targets, **_kwargs): |
| 764 | self.primary_calls += 1 |
| 765 | return CapabilityExecutionResult( |
| 766 | ("YAHOO_FINANCE_MCP:CURRENT_NEWS",), |
| 767 | {target.requirement_id: self.reason for target in targets}, |
| 768 | ) |
| 769 | |
| 770 | async def execute_approved_fallbacks(self, _instrument_id, _targets): |
| 771 | self.fallback_calls += 1 |
| 772 | return CapabilityExecutionResult() |
| 773 | |
| 774 | |
| 775 | @pytest.mark.asyncio |
| 776 | async def test_individual_capability_failure_returns_partial_result_with_reason() -> None: |
| 777 | source = StateDataSource({"CURRENT_NEWS"}) |
| 778 | executor = EmptyFailureExecutor("EXTERNAL_CAPABILITY_UNSUPPORTED") |
| 779 | runtime = ResearchReadinessRuntime(RuntimeRepository(), source, executor) # type: ignore[arg-type] |
| 780 | |
| 781 | result = await runtime.ensure( |
| 782 | INSTRUMENT_ID, jurisdiction="INDIA", requirement_ids=["CURRENT_NEWS"] |
| 783 | ) |
| 784 | |
| 785 | assert result.failures == {"CURRENT_NEWS": "EXTERNAL_CAPABILITY_UNSUPPORTED"} |
| 786 | assert result.readiness.for_requirement("CURRENT_NEWS").status == ResearchRequirementStatus.FAILED |
| 787 | assert executor.primary_calls == executor.fallback_calls == 1 |
| 788 | |
| 789 | |
| 790 | class MetadataOnlyOrchestrator: |
| 791 | def __init__(self, profile: CompanyResearchProfile) -> None: |
| 792 | self.profile = profile |
| 793 | self.calls: list[str] = [] |
| 794 | self.portfolio_mutations = 0 |
| 795 | self.watchlist_mutations = 0 |
| 796 | |
| 797 | async def global_instrument_metadata(self, instrument_id, **_kwargs): |
| 798 | self.calls.append("GET_CANONICAL_IDENTITY") |
| 799 | return { |
| 800 | "globalInstrumentId": str(instrument_id), |
| 801 | "canonicalSector": "Industrials", |
| 802 | "updatedAt": NOW.isoformat(), |
| 803 | } |
| 804 | |
| 805 | def register_global_profile_metadata(self, instrument_id, _metadata): |
| 806 | return instrument_id == self.profile.instrument_id |
| 807 | |
| 808 | |
| 809 | def test_readiness_get_is_read_only_and_returns_source_freshness(monkeypatch) -> None: |
| 810 | profile = _profile() |
| 811 | repository = DurableRepositoryFixture(profile) |
| 812 | adapter = RepositoryResearchReadinessAdapter(repository) |
| 813 | executor = UpdatingExecutor(StateDataSource(set())) |
| 814 | runtime = ResearchReadinessRuntime(repository, adapter, executor) # type: ignore[arg-type] |
| 815 | orchestrator = MetadataOnlyOrchestrator(profile) |
| 816 | monkeypatch.setattr(main, "repository", repository) |
| 817 | monkeypatch.setattr(main, "portfolio_orchestrator", orchestrator) |
| 818 | monkeypatch.setattr(main, "research_readiness_adapter", adapter) |
| 819 | monkeypatch.setattr(main, "research_readiness_runtime", runtime) |
| 820 | |
| 821 | response = TestClient(main.app).get( |
| 822 | f"/api/v1/research/readiness/{profile.instrument_id}", |
| 823 | headers={ |
| 824 | "X-AIP-User-Id": "user", |
| 825 | "X-AIP-User-Issuer": "gateway", |
| 826 | "X-AIP-User-Subject": "subject", |
| 827 | }, |
| 828 | ) |
| 829 | |
| 830 | assert response.status_code == 200 |
| 831 | body = response.json() |
| 832 | assert body["globalInstrumentId"] == str(profile.instrument_id) |
| 833 | assert body["overallCompletenessPct"] > 0 |
| 834 | quarterly = next(item for item in body["requirements"] if item["requirementId"] == "QUARTERLY_FINANCIALS") |
| 835 | assert quarterly["sourceProvider"] == "NSE" |
| 836 | assert quarterly["sourceUrl"] == SOURCE_URL |
| 837 | assert quarterly["freshnessPolicy"]["mode"] == "RELEASE_AWARE_QUARTERLY" |
| 838 | assert executor.primary_calls == [] |
| 839 | assert orchestrator.calls == ["GET_CANONICAL_IDENTITY"] |
| 840 | assert orchestrator.portfolio_mutations == orchestrator.watchlist_mutations == 0 |
| 841 | |
| 842 | |
| 843 | def test_readiness_ensure_api_expands_one_area_and_executes_only_its_missing_requirement( |
| 844 | monkeypatch, |
| 845 | ) -> None: |
| 846 | profile = _profile() |
| 847 | repository = DurableRepositoryFixture(profile, complete=False) |
| 848 | identity_adapter = RepositoryResearchReadinessAdapter(repository) |
| 849 | source = StateDataSource({"CURRENT_NEWS", "SHAREHOLDING"}) |
| 850 | executor = UpdatingExecutor(source) |
| 851 | runtime = _runtime(source, executor) |
| 852 | orchestrator = MetadataOnlyOrchestrator(profile) |
| 853 | monkeypatch.setattr(main, "repository", repository) |
| 854 | monkeypatch.setattr(main, "portfolio_orchestrator", orchestrator) |
| 855 | monkeypatch.setattr(main, "research_readiness_adapter", identity_adapter) |
| 856 | monkeypatch.setattr(main, "research_readiness_runtime", runtime) |
| 857 | |
| 858 | response = TestClient(main.app).post( |
| 859 | f"/api/v1/research/readiness/{profile.instrument_id}/ensure", |
| 860 | headers={ |
| 861 | "X-AIP-User-Id": "user", |
| 862 | "X-AIP-User-Issuer": "gateway", |
| 863 | "X-AIP-User-Subject": "subject", |
| 864 | }, |
| 865 | json={"requirements": ["NEWS_GEOPOLITICAL_EVENTS"]}, |
| 866 | ) |
| 867 | |
| 868 | assert response.status_code == 200 |
| 869 | assert response.json()["refreshState"] == { |
| 870 | "plannedRequirements": ["CURRENT_NEWS"], |
| 871 | "executedCapabilities": ["RECORDED_PRIMARY"], |
| 872 | "reusedSingleFlight": False, |
| 873 | "failureReasons": {}, |
| 874 | } |
| 875 | assert executor.primary_calls == [{"CURRENT_NEWS"}] |
| 876 | assert "SHAREHOLDING" in source.missing |
| 877 | assert repository.portfolio_mutations == repository.watchlist_mutations == 0 |
| 878 | |
| 879 | |
| 880 | def test_readiness_ensure_api_rejects_unknown_requirement_without_provider_work( |
| 881 | monkeypatch, |
| 882 | ) -> None: |
| 883 | profile = _profile() |
| 884 | repository = DurableRepositoryFixture(profile, complete=False) |
| 885 | identity_adapter = RepositoryResearchReadinessAdapter(repository) |
| 886 | source = StateDataSource({"CURRENT_NEWS"}) |
| 887 | executor = UpdatingExecutor(source) |
| 888 | runtime = _runtime(source, executor) |
| 889 | monkeypatch.setattr(main, "repository", repository) |
| 890 | monkeypatch.setattr(main, "portfolio_orchestrator", MetadataOnlyOrchestrator(profile)) |
| 891 | monkeypatch.setattr(main, "research_readiness_adapter", identity_adapter) |
| 892 | monkeypatch.setattr(main, "research_readiness_runtime", runtime) |
| 893 | |
| 894 | response = TestClient(main.app).post( |
| 895 | f"/api/v1/research/readiness/{profile.instrument_id}/ensure", |
| 896 | headers={ |
| 897 | "X-AIP-User-Id": "user", |
| 898 | "X-AIP-User-Issuer": "gateway", |
| 899 | "X-AIP-User-Subject": "subject", |
| 900 | }, |
| 901 | json={"requirements": ["UNREGISTERED_PROVIDER_FACT"]}, |
| 902 | ) |
| 903 | |
| 904 | assert response.status_code == 400 |
| 905 | assert response.json()["detail"] == "UNKNOWN_RESEARCH_REQUIREMENT:UNREGISTERED_PROVIDER_FACT" |
| 906 | assert executor.primary_calls == executor.fallback_calls == [] |
| 907 | |
| 908 | |
| 909 | def test_readiness_response_is_deterministic_and_has_no_stock_score() -> None: |
| 910 | source = StateDataSource({"LATEST_PRICE"}) |
| 911 | result = ResearchReadinessService(source).assess( |
| 912 | INSTRUMENT_ID, jurisdiction="INDIA", now=NOW |
| 913 | ) |
| 914 | first = readiness_response(result, ResearchRequirementRegistry.default()) |
| 915 | second = readiness_response(result, ResearchRequirementRegistry.default()) |
| 916 | |
| 917 | assert first == second |
| 918 | assert first["criticalCompletenessPct"] < 100 |
| 919 | assert first["overallCompletenessPct"] < 100 |
| 920 | assert "score" not in first |
| 921 | assert first["confidence"] in {"LOW", "MEDIUM", "HIGH"} |
| 922 | |
| 923 | |
| 924 | def test_nse_quarterly_pdf_new_write_persists_metadata_and_facts_without_content(tmp_path) -> None: |
| 925 | persistence = SqliteResearchPersistence(tmp_path / "research.sqlite") |
| 926 | repository = ResearchRepository( |
| 927 | settings=Settings(research_demo_enabled=False), persistence=persistence |
| 928 | ) |
| 929 | profile = _profile(uuid4()) |
| 930 | repository.profiles.append(profile) |
| 931 | text = ( |
| 932 | "Readiness India Limited READY INE000A01010 Statement of Unaudited Financial " |
| 933 | "Results for the quarter ended 30 June 2026 Amounts in Rs. Crore Particulars " |
| 934 | "Quarter ended 30.06.2026 31.03.2026 30.06.2025 Revenue from operations " |
| 935 | "100.00 90.00 80.00 Profit for the period 12.00 10.00 8.00 Basic EPS 5.00 4.00 3.00" |
| 936 | ) |
| 937 | |
| 938 | document = repository.ingest_fixture( |
| 939 | original_url=SOURCE_URL, |
| 940 | source_type=SourceType.EXCHANGE_ANNOUNCEMENT, |
| 941 | source_name="NSE corporate announcements", |
| 942 | publisher="NSE", |
| 943 | content_type="application/pdf", |
| 944 | body=text, |
| 945 | reliability=ReliabilityLevel.LEVEL_A, |
| 946 | source_mode=SourceMode.REAL, |
| 947 | source_classification=SourceClassification.EXCHANGE, |
| 948 | discovery_provider="NSE_OFFICIAL_API", |
| 949 | expected_profile=profile, |
| 950 | document_status=DocumentStatus.PARSED, |
| 951 | _trusted_profile_identity=_TRUSTED_NSE_PROFILE_IDENTITY, |
| 952 | ) |
| 953 | |
| 954 | persisted = next( |
| 955 | item for item in persistence.load_documents() if item.document_id == document.document_id |
| 956 | ) |
| 957 | facts = repository.financial_facts_for(profile.instrument_id) |
| 958 | adapter = RepositoryResearchReadinessAdapter(repository) |
| 959 | readiness = ResearchReadinessService(adapter).assess( |
| 960 | profile.instrument_id, jurisdiction="INDIA", now=NOW |
| 961 | ) |
| 962 | |
| 963 | assert document.normalized_text |
| 964 | assert persisted.normalized_text is None |
| 965 | assert persisted.raw_text is None |
| 966 | assert persisted.canonical_url == SOURCE_URL |
| 967 | assert any(fact.key.metric == "revenue" and fact.key.period_type == "QUARTERLY" for fact in facts) |
| 968 | assert any(fact.value.source_url == SOURCE_URL for fact in facts) |
| 969 | assert readiness.for_requirement("QUARTERLY_FINANCIALS").source_url == SOURCE_URL |
| 970 | assert list(tmp_path.glob("*.pdf")) == [] |
| 971 | |
| 972 | |
| 973 | def test_nse_financial_result_classification_never_persists_unparsed_pdf_text(tmp_path) -> None: |
| 974 | persistence = SqliteResearchPersistence(tmp_path / "research.sqlite") |
| 975 | repository = ResearchRepository( |
| 976 | settings=Settings(research_demo_enabled=False), persistence=persistence |
| 977 | ) |
| 978 | profile = _profile(uuid4()) |
| 979 | repository.profiles.append(profile) |
| 980 | |
| 981 | document = repository.ingest_fixture( |
| 982 | original_url=SOURCE_URL, |
| 983 | source_type=SourceType.EXCHANGE_ANNOUNCEMENT, |
| 984 | source_name="NSE corporate announcements", |
| 985 | publisher="NSE", |
| 986 | content_type="application/pdf", |
| 987 | body=( |
| 988 | "Readiness India Limited READY INE000A01010 official quarterly financial " |
| 989 | "result attachment whose tabular facts could not be normalized safely." |
| 990 | ), |
| 991 | reliability=ReliabilityLevel.LEVEL_A, |
| 992 | source_mode=SourceMode.REAL, |
| 993 | source_classification=SourceClassification.EXCHANGE, |
| 994 | discovery_provider="NSE_OFFICIAL_API", |
| 995 | expected_profile=profile, |
| 996 | document_status=DocumentStatus.PARSED, |
| 997 | _trusted_profile_identity=_TRUSTED_NSE_PROFILE_IDENTITY, |
| 998 | _metadata_only_nse_financial_result=True, |
| 999 | ) |
| 1000 | |
| 1001 | persisted = next( |
| 1002 | item for item in persistence.load_documents() if item.document_id == document.document_id |
| 1003 | ) |
| 1004 | |
| 1005 | assert document.normalized_text |
| 1006 | assert persisted.normalized_text is None |
| 1007 | assert persisted.raw_text is None |
| 1008 | assert persisted.canonical_url == SOURCE_URL |
| 1009 | assert repository.financial_facts_for(profile.instrument_id) == [] |
| 1010 | assert list(tmp_path.glob("*.pdf")) == [] |
| 1011 | |
| 1012 | |
| 1013 | @pytest.mark.asyncio |
| 1014 | async def test_repository_targeted_boundary_forwards_only_selected_legacy_categories( |
| 1015 | tmp_path, |
| 1016 | ) -> None: |
| 1017 | persistence = SqliteResearchPersistence(tmp_path / "targeted.sqlite") |
| 1018 | repository = ResearchRepository( |
| 1019 | settings=Settings(research_live_enabled=True, research_demo_enabled=False), |
| 1020 | persistence=persistence, |
| 1021 | ) |
| 1022 | profile = _profile(uuid4()) |
| 1023 | repository.profiles.append(profile) |
| 1024 | calls: list[tuple[set[str], bool]] = [] |
| 1025 | |
| 1026 | async def record_targeted(_instrument_id, _pre_resolved, *, force, requested_categories): |
| 1027 | calls.append((set(requested_categories), force)) |
| 1028 | |
| 1029 | repository._refresh_live = record_targeted # type: ignore[method-assign] |
| 1030 | await repository.refresh_targeted_categories( |
| 1031 | profile.instrument_id, |
| 1032 | {"CATALYSTS", "RISKS"}, |
| 1033 | correlation_id="readiness-test", |
| 1034 | allow_demo=False, |
| 1035 | ) |
| 1036 | |
| 1037 | assert calls == [({"CATALYSTS", "RISKS"}, True)] |
| 1038 | |
| 1039 | |
| 1040 | @pytest.mark.asyncio |
| 1041 | async def test_repository_targeted_refresh_cancels_owned_provider_work_and_records_reason( |
| 1042 | tmp_path, |
| 1043 | ) -> None: |
| 1044 | persistence = SqliteResearchPersistence(tmp_path / "targeted-cancel.sqlite") |
| 1045 | repository = ResearchRepository( |
| 1046 | settings=Settings(research_live_enabled=True, research_demo_enabled=False), |
| 1047 | persistence=persistence, |
| 1048 | ) |
| 1049 | profile = _profile(uuid4()) |
| 1050 | repository.profiles.append(profile) |
| 1051 | started = asyncio.Event() |
| 1052 | cancelled = asyncio.Event() |
| 1053 | |
| 1054 | async def block_targeted(_instrument_id, _pre_resolved, *, force, requested_categories): |
| 1055 | assert force is True |
| 1056 | assert requested_categories == {"CATALYSTS"} |
| 1057 | started.set() |
| 1058 | try: |
| 1059 | await asyncio.sleep(10) |
| 1060 | except asyncio.CancelledError: |
| 1061 | cancelled.set() |
| 1062 | raise |
| 1063 | |
| 1064 | repository._refresh_live = block_targeted # type: ignore[method-assign] |
| 1065 | task = asyncio.create_task( |
| 1066 | repository.refresh_targeted_categories( |
| 1067 | profile.instrument_id, |
| 1068 | {"CATALYSTS"}, |
| 1069 | correlation_id="readiness-cancel-test", |
| 1070 | allow_demo=False, |
| 1071 | ) |
| 1072 | ) |
| 1073 | await started.wait() |
| 1074 | task.cancel() |
| 1075 | with pytest.raises(asyncio.CancelledError): |
| 1076 | await task |
| 1077 | await asyncio.sleep(0) |
| 1078 | |
| 1079 | assert cancelled.is_set() |
| 1080 | assert profile.instrument_id not in repository._instrument_refresh_flights |
| 1081 | row = persistence._connection.execute( |
| 1082 | """ |
| 1083 | SELECT status, safe_error_code |
| 1084 | FROM research_refresh_runs |
| 1085 | WHERE correlation_id = ? |
| 1086 | """, |
| 1087 | ("readiness-cancel-test",), |
| 1088 | ).fetchone() |
| 1089 | assert tuple(row) == ("FAILED", "ACQUISITION_CANCELLED") |
| 1090 | |
| 1091 | |
| 1092 | def test_jurisdiction_mapping_is_provider_neutral() -> None: |
| 1093 | assert jurisdiction_for_profile(_profile()) == "INDIA" |
| 1094 | european = _profile(uuid4()).model_copy(update={"country": "DE", "exchange": "XETR"}) |
| 1095 | american = _profile(uuid4()).model_copy(update={"country": "US", "exchange": "XNAS"}) |
| 1096 | assert jurisdiction_for_profile(european) == "EUROPE" |
| 1097 | assert jurisdiction_for_profile(american) == "USA" |