spb/sorti-ka

1# ==============================================================================2# Author: Simon-Pierre Boucher <contact@spboucher.ai>3# File:   tests/test_db.py4# Desc:   Tests de la persistance — upsert, détection de changements, délai de5#         grâce (2 syncs), dérive, clé de déduplication inter-sources.6# ==============================================================================7from __future__ import annotations89from sortika import db10from sortika.schema import Event111213def _ev(eid="1", title="Fête au parc", price="Gratuit", source="sitq"):14    return Event(source=source, external_id=eid, url=f"https://x/{eid}",15                 title=title, city="Québec", tourist_region="Québec",16                 start_date="2027-01-15", price_label=price).finalize()171819def test_sync_lifecycle(tmp_path):20    con = db.connect(tmp_path / "t.db")2122    s = db.sync_source(con, "sitq", [_ev()])23    assert (s["added"], s["removed"]) == (1, 0)2425    # inchangé26    s = db.sync_source(con, "sitq", [_ev()])27    assert (s["added"], s["updated"], s["unchanged"]) == (0, 0, 1)2829    # changement de contenu détecté30    s = db.sync_source(con, "sitq", [_ev(price="Adulte 10 $")])31    assert s["updated"] == 132    row = con.execute("SELECT is_free, price_min FROM events").fetchone()33    assert (row["is_free"], row["price_min"]) == (0, 10.0)3435    # disparition : grâce au 1er raté, retrait au 2e36    s = db.sync_source(con, "sitq", [])37    assert s["removed"] == 038    s = db.sync_source(con, "sitq", [])39    assert s["removed"] == 140    assert con.execute("SELECT active FROM events").fetchone()[0] == 0414243def test_drift_suspends_removals(tmp_path):44    con = db.connect(tmp_path / "t.db")45    batch = [_ev(eid=str(i), title=f"Événement {i}") for i in range(20)]46    for _ in range(3):47        db.sync_source(con, "sitq", batch)48    # chute brutale à 2 événements → alerte, aucun retrait49    s = db.sync_source(con, "sitq", batch[:2])50    assert "alert" in s and s["removed"] == 051    n = con.execute("SELECT COUNT(*) FROM events WHERE active=1").fetchone()[0]52    assert n == 20535455def test_dedup_key_cross_source():56    a = _ev(source="sitq", title="Grand marché de Noël")57    b = Event(source="montreal", external_id="99", url="https://y",58              title="  GRAND MARCHÉ de Noël ", city="Québec",59              start_date="2027-01-15").finalize()60    assert a.dedup_key() == b.dedup_key()61    assert a.dedup_key() != _ev(title="Autre événement").dedup_key()62