"""Tests for the SEC EDGAR filings adapter. Validates request building, response parsing, and error handling. """ from services.adapters.filings_adapter import FilingsDataAdapter, SECEdgarAdapter # --- Fake EDGAR EFTS responses --- EFTS_RESPONSE = { "hits": { "total": {"value": 3, "relation": "eq"}, "hits": [ { "_id": "0001234567-26-000001", "_source": { "adsh": "0001234567-26-000001", "ciks": ["0000320193"], "file_date": "2026-04-01", "form": "8-K", "form_type": "8-K", "display_names": ["Apple Inc. (CIK 0000320193)"], "entity_name": "Apple Inc.", "file_num": "001-36743", "file_type": "HTML", "file_description": "Current Report", "period_of_report": "2026-03-31", }, }, { "_id": "0001234567-26-000002", "_source": { "adsh": "0001234567-26-000002", "ciks": ["0000320193"], "file_date": "2026-03-15", "form": "10-Q", "form_type": "10-Q", "display_names": ["Apple Inc. (CIK 0000320193)"], "entity_name": "Apple Inc.", "file_num": "001-36743", "file_type": "HTML", "file_description": "Quarterly Report", "period_of_report": "2026-03-15", }, }, { "_id": "0001234567-26-000003", "_source": { "adsh": "0001234567-26-000003", "ciks": ["0000320193"], "file_date": "2026-01-30", "form": "10-K", "form_type": "10-K", "display_names": ["Apple Inc. (CIK 0000320193)"], "entity_name": "Apple Inc.", "file_num": "001-36743", "file_type": "HTML", "file_description": "Annual Report", "period_of_report": "2025-12-31", }, }, ], } } EMPTY_EFTS_RESPONSE = { "hits": { "total": {"value": 0, "relation": "eq"}, "hits": [], } } class TestSECEdgarSourceType: def test_source_type(self): adapter = SECEdgarAdapter() assert adapter.source_type() == "filings_api" def test_inherits_filings_data_adapter(self): assert issubclass(SECEdgarAdapter, FilingsDataAdapter) def test_bucket_name(self): adapter = SECEdgarAdapter() assert adapter.bucket_name() == "stonks-raw-filings" class TestSECEdgarBuildRequest: def setup_method(self): self.adapter = SECEdgarAdapter( base_url="https://efts.sec.gov", user_agent="TestAgent/1.0", ) def test_default_request(self): url, params, headers = self.adapter._build_request("AAPL", {}) assert url == "https://efts.sec.gov/LATEST/search-index" assert params["q"] == '"AAPL"' assert params["forms"] == "8-K,10-Q,10-K" assert headers["User-Agent"] == "TestAgent/1.0" def test_custom_forms(self): _, params, _ = self.adapter._build_request("AAPL", {"forms": "8-K"}) assert params["forms"] == "8-K" def test_date_range(self): config = {"start_date": "2026-01-01", "end_date": "2026-04-10"} _, params, _ = self.adapter._build_request("AAPL", config) assert params["dateRange"] == "custom" assert params["startdt"] == "2026-01-01" assert params["enddt"] == "2026-04-10" def test_cik_filter(self): _, params, _ = self.adapter._build_request("AAPL", {"cik": "0000320193"}) assert "cik:0000320193" in params["q"] assert '"AAPL"' in params["q"] def test_custom_query_override(self): _, params, _ = self.adapter._build_request("AAPL", {"query": "apple AND revenue"}) assert params["q"] == "apple AND revenue" def test_trailing_slash_stripped(self): adapter = SECEdgarAdapter(base_url="https://efts.sec.gov/") url, _, _ = adapter._build_request("AAPL", {}) assert "//LATEST" not in url def test_no_date_params_when_absent(self): _, params, _ = self.adapter._build_request("AAPL", {}) assert "dateRange" not in params assert "startdt" not in params assert "enddt" not in params class TestSECEdgarExtractItems: def setup_method(self): self.adapter = SECEdgarAdapter() def test_extract_filings(self): items = self.adapter._extract_items(EFTS_RESPONSE) assert len(items) == 3 assert items[0]["adsh"] == "0001234567-26-000001" assert items[0]["form"] == "8-K" def test_extract_empty_results(self): items = self.adapter._extract_items(EMPTY_EFTS_RESPONSE) assert items == [] def test_extract_missing_hits_key(self): items = self.adapter._extract_items({"status": "OK"}) assert items == [] def test_extract_non_dict_hits(self): items = self.adapter._extract_items({"hits": "unexpected"}) assert items == [] def test_extract_non_list_inner_hits(self): items = self.adapter._extract_items({"hits": {"hits": "bad"}}) assert items == [] class TestSECEdgarTotalHits: def setup_method(self): self.adapter = SECEdgarAdapter() def test_total_hits_dict(self): assert self.adapter._total_hits(EFTS_RESPONSE) == 3 def test_total_hits_int(self): data = {"hits": {"total": 5, "hits": []}} assert self.adapter._total_hits(data) == 5 def test_total_hits_missing(self): assert self.adapter._total_hits({}) == 0 def test_total_hits_non_dict_hits(self): assert self.adapter._total_hits({"hits": "bad"}) == 0 class TestSECEdgarErrorResult: def test_error_result_fields(self): adapter = SECEdgarAdapter() result = adapter._error_result("AAPL", "rate limited", 150.0, http_status=429, raw=b"slow down") assert not result.ok assert result.error == "rate limited" assert result.http_status == 429 assert result.response_time_ms == 150.0 assert result.raw_payload == b"slow down" assert result.metadata["provider"] == "sec_edgar" assert result.source_type == "filings_api" def test_error_result_defaults(self): adapter = SECEdgarAdapter() result = adapter._error_result("MSFT", "timeout", 200.0) assert result.http_status is None assert result.raw_payload == b"" assert result.ticker == "MSFT"