File size: 5,443 Bytes
097a290 35676b4 097a290 d1e793b 097a290 d1e793b 097a290 d1e793b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | 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
|