Spaces:
Running
Running
File size: 6,726 Bytes
1f5f31a e740385 1f5f31a a32af27 1f5f31a e740385 a32af27 e740385 | 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 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """
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) ───────────────────────────────────────────────
@pytest.fixture
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"
|