| 1 | from dataclasses import replace |
| 2 | from datetime import datetime, time, timedelta, timezone |
| 3 | |
| 4 | import pytest |
| 5 | |
| 6 | from app.research_applicability import RequirementApplicability, classify_requirements |
| 7 | from app.research_readiness import ResearchRequirementStatus, ResearchRequirementRegistry |
| 8 | from app.research_readiness_runtime import RepositoryResearchReadinessAdapter |
| 9 | from app.market_sessions import MarketTradingSchedule, MarketCalendarException, latest_completed_session, next_session_open |
| 10 | from app.structured_market import _normalize_yfinance_news |
| 11 | from test_research_readiness import complete_snapshot, assess |
| 12 | from test_research_readiness_runtime import _profile, _fact |
| 13 | |
| 14 | |
| 15 | @pytest.mark.parametrize("industry,expected", [ |
| 16 | ("Banks - Regional", "NOT_APPLICABLE"), |
| 17 | ("Financial Data & Stock Exchanges", "NOT_APPLICABLE"), |
| 18 | ("Electrical Equipment & Parts", "APPLICABLE"), |
| 19 | ("Engineering & Construction", "APPLICABLE"), |
| 20 | ("Aerospace & Defense", "APPLICABLE"), |
| 21 | ("Utilities - Regulated Electric", "APPLICABLE"), |
| 22 | ("Software - Application", "PARTIALLY_APPLICABLE"), |
| 23 | (None, "UNKNOWN"), |
| 24 | ]) |
| 25 | def test_business_classification_only_controls_applicability(industry, expected): |
| 26 | decision = classify_requirements(None, industry, "canonical-reference")["ORDER_BOOK_CAPEX_GUIDANCE"] |
| 27 | assert decision.state == expected |
| 28 | |
| 29 | |
| 30 | def test_not_applicable_is_excluded_from_both_denominators_without_fake_coverage(): |
| 31 | # News is optional; use a mandatory input to exercise both denominators. |
| 32 | snapshot = complete_snapshot(omit={"GROWTH_FACTS", "ORDER_BOOK_CAPEX_GUIDANCE"}) |
| 33 | decisions = {key: RequirementApplicability("NOT_APPLICABLE", "DOMAIN_TEST") |
| 34 | for key in ("GROWTH_FACTS", "ORDER_BOOK_CAPEX_GUIDANCE")} |
| 35 | _, baseline, _ = assess(snapshot) |
| 36 | _, result, plan = assess(replace(snapshot, applicability_by_requirement=decisions)) |
| 37 | assert baseline.overall_completeness_pct < 100 |
| 38 | assert baseline.critical_completeness_pct < 100 |
| 39 | assert result.overall_completeness_pct == 100 |
| 40 | assert result.critical_completeness_pct == 100 |
| 41 | for key in decisions: |
| 42 | row = result.for_requirement(key) |
| 43 | assert row.status == ResearchRequirementStatus.NOT_APPLICABLE |
| 44 | assert row.evidence_ids == () |
| 45 | assert row.covered_input_ids == () |
| 46 | assert row.supported_actions == () |
| 47 | assert key not in {target.requirement_id for target in plan.targets} |
| 48 | |
| 49 | |
| 50 | @pytest.mark.parametrize("failure", ["EXTERNAL_CAPABILITY_UNSUPPORTED", "ACQUISITION_TIMEOUT", "NO_DATA"]) |
| 51 | def test_provider_failure_cannot_establish_non_applicability(failure): |
| 52 | snapshot = complete_snapshot(omit={"ORDER_BOOK_CAPEX_GUIDANCE"}, failures={"ORDER_BOOK_CAPEX_GUIDANCE": failure}) |
| 53 | _, result, _ = assess(snapshot) |
| 54 | assert result.for_requirement("ORDER_BOOK_CAPEX_GUIDANCE").status == ResearchRequirementStatus.FAILED |
| 55 | assert result.overall_completeness_pct < 100 |
| 56 | |
| 57 | |
| 58 | def test_quarterly_comparison_requires_same_basis_revenue_and_pat_not_duplicate_date_strings(): |
| 59 | profile = _profile() |
| 60 | values = {item.requirement_id: [] for item in ResearchRequirementRegistry.default().requirements} |
| 61 | facts = [_fact(profile, "revenue", "100", "2026-06-30", "QUARTERLY"), |
| 62 | _fact(profile, "pat", "10", "2026-06-30T00:00:00", "QUARTERLY")] |
| 63 | RepositoryResearchReadinessAdapter._append_financial_evidence(values, facts) |
| 64 | assert not any("COMPARABLE_QUARTERS" in item.covered_input_ids for item in values["QUARTERLY_FINANCIALS"]) |
| 65 | facts += [_fact(profile, "revenue", "90", "2026-03-31", "QUARTERLY"), |
| 66 | _fact(profile, "pat", "9", "2026-03-31", "QUARTERLY")] |
| 67 | RepositoryResearchReadinessAdapter._append_financial_evidence(values, facts) |
| 68 | assert any("COMPARABLE_QUARTERS" in item.covered_input_ids for item in values["QUARTERLY_FINANCIALS"]) |
| 69 | assert any("PROFITABILITY_HISTORY" in item.covered_input_ids for item in values["BUSINESS_QUALITY_FACTS"]) |
| 70 | |
| 71 | |
| 72 | def test_session_validity_uses_persisted_weekend_and_holiday_calendar(): |
| 73 | schedules = [MarketTradingSchedule("NSE", "XNSE", "IN", "Asia/Kolkata", day, time(9,15), time(15,30)) for day in range(5)] |
| 74 | friday = datetime(2026,9,11,10,0,tzinfo=timezone.utc) |
| 75 | assert latest_completed_session("XNSE", schedules, [], friday) == friday |
| 76 | assert next_session_open("XNSE", schedules, [], friday) == datetime(2026,9,14,3,45,tzinfo=timezone.utc) |
| 77 | holiday = MarketCalendarException("NSE", datetime(2026,9,14).date(), "CLOSED") |
| 78 | assert next_session_open("XNSE", schedules, [holiday], friday) == datetime(2026,9,15,3,45,tzinfo=timezone.utc) |
| 79 | assert next_session_open("UNKNOWN", schedules, [], friday) is None |
| 80 | |
| 81 | |
| 82 | def test_yahoo_iso_news_timestamp_is_preserved_not_replaced_with_retrieval_time(): |
| 83 | retrieved = datetime(2026,9,12,12,tzinfo=timezone.utc) |
| 84 | result = _normalize_yfinance_news([{"content": {"title":"Reported event", "canonicalUrl":{"url":"https://example.test/event"}, "pubDate":"2026-09-11T10:00:00Z"}}], retrieved) |
| 85 | assert result[0]["publishedAt"] == datetime(2026,9,11,10,tzinfo=timezone.utc) |
| 86 | assert result[0]["publishedAt"] != retrieved |
| 87 | |
| 88 | @pytest.mark.parametrize("value", [float("nan"), float("inf"), "NaN", "-Infinity"]) |
| 89 | def test_absent_or_nonfinite_statement_cells_are_not_financial_facts(value): |
| 90 | from app.structured_market import _decimal |
| 91 | assert _decimal(value) is None |