Blame for tests/test_intel.py

"""
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]]]}


@pytest.fixture
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))


@pytest.fixture
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 ───────────────────────────────────────────────────────────────

@pytest.mark.parametrize("year,expected", [(2024, date(2024, 3, 31)), (2025, date(2025, 4, 20)), (2026, date(2026, 4, 5))])
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