Spaces:
Running
Running
| """ | |
| Unit tests for the intel store (Phase 0 + 1, docs/blueprint/intelligence-system.md): | |
| geometry, gazetteer matching and text resolution, Layer 1 freshness, handoff | |
| receipts, the calendar, resumable weather backfill, category mapping, and the | |
| news → events hook. Everything runs on in-memory SQLite with synthetic data — | |
| no network, and no dependency on the gitignored raw downloads. | |
| """ | |
| from datetime import date, datetime, timedelta, timezone | |
| import pytest | |
| from intel import db, handoff | |
| from intel.freshness import record_attempt, report, source_status | |
| from intel.geo import PolygonIndex, centroid, contains, haversine_km | |
| from intel.ingest import calendar as cal | |
| from intel.ingest import weather | |
| from intel.locations import match_boundaries, normalise, resolve | |
| from intel.subsector_map import subsector_for | |
| def square(x0, y0, x1, y1): | |
| return {"type": "Polygon", "coordinates": [[[x0, y0], [x1, y0], [x1, y1], [x0, y1], [x0, y0]]]} | |
| def con(): | |
| c = db.connect(":memory:") | |
| db.init(c) | |
| yield c | |
| c.close() | |
| def add_location(con, lid, level, name, parent=None, lat=0.0, lon=0.0, aliases=()): | |
| con.execute("insert into locations (id, level, name, parent_id, lat, lon, source) values (?, ?, ?, ?, ?, ?, 'test')", | |
| (lid, level, name, parent, lat, lon)) | |
| for a in (name, *aliases): | |
| con.execute("insert into location_aliases (alias, location_id, origin) values (?, ?, 'test')", (normalise(a), lid)) | |
| def gazetteer(con): | |
| add_location(con, "ng", "country", "Nigeria") | |
| add_location(con, "ng.lagos", "state", "Lagos", "ng") | |
| add_location(con, "ng.oyo", "state", "Oyo", "ng") | |
| add_location(con, "ng.bauchi", "state", "Bauchi", "ng") | |
| add_location(con, "ng.lagos.surulere", "lga", "Surulere", "ng.lagos") | |
| add_location(con, "ng.oyo.surulere", "lga", "Surulere", "ng.oyo") | |
| add_location(con, "ng.lagos.eti-osa", "lga", "Eti-Osa", "ng.lagos", aliases=("Ajah", "Lekki")) | |
| add_location(con, "ng.bauchi.bauchi", "lga", "Bauchi", "ng.bauchi") | |
| con.commit() | |
| return con | |
| # ─── geometry ─────────────────────────────────────────────────────────────── | |
| def test_centroid_of_square(): | |
| assert centroid(square(0, 0, 2, 2)) == pytest.approx((1.0, 1.0)) | |
| def test_centroid_respects_holes(): | |
| g = {"type": "Polygon", "coordinates": [ | |
| [[0, 0], [4, 0], [4, 4], [0, 4], [0, 0]], | |
| [[2, 0], [4, 0], [4, 4], [2, 4], [2, 0]], # right half removed | |
| ]} | |
| assert centroid(g)[0] == pytest.approx(1.0) | |
| def test_contains_and_index(): | |
| idx = PolygonIndex([("a", square(0, 0, 1, 1)), ("b", square(1, 0, 2, 1))]) | |
| assert idx.lookup(0.5, 0.5) == "a" | |
| assert idx.lookup(1.5, 0.5) == "b" | |
| assert idx.lookup(5, 5) is None | |
| assert not contains(square(0, 0, 1, 1), 1.5, 0.5) | |
| def test_haversine_one_degree_at_equator(): | |
| assert haversine_km(0, 0, 1, 0) == pytest.approx(111.2, abs=0.2) | |
| # ─── gazetteer ────────────────────────────────────────────────────────────── | |
| def _feature(name, geom): | |
| return {"properties": {"shapeName": name}, "geometry": geom} | |
| def test_match_boundaries_exact_fuzzy_and_state_placement(): | |
| states = [{"name": "Lagos", "lgas": ["Ifako-Ijaiye", "Surulere"]}, {"name": "Oyo", "lgas": ["Surulere"]}] | |
| adm1 = [_feature("Lagos", square(0, 0, 2, 1)), _feature("Oyo", square(0, 1, 2, 2))] | |
| adm2 = [ | |
| _feature("Ifako/Ijaye", square(0, 0, 1, 1)), # fuzzy | |
| _feature("Surulere", square(1, 0, 2, 1)), # exact, in Lagos by position | |
| _feature("Surulere", square(0, 1, 2, 2)), # same name, in Oyo by position | |
| ] | |
| matched, problems = match_boundaries(states, adm1, adm2) | |
| assert problems == [] | |
| assert matched[("Lagos", "Ifako-Ijaiye")]["properties"]["shapeName"] == "Ifako/Ijaye" | |
| assert centroid(matched[("Oyo", "Surulere")]["geometry"])[1] == pytest.approx(1.5) | |
| def test_match_boundaries_reports_rather_than_guesses(): | |
| states = [{"name": "Lagos", "lgas": ["Agege"]}] | |
| adm1 = [_feature("Lagos", square(0, 0, 1, 1))] | |
| adm2 = [_feature("Completely Different", square(0, 0, 1, 1))] | |
| _, problems = match_boundaries(states, adm1, adm2) | |
| assert any("Agege" in p for p in problems) | |
| def test_resolve_alias_to_lga(gazetteer): | |
| assert resolve(gazetteer, "Flooding in Ajah after heavy rain") == "ng.lagos.eti-osa" | |
| def test_resolve_shared_lga_name_needs_state_context(gazetteer): | |
| assert resolve(gazetteer, "Traders in Surulere protest") is None | |
| assert resolve(gazetteer, "Surulere, Lagos market shut") == "ng.lagos.surulere" | |
| def test_resolve_state_beats_same_named_lga(gazetteer): | |
| assert resolve(gazetteer, "Bauchi governor signs levy") == "ng.bauchi" | |
| def test_resolve_no_mention(gazetteer): | |
| assert resolve(gazetteer, "Celebrity wedding photos") is None | |
| assert resolve(gazetteer, None) is None | |
| def test_resolve_word_boundaries(gazetteer): | |
| # "Oyo" must not match inside another word. | |
| assert resolve(gazetteer, "Coyotes spotted") is None | |
| # ─── Layer 1: freshness ───────────────────────────────────────────────────── | |
| def _source(con, sid): | |
| return con.execute("select * from sources where id = ?", (sid,)).fetchone() | |
| def test_never_fetched_source_is_never(con): | |
| assert source_status(_source(con, "calendar"), datetime.now(timezone.utc))["status"] == "never" | |
| def test_fresh_then_stale_on_fetch_clock(con): | |
| record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=date.today().isoformat()) | |
| now = datetime.now(timezone.utc) | |
| assert source_status(_source(con, "open_meteo_forecast"), now)["status"] == "fresh" | |
| assert source_status(_source(con, "open_meteo_forecast"), now + timedelta(days=2))["status"] == "late" | |
| assert source_status(_source(con, "open_meteo_forecast"), now + timedelta(days=4))["status"] == "stale" | |
| def test_data_clock_catches_old_data_behind_a_fresh_fetch(con): | |
| # Fetched just now, but the newest data is a month old on a daily source. | |
| old = (date.today() - timedelta(days=30)).isoformat() | |
| record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=old) | |
| s = source_status(_source(con, "open_meteo_forecast"), datetime.now(timezone.utc)) | |
| assert s["status"] == "stale" | |
| assert "newest data" in s["reason"] | |
| def test_archive_latency_is_allowed(con): | |
| # The archive legitimately trails by ~a week; 6-day-old data is on time. | |
| record_attempt(con, "open_meteo_archive", ok=True, latest_data_period=(date.today() - timedelta(days=6)).isoformat()) | |
| assert source_status(_source(con, "open_meteo_archive"), datetime.now(timezone.utc))["status"] == "fresh" | |
| def test_failure_after_success_is_failing(con): | |
| record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=date.today().isoformat()) | |
| record_attempt(con, "open_meteo_forecast", ok=False, error="timeout") | |
| s = source_status(_source(con, "open_meteo_forecast"), datetime.now(timezone.utc)) | |
| assert s["status"] == "failing" and "timeout" in s["reason"] | |
| # ─── handoffs ─────────────────────────────────────────────────────────────── | |
| def test_handoff_receipt_once(con): | |
| hid = handoff.produce(con, "r1", handoff.DATA_STREAM, handoff.STALE_AWARENESS, {"ok": True}) | |
| assert [h["id"] for h in handoff.pending(con, handoff.STALE_AWARENESS)] == [hid] | |
| assert handoff.receive(con, hid, "test") is True | |
| assert handoff.receive(con, hid, "test") is False | |
| assert handoff.pending(con, handoff.STALE_AWARENESS) == [] | |
| def test_unreceived_handoff_to_active_layer_goes_overdue(con): | |
| handoff.produce(con, "r1", handoff.DATA_STREAM, handoff.STALE_AWARENESS, {}) | |
| later = datetime.now(timezone.utc) + timedelta(days=2) | |
| assert len(report(con, later)["handoffs_overdue"]) == 1 | |
| def test_handoff_to_unbuilt_layer_is_queued_not_overdue(con): | |
| handoff.produce(con, "r1", handoff.DATA_STREAM, handoff.INTERFACE, {}) | |
| r = report(con, datetime.now(timezone.utc) + timedelta(days=30)) | |
| assert r["handoffs_overdue"] == [] and r["handoffs_queued_for_unbuilt_layers"] == 1 | |
| # ─── calendar ─────────────────────────────────────────────────────────────── | |
| def test_easter(year, expected): | |
| assert cal.easter(year) == expected | |
| def test_democracy_day_moved_in_2019(): | |
| dd = lambda y: next(e["starts_on"] for e in cal.events_for_year(y) if e["key"] == "democracy_day") # noqa: E731 | |
| assert dd(2018) == date(2018, 5, 29) | |
| assert dd(2019) == date(2019, 6, 12) | |
| def test_islamic_holidays_are_estimates(): | |
| events = {e["key"]: e for e in cal.events_for_year(2025)} | |
| assert events["eid_al_fitr"]["starts_on"] == date(2025, 3, 30) | |
| assert events["eid_al_fitr"]["is_estimate"] and not events["christmas"]["is_estimate"] | |
| def test_calendar_ingest_is_idempotent(gazetteer): | |
| first = cal.ingest(gazetteer, 2024, 2025)["events"] | |
| cal.ingest(gazetteer, 2024, 2025) | |
| assert gazetteer.execute("select count(*) from events where source_id = 'calendar'").fetchone()[0] <= first | |
| # ─── weather backfill ─────────────────────────────────────────────────────── | |
| def fake_fetch(calls): | |
| def fetch(url, params): | |
| calls.append(params) | |
| a, b = date.fromisoformat(params.get("start_date") or date.today().isoformat()), \ | |
| date.fromisoformat(params.get("end_date") or (date.today() + timedelta(days=params.get("forecast_days", 1) - 1)).isoformat()) | |
| days = [(a + timedelta(days=i)).isoformat() for i in range((b - a).days + 1)] | |
| return {"daily": {"time": days, **{v: [1.0] * len(days) for v in weather.VARIABLES}}} | |
| return fetch | |
| def test_call_weight(): | |
| assert weather.call_weight(14) == 1 | |
| assert weather.call_weight(365) == pytest.approx(365 / 14) | |
| def test_backfill_stops_at_budget_and_resumes(gazetteer): | |
| start = date(date.today().year - 3, 1, 1) | |
| calls = [] | |
| first = weather.backfill_history(gazetteer, start=start, budget=60, fetch=fake_fetch(calls), sleep=lambda s: None) | |
| assert first["stopped"] == "run budget reached" and first["year_chunks_remaining"] > 0 | |
| weather.backfill_history(gazetteer, start=start, budget=10_000, fetch=fake_fetch(calls), sleep=lambda s: None) | |
| assert weather._missing_years(gazetteer, "ng.lagos", start, date.today() - timedelta(days=weather.ARCHIVE_LAG_DAYS)) == [] | |
| # A third run has nothing to fetch. | |
| n = len(calls) | |
| weather.backfill_history(gazetteer, start=start, budget=10_000, fetch=fake_fetch(calls), sleep=lambda s: None) | |
| assert len(calls) == n | |
| def test_backfill_resumes_current_year_from_last_day(gazetteer): | |
| end = date.today() - timedelta(days=weather.ARCHIVE_LAG_DAYS) | |
| start = date(end.year, 1, 1) | |
| weather.backfill_history(gazetteer, start=start, fetch=fake_fetch([]), sleep=lambda s: None) | |
| gazetteer.execute("delete from observations where period_start > ?", ((end - timedelta(days=3)).isoformat(),)) | |
| spans = weather._missing_years(gazetteer, "ng.lagos", start, end) | |
| assert spans == [(end - timedelta(days=2), end)] | |
| def test_quota_429_is_partial_success_not_error(gazetteer): | |
| def fetch(url, params): | |
| raise weather.QuotaExhausted("daily limit") | |
| out = weather.backfill_history(gazetteer, start=date(date.today().year, 1, 1), fetch=fetch, sleep=lambda s: None) | |
| assert "quota" in out["stopped"] | |
| def test_forecast_rows_are_marked_and_versioned(gazetteer): | |
| weather.fetch_forecast(gazetteer, days=3, fetch=fake_fetch([])) | |
| rows = gazetteer.execute("select distinct is_forecast, issued_at from observations").fetchall() | |
| assert [tuple(r) for r in rows] == [(1, date.today().isoformat())] | |
| # ─── places category mapping ──────────────────────────────────────────────── | |
| def test_subsector_prefers_detailed_taxonomy(): | |
| assert subsector_for("personal_or_beauty_service", "barber") == "hair_services" | |
| assert subsector_for("personal_or_beauty_service", "makeup_artist") == "skincare_cosmetics" | |
| assert subsector_for("restaurant", None) == "catering_food_service" | |
| def test_context_places_have_no_subsector(): | |
| assert subsector_for("christian_place_of_worship", None) is None | |
| assert subsector_for(None, None) is None | |
| def test_mapped_subsectors_are_real_ids(): | |
| from news.business_types import VALID_IDS | |
| from intel.subsector_map import BASIC_TO_SUBSECTOR, TAXONOMY_TO_SUBSECTOR | |
| assert set(BASIC_TO_SUBSECTOR.values()) | set(TAXONOMY_TO_SUBSECTOR.values()) <= set(VALID_IDS) | |
| # ─── news → events ────────────────────────────────────────────────────────── | |
| def test_record_news_item_is_noop_when_store_off(monkeypatch): | |
| from intel.events import record_news_item | |
| monkeypatch.delenv("INTEL_DB_PATH", raising=False) | |
| assert record_news_item({"url_hash": "x", "title": "t"}) is False | |
| def test_record_news_item_writes_located_event(monkeypatch, tmp_path): | |
| from intel.events import record_news_item | |
| path = tmp_path / "intel.db" | |
| monkeypatch.setenv("INTEL_DB_PATH", str(path)) | |
| c = db.connect() | |
| db.init(c) | |
| add_location(c, "ng", "country", "Nigeria") | |
| add_location(c, "ng.lagos", "state", "Lagos", "ng") | |
| add_location(c, "ng.lagos.eti-osa", "lga", "Eti-Osa", "ng.lagos", aliases=("Lekki",)) | |
| c.commit() | |
| item = {"url_hash": "abc", "title": "Lekki traders face new levy", "category": "policy", "location": "Lekki", | |
| "published_at": "2026-10-01T09:00:00+00:00", "raw_text": "..."} | |
| assert record_news_item(item) is True | |
| assert record_news_item(item) is True # idempotent upsert | |
| rows = c.execute("select family, location_id, starts_on from events where source_id = 'news_pipeline'").fetchall() | |
| assert [tuple(r) for r in rows] == [("policy", "ng.lagos.eti-osa", "2026-10-01")] | |
| c.close() | |
| def test_record_news_item_never_raises(monkeypatch, tmp_path): | |
| from intel.events import record_news_item | |
| monkeypatch.setenv("INTEL_DB_PATH", str(tmp_path / "missing-schema.db")) | |
| assert record_news_item({"url_hash": "x", "title": "t"}) is False | |