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_start_time_and_artists_persisted(tmp_path):56    con = db.connect(tmp_path / "t.db")57    ev = Event(source="evenko", external_id="s1", url="https://x/s1",58               title="Deep Purple", city="Montréal",59               start_date="2027-01-15", start_time="19h30",60               artists=["Deep Purple", " deep purple "]).finalize()61    assert ev.start_time == "19:30"          # normalisée "HH:MM"62    assert ev.artists == ["Deep Purple"]     # dédupliqués63    db.sync_source(con, "evenko", [ev])64    row = con.execute("SELECT start_time, artists FROM events").fetchone()65    assert row["start_time"] == "19:30"66    assert row["artists"] == '["Deep Purple"]'676869def test_migration_adds_columns(tmp_path):70    p = tmp_path / "old.db"71    con = db.connect(p)72    # simuler une base d'avant la vague 2 (sans les colonnes additives)73    con.execute("ALTER TABLE events DROP COLUMN start_time")74    con.execute("ALTER TABLE events DROP COLUMN artists")75    con.commit()76    con.close()77    con = db.connect(p)        # la connexion migre (colonnes additives)78    cols = {r["name"] for r in con.execute("PRAGMA table_info(events)")}79    assert {"start_time", "artists"} <= cols808182def test_dedup_key_cross_source():83    a = _ev(source="sitq", title="Grand marché de Noël")84    b = Event(source="montreal", external_id="99", url="https://y",85              title="  GRAND MARCHÉ de Noël ", city="Québec",86              start_date="2027-01-15").finalize()87    assert a.dedup_key() == b.dedup_key()88    assert a.dedup_key() != _ev(title="Autre événement").dedup_key()89