from unittest.mock import patch, MagicMock, call from ingest import _period_to_av_quarter def test_quarterly_period(): assert _period_to_av_quarter("Q42024") == "2024Q4" assert _period_to_av_quarter("Q12025") == "2025Q1" assert _period_to_av_quarter("Q22024") == "2024Q2" assert _period_to_av_quarter("Q32024") == "2024Q3" def test_annual_period_maps_to_q4(): assert _period_to_av_quarter("FY2024") == "2024Q4" def test_empty_period_returns_empty(): assert _period_to_av_quarter("") == "" def test_malformed_period_returns_empty(): assert _period_to_av_quarter("UNKNOWN") == "" def _make_edgar(period, form_type): """Build a minimal EdgarData-like object for ingest tests.""" from ingestion.edgar import EdgarData return EdgarData( ticker="NVDA", company_name="NVIDIA", filing_date="2025-02-26", period=period, form_type=form_type, revenue=None, revenue_yoy_pct=None, eps=None, gross_margin=None, operating_margin=None, free_cash_flow=None, mda_text="", risk_factors_text="", ) def test_10k_fetches_q4_transcript(): """ingest() must call fetch_transcript with the Q4 quarter code for 10-K filings.""" from ingest import ingest edgar_10k = _make_edgar("FY2025", "10-K") with ( patch("ingest.fetch_all_edgar_data", return_value=[edgar_10k]), patch("ingest.fill_missing_metrics", side_effect=lambda x: x), patch("ingest.clear_ticker_data"), patch("ingest._extract_guidance", return_value=None), patch("ingest.parse_guidance", return_value={}), patch("ingest.upsert_metrics"), patch("ingest.upsert_section"), patch("ingest.embed_and_store_filing"), patch("ingest.embed_and_store_transcript"), patch("ingest.prune_old_metrics"), patch("ingest.init_db"), patch("ingest.init_sections_db"), patch("ingest.get_all_metrics", return_value=[]), patch("ingest.get_section", return_value=None), patch("ingest.fetch_transcript", return_value="Q4 transcript text") as mock_transcript, ): ingest("NVDA") mock_transcript.assert_called_once_with("NVDA", "2025Q4") def test_10q_fetches_quarterly_transcript(): """ingest() must call fetch_transcript with the correct quarterly code for 10-Q filings.""" from ingest import ingest edgar_10q = _make_edgar("Q12025", "10-Q") with ( patch("ingest.fetch_all_edgar_data", return_value=[edgar_10q]), patch("ingest.fill_missing_metrics", side_effect=lambda x: x), patch("ingest.clear_ticker_data"), patch("ingest._extract_guidance", return_value=None), patch("ingest.parse_guidance", return_value={}), patch("ingest.upsert_metrics"), patch("ingest.upsert_section"), patch("ingest.embed_and_store_filing"), patch("ingest.embed_and_store_transcript"), patch("ingest.prune_old_metrics"), patch("ingest.init_db"), patch("ingest.init_sections_db"), patch("ingest.get_all_metrics", return_value=[]), patch("ingest.get_section", return_value=None), patch("ingest.fetch_transcript", return_value="Q1 transcript text") as mock_transcript, ): ingest("NVDA") mock_transcript.assert_called_once_with("NVDA", "2025Q1") def test_delta_backfills_missing_company_profile_sections_without_refetching_transcript(): """A previously ingested 10-K must be revisited when its profile sections are absent.""" from ingest import ingest edgar_10k = _make_edgar("FY2025", "10-K") edgar_10k.business_text = "NVIDIA designs accelerated computing platforms." edgar_10k.segments_geography_text = "Revenue is disclosed by reportable segment and geography." def stored_section(_ticker, _period, section): return "cached transcript" if section == "transcript" else None with ( patch("ingest.fetch_all_edgar_data", return_value=[edgar_10k]), patch("ingest.fill_missing_metrics", side_effect=lambda x: x), patch("ingest.get_all_metrics", return_value=[{"period": "FY2025"}]), patch("ingest.get_section", side_effect=stored_section), patch("ingest._extract_guidance", return_value=None), patch("ingest.parse_guidance", return_value={}), patch("ingest.upsert_metrics"), patch("ingest.upsert_section") as mock_upsert_section, patch("ingest.embed_and_store_filing") as mock_embed_filing, patch("ingest.embed_and_store_transcript"), patch("ingest.prune_old_metrics"), patch("ingest.init_db"), patch("ingest.init_sections_db"), patch("ingest.fetch_transcript") as mock_fetch_transcript, ): ingest("NVDA") mock_fetch_transcript.assert_not_called() mock_embed_filing.assert_called_once() assert mock_embed_filing.call_args.kwargs["business_text"] == edgar_10k.business_text assert ( mock_embed_filing.call_args.kwargs["segments_geography_text"] == edgar_10k.segments_geography_text ) assert call( "NVDA", "FY2025", "10-K", "business", edgar_10k.business_text, ) in mock_upsert_section.call_args_list assert call( "NVDA", "FY2025", "10-K", "segments_geography", edgar_10k.segments_geography_text, ) in mock_upsert_section.call_args_list