diff --git a/openfoia/cli.py b/openfoia/cli.py index 49e79b9..030111f 100644 --- a/openfoia/cli.py +++ b/openfoia/cli.py @@ -4538,6 +4538,10 @@ def _progress(event: str, msg: str) -> None: rprint(f" Sources used: {', '.join(report.sources_used)}") rprint(f" Total hits: {report.total_hits}") rprint(f" Entities flagged: {report.total_flagged}") + source_errors = getattr(report, "source_errors", {}) + if source_errors: + details = ", ".join(f"{source} ({error})" for source, error in source_errors.items()) + rprint(f" [red]Sources errored: {len(source_errors)} — {details}[/red]") rprint("=" * 60) for result in report.results: @@ -4567,7 +4571,13 @@ def _progress(event: str, msg: str) -> None: rprint(f" {hit.url}") if not report.total_flagged: - rprint("\n [green]No cross-reference hits found.[/green]") + if source_errors: + rprint( + "\n [yellow]No cross-reference hits found among completed source checks; " + "report is incomplete.[/yellow]" + ) + else: + rprint("\n [green]No cross-reference hits found.[/green]") # Export as FollowTheMoney if ftm: @@ -4583,10 +4593,12 @@ def _progress(event: str, msg: str) -> None: "total_hits": report.total_hits, "total_flagged": report.total_flagged, "sources": report.sources_used, + "source_errors": source_errors, "results": [ { "entity": r.entity_name, "type": r.entity_type, + "source_statuses": getattr(r, "source_statuses", {}), "hits": [ { "source": h.source, @@ -4599,6 +4611,7 @@ def _progress(event: str, msg: str) -> None: } for r in report.results if r.hits + or any(status != "checked" for status in getattr(r, "source_statuses", {}).values()) ], } output.write_text(json.dumps(report_data, indent=2)) diff --git a/openfoia/crossref.py b/openfoia/crossref.py index d2f29f0..077cbe5 100644 --- a/openfoia/crossref.py +++ b/openfoia/crossref.py @@ -44,6 +44,7 @@ class CrossRefResult: entity_type: str hits: list[CrossRefHit] sources_checked: list[str] + source_statuses: dict[str, str] = field(default_factory=dict) @property def flagged(self) -> bool: @@ -59,6 +60,7 @@ class CrossRefReport: total_hits: int total_flagged: int sources_used: list[str] + source_errors: dict[str, str] = field(default_factory=dict) # Entity types worth cross-referencing (skip dates, money, etc.) @@ -92,11 +94,21 @@ class _RateLimited(BaseException): """ +class _SourceCheckError(RuntimeError): + """A checker failure that must be visible in the cross-reference report.""" + + def __init__(self, error: BaseException): + self.error_type = type(error).__name__ + super().__init__(self.error_type) + + def _check_rate_limit(result: Any) -> None: """Raise _RateLimited if the search result indicates a rate limit error.""" err = getattr(result, "error", None) - if err and ("rate limit" in err.lower() or "429" in err.lower()): - raise _RateLimited(err) + if err: + if "rate limit" in err.lower() or "429" in err.lower(): + raise _RateLimited(err) + raise _SourceCheckError(RuntimeError(err)) def _deduplicate_entities(entities: list[Any]) -> list[Any]: @@ -194,12 +206,14 @@ async def crossref_entities( ) results: list[CrossRefResult] = [] + source_errors: dict[str, str] = {} # Track sources that hit rate limits — skip them for remaining entities exhausted_sources: set[str] = set() for idx, entity in enumerate(targets): hits: list[CrossRefHit] = [] sources_checked: list[str] = [] + source_statuses: dict[str, str] = {} if on_progress: on_progress( @@ -209,24 +223,38 @@ async def crossref_entities( for source_name, checker in available_sources.items(): if source_name in exhausted_sources: + source_statuses[source_name] = "skipped(rate-limited)" continue sources_checked.append(source_name) try: source_hits = await checker(entity.normalized_text, entity.entity_type) hits.extend(source_hits) + source_statuses[source_name] = "matched" if source_hits else "checked" except _RateLimited: logger.warning( "CrossRef %s rate limited — skipping for remaining entities", source_name, ) exhausted_sources.add(source_name) + source_statuses[source_name] = "ERRORED(RateLimited)" + source_errors.setdefault(source_name, "RateLimited") + except _SourceCheckError as exc: + logger.warning( + "CrossRef %s failed: %s", + source_name, + exc.error_type, + ) + source_statuses[source_name] = f"ERRORED({exc.error_type})" + source_errors.setdefault(source_name, exc.error_type) except Exception as e: logger.warning( - "CrossRef %s failed for '%s': %s", + "CrossRef %s failed: %s", source_name, - entity.normalized_text, - e, + type(e).__name__, ) + error_type = type(e).__name__ + source_statuses[source_name] = f"ERRORED({error_type})" + source_errors.setdefault(source_name, error_type) # Rate limit: respect each API's documented limits. Sleep the base # delay times a randomized ~[1.0, 1.5) jitter factor rather than @@ -246,6 +274,7 @@ async def crossref_entities( entity_type=entity.entity_type.value, hits=hits, sources_checked=sources_checked, + source_statuses=source_statuses, ) ) @@ -258,6 +287,7 @@ async def crossref_entities( total_hits=total_hits, total_flagged=len(flagged), sources_used=list(available_sources.keys()), + source_errors=source_errors, ) @@ -327,8 +357,12 @@ async def _check_muckrock( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for req in result.entities: @@ -374,8 +408,12 @@ async def _check_opencorporates( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for ent in result.entities: @@ -417,8 +455,12 @@ async def _check_sec( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] seen_ciks: set[str] = set() @@ -455,9 +497,11 @@ async def _check_icij(name: str, entity_type: EntityType, data_dir: str) -> list data_path = Path(data_dir) name_lower = name.lower() - # Search across all ICIJ CSV files - for csv_file in data_path.glob("*.csv"): - try: + # Search across all ICIJ CSV files. A local data read failure makes this + # source incomplete, so surface it to the report rather than treating it + # as an uneventful no-match. + try: + for csv_file in data_path.glob("*.csv"): with open(csv_file, encoding="utf-8", errors="ignore") as f: reader = csv.DictReader(f) for row in reader: @@ -480,8 +524,8 @@ async def _check_icij(name: str, entity_type: EntityType, data_dir: str) -> list ) ) break # one hit per row is enough - except Exception as e: - logger.warning("Failed to search ICIJ file %s: %s", csv_file, e) + except Exception as exc: + raise _SourceCheckError(exc) from exc return hits[:10] # cap to avoid flooding @@ -496,8 +540,12 @@ async def _check_fec( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for contrib in result.entities: @@ -529,8 +577,12 @@ async def _check_regulations( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for doc in result.entities: @@ -564,8 +616,12 @@ async def _check_govinfo( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for doc in result.entities: @@ -604,8 +660,12 @@ async def _check_nonprofits( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for org in result.entities: @@ -646,8 +706,12 @@ async def _check_usaspending( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for award in result.entities: @@ -687,8 +751,12 @@ async def _check_documentcloud( try: result = await adapter.search(name, page_size=5) _check_rate_limit(result) - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc hits = [] for doc in result.entities: @@ -736,11 +804,17 @@ async def _check_opensanctions( params={"q": name, "limit": 5}, headers={"Accept": "application/json"}, ) + if resp.status_code == 429: + raise _RateLimited("OpenSanctions rate limit") if resp.status_code != 200: - return [] + raise _SourceCheckError(RuntimeError(f"HTTP {resp.status_code}")) data = resp.json() - except Exception: - return [] + except _RateLimited: + raise + except _SourceCheckError: + raise + except Exception as exc: + raise _SourceCheckError(exc) from exc for result in data.get("results", []): score = result.get("score", 0) diff --git a/tests/test_crossref_status.py b/tests/test_crossref_status.py new file mode 100644 index 0000000..f7982cb --- /dev/null +++ b/tests/test_crossref_status.py @@ -0,0 +1,109 @@ +"""Regression tests for honest cross-reference source statuses.""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import pytest + +from openfoia import crossref as crossref_mod +from openfoia.crossref import CrossRefHit, _check_icij, crossref_entities +from openfoia.models import EntityType + + +def test_crossref_reports_a_failed_source_without_hiding_successful_hits(monkeypatch, caplog): + """A broken source must not look like a clean no-match result.""" + + async def failing_checker(name, entity_type): + raise RuntimeError("upstream response included the query") + + async def matching_checker(name, entity_type): + return [ + CrossRefHit( + source="working", + entity_name=name, + match_type="exact", + details="match", + ) + ] + + async def no_sleep(duration): + return None + + monkeypatch.setattr( + crossref_mod, + "_get_available_sources", + lambda icij_data_dir, egress: {"broken": failing_checker, "working": matching_checker}, + ) + monkeypatch.setattr(asyncio, "sleep", no_sleep) + + entity = SimpleNamespace( + entity_type=EntityType.ORGANIZATION, + normalized_text="Acme Corp", + confidence=1.0, + ) + report = asyncio.run(crossref_entities([entity], allow_network=True)) + + assert report.total_hits == 1 + assert report.source_errors == {"broken": "RuntimeError"} + assert report.results[0].source_statuses == { + "broken": "ERRORED(RuntimeError)", + "working": "matched", + } + assert "Acme Corp" not in caplog.text + assert "upstream response included the query" not in caplog.text + + +def test_crossref_reports_rate_limit_as_an_incomplete_source(monkeypatch): + """Rate-limited searches must not be represented as clean no-matches.""" + + async def rate_limited_checker(name, entity_type): + raise crossref_mod._RateLimited("source limit") + + async def working_checker(name, entity_type): + return [] + + async def no_sleep(duration): + return None + + monkeypatch.setattr( + crossref_mod, + "_get_available_sources", + lambda icij_data_dir, egress: {"limited": rate_limited_checker, "working": working_checker}, + ) + monkeypatch.setattr(asyncio, "sleep", no_sleep) + entities = [ + SimpleNamespace( + entity_type=EntityType.ORGANIZATION, + normalized_text=name, + confidence=1.0, + ) + for name in ("Acme Corp", "Globex Corp") + ] + + report = asyncio.run(crossref_entities(entities, allow_network=True)) + + assert report.source_errors == {"limited": "RateLimited"} + assert report.results[0].source_statuses["limited"] == "ERRORED(RateLimited)" + assert report.results[1].source_statuses["limited"] == "skipped(rate-limited)" + + +def test_icij_read_failure_propagates_as_a_redacted_source_error(monkeypatch, tmp_path): + """Unreadable local ICIJ data must reach the report error boundary.""" + + csv_file = tmp_path / "offshore.csv" + csv_file.write_text("name\nAcme Corp\n") + original_open = open + + def fail_open(path, *args, **kwargs): + if path == csv_file: + raise OSError("sensitive local path details") + return original_open(path, *args, **kwargs) + + monkeypatch.setattr("builtins.open", fail_open) + + with pytest.raises(crossref_mod._SourceCheckError) as error: + asyncio.run(_check_icij("Acme Corp", EntityType.ORGANIZATION, str(tmp_path))) + + assert error.value.error_type == "OSError" diff --git a/tests/test_security_egress_cli.py b/tests/test_security_egress_cli.py index 341f50d..ec2f725 100644 --- a/tests/test_security_egress_cli.py +++ b/tests/test_security_egress_cli.py @@ -18,6 +18,7 @@ from __future__ import annotations +import json from contextlib import contextmanager from pathlib import Path from types import SimpleNamespace @@ -358,6 +359,46 @@ async def _crossref_entities(entities, **kwargs): assert captured["egress"].mode is EgressMode.DIRECT +def test_crossref_reports_and_exports_source_errors(monkeypatch, fake_entities_db, tmp_path): + """A failed source must not be represented as an uneventful no-match.""" + + async def _crossref_entities(entities, **kwargs): + return SimpleNamespace( + total_entities=1, + sources_used=["sec", "muckrock"], + total_hits=0, + total_flagged=0, + source_errors={"sec": "RateLimited"}, + results=[ + SimpleNamespace( + entity_name="Jane Doe", + entity_type="PERSON", + source_statuses={"sec": "ERRORED(RateLimited)", "muckrock": "checked"}, + hits=[], + ), + SimpleNamespace( + entity_name="John Smith", + entity_type="PERSON", + source_statuses={"sec": "skipped(rate-limited)", "muckrock": "checked"}, + hits=[], + ), + ], + ) + + monkeypatch.setattr("openfoia.crossref.crossref_entities", _crossref_entities) + output = tmp_path / "crossref-report.json" + + result = runner.invoke(app, ["crossref", "--no-tor", "--yes", "--output", str(output)]) + + assert result.exit_code == 0, result.output + assert "Sources errored: 1" in result.output + assert "incomplete." in result.output + saved = json.loads(output.read_text()) + assert saved["source_errors"] == {"sec": "RateLimited"} + assert saved["results"][0]["source_statuses"]["sec"] == "ERRORED(RateLimited)" + assert saved["results"][1]["source_statuses"]["sec"] == "skipped(rate-limited)" + + def test_crossref_tor_mode_passes_egress_policy_through(monkeypatch, fake_entities_db): async def _reachable(*a, **k): return True