Spaces:
Running
Running
| """ | |
| Unit tests for the business news system (ADR-0015): dedup hashing, relevance | |
| windows, business-type validation, and classification completeness logic. | |
| classify_item goes through real Groq in production; here it's exercised only | |
| in E2E_FIXTURE_MODE against canned fixtures, same pattern as | |
| test_groq_fixture_mode.py — no network calls. | |
| """ | |
| import importlib | |
| from datetime import date, datetime, timezone | |
| import pytest | |
| from news.dedup import url_hash | |
| from news.windows import compute_expiry, CATEGORY_WINDOW_DAYS, DEFAULT_WINDOW_DAYS | |
| from news.business_types import valid_subset, VALID_IDS | |
| from news.classify import ClassificationResult | |
| from news.pipeline import _route | |
| # ─── dedup ────────────────────────────────────────────────────────────────── | |
| def test_url_hash_ignores_trailing_slash(): | |
| assert url_hash("https://example.com/a") == url_hash("https://example.com/a/") | |
| def test_url_hash_ignores_query_string(): | |
| assert url_hash("https://example.com/a") == url_hash("https://example.com/a?utm_source=x") | |
| def test_url_hash_ignores_fragment(): | |
| assert url_hash("https://example.com/a") == url_hash("https://example.com/a#section") | |
| def test_url_hash_distinguishes_different_urls(): | |
| assert url_hash("https://example.com/a") != url_hash("https://example.com/b") | |
| # ─── windows ──────────────────────────────────────────────────────────────── | |
| def test_compute_expiry_uses_category_window(): | |
| scraped_at = datetime(2026, 1, 1, tzinfo=timezone.utc) | |
| expiry = compute_expiry("weather", scraped_at, None) | |
| assert expiry == scraped_at + __import__("datetime").timedelta(days=CATEGORY_WINDOW_DAYS["weather"]) | |
| def test_compute_expiry_falls_back_to_default_for_unknown_category(): | |
| scraped_at = datetime(2026, 1, 1, tzinfo=timezone.utc) | |
| expiry = compute_expiry(None, scraped_at, None) | |
| assert expiry == scraped_at + __import__("datetime").timedelta(days=DEFAULT_WINDOW_DAYS) | |
| def test_compute_expiry_deadline_overrides_category_window(): | |
| scraped_at = datetime(2026, 1, 1, tzinfo=timezone.utc) | |
| deadline = date(2026, 12, 1) | |
| expiry = compute_expiry("funding", scraped_at, deadline) | |
| assert expiry == datetime(2026, 12, 1, tzinfo=timezone.utc) | |
| # ─── business_types ─────────────────────────────────────────────────────── | |
| def test_valid_subset_drops_hallucinated_ids(): | |
| real_id = next(iter(VALID_IDS)) | |
| assert valid_subset([real_id, "not-a-real-subsector"]) == [real_id] | |
| def test_valid_subset_empty_input(): | |
| assert valid_subset([]) == [] | |
| # ─── classify (fixture mode) ─────────────────────────────────────────────── | |
| def classify_module(monkeypatch): | |
| monkeypatch.setenv("E2E_FIXTURE_MODE", "1") | |
| import utils.groq_client as gc | |
| importlib.reload(gc) | |
| import news.classify as classify | |
| importlib.reload(classify) | |
| yield classify | |
| monkeypatch.delenv("E2E_FIXTURE_MODE", raising=False) | |
| importlib.reload(gc) | |
| importlib.reload(classify) | |
| def test_classify_item_complete_result_is_complete(classify_module): | |
| result = classify_module.classify_item("Lagos flood warning", "text", fixture="news_classify_complete") | |
| assert result.is_complete | |
| assert result.review_reason() is None | |
| assert result.impact == "neutral" | |
| assert result.category == "weather" | |
| def test_classify_item_incomplete_result_flags_missing_fields(classify_module): | |
| result = classify_module.classify_item("Unclear item", "text", fixture="news_classify_incomplete") | |
| assert not result.is_complete | |
| reason = result.review_reason() | |
| assert "category" in reason | |
| assert "business_types" in reason | |
| assert "location" in reason | |
| def test_classify_item_opportunity_keeps_deadline(classify_module): | |
| result = classify_module.classify_item("State SME grant", "text", fixture="news_classify_opportunity") | |
| assert result.impact == "opportunity" | |
| assert result.deadline == date(2026, 12, 1) | |
| def test_classify_item_deadline_dropped_when_not_opportunity(classify_module): | |
| result = classify_module.classify_item("Lagos flood warning", "text", fixture="news_classify_complete") | |
| assert result.deadline is None | |
| def test_classification_result_drops_invalid_enum_values(): | |
| result = __import__("news.classify", fromlist=["ClassificationResult"]).ClassificationResult({ | |
| "impact": "not-a-real-impact", | |
| "category": "not-a-real-category", | |
| "business_types": [], | |
| "location": "Lagos", | |
| }) | |
| assert result.impact is None | |
| assert result.category is None | |
| def test_classification_result_malformed_deadline_is_ignored(): | |
| from news.classify import ClassificationResult | |
| result = ClassificationResult({ | |
| "impact": "opportunity", | |
| "category": "funding", | |
| "business_types": [], | |
| "location": "Lagos", | |
| "deadline": "not-a-date", | |
| }) | |
| assert result.deadline is None | |
| # ─── confidence routing ───────────────────────────────────────────────────── | |
| _SOME_VALID_ID = next(iter(VALID_IDS)) | |
| def _result(impact="neutral", category="weather", business_types=None, location="Lagos", confidence=None): | |
| business_types = [_SOME_VALID_ID] if business_types is None else business_types | |
| return ClassificationResult({ | |
| "impact": impact, "category": category, | |
| "business_types": business_types, "location": location, | |
| "confidence": confidence, | |
| }) | |
| def test_route_auto_publishes_complete_high_confidence(): | |
| assert _route(_result(confidence=0.9)) == "classified" | |
| def test_route_sends_complete_low_confidence_to_review(): | |
| assert _route(_result(confidence=0.6)) == "needs_review" | |
| def test_route_auto_discards_incomplete_low_confidence(): | |
| # e.g. celebrity gossip the model correctly isn't confident about | |
| assert _route(_result(category=None, business_types=(), location=None, confidence=0.2)) == "rejected" | |
| def test_route_sends_incomplete_but_confident_to_review(): | |
| assert _route(_result(location=None, confidence=0.7)) == "needs_review" | |
| def test_route_treats_missing_confidence_as_zero(): | |
| assert _route(_result(category=None, business_types=(), location=None, confidence=None)) == "rejected" | |