# ============================================================================== # Author: Simon-Pierre Boucher # File: tests/test_expansion.py # Desc: Phase 3 (expansion des sources) — tests des nouveaux connecteurs sur # fixtures réelles + annuaire de salles + plages de dates françaises. # Aucun appel réseau (CLAUDE.md §18). # ============================================================================== from __future__ import annotations import json from pathlib import Path import pytest from sortika.connectors import CONNECTORS from sortika.normalize import parse_date_range_fr FIXTURES = Path(__file__).parent / "fixtures" def _no_cache(monkeypatch, connector): """Cache disque neutralisé : ni lecture ni écriture pendant les tests.""" monkeypatch.setattr(connector, "load_cache", lambda: {}) monkeypatch.setattr(connector, "save_cache", lambda cache: None) def test_expansion_registry_autodiscovers(): for sid in ("lavalweb", "outaouais", "lanaudiere", "saguenaysorties", "bassaintlaurent", "quebecanimee", "placedesarts", "lhjmq", "sallesjsonld"): assert sid in CONNECTORS, sid def test_parse_date_range_fr(): assert parse_date_range_fr("Du 11 au 13 septembre 2026") == \ ("2026-09-11", "2026-09-13") assert parse_date_range_fr("Du 4 juin au 22 août 2026") == \ ("2026-06-04", "2026-08-22") assert parse_date_range_fr("Du 27 août 2026 au 3 janvier 2027") == \ ("2026-08-27", "2027-01-03") assert parse_date_range_fr("17 août 2026") == ("2026-08-17", "2026-08-17") assert parse_date_range_fr("n'importe quoi") == (None, None) assert parse_date_range_fr(None) == (None, None) def test_lavalweb_parses_fixture(monkeypatch): from sortika.connectors.lavalweb import LavalWebConnector c = LavalWebConnector() _no_cache(monkeypatch, c) html = (FIXTURES / "lavalweb_page.html").read_text(encoding="utf-8") pages = [html, ""] # page 2 vide → arrêt pagination monkeypatch.setattr(c, "get", lambda url: type("R", (), {"text": pages.pop(0)})()) monkeypatch.setattr(c, "_enrich_fiches", lambda rows: None) events = [e.finalize() for e in c.fetch()] assert len(events) == 3 ev = events[0] assert ev.source == "lavalweb" assert ev.city == "Laval" and ev.region == "Laval" assert ev.start_date and ev.start_date.startswith("20") assert ev.url.startswith("https://www.laval.ca/calendrier-activites/") # une carte-séance porte la date ET l'heure (« 18 août 2026 | 18 h ») seance = [e for e in events if e.start_time] assert seance and seance[0].venue def test_outaouais_parses_fixture(monkeypatch): from sortika.connectors.outaouais import OutaouaisConnector c = OutaouaisConnector() html = (FIXTURES / "outaouais_page.html").read_text(encoding="utf-8") monkeypatch.setattr(c, "get", lambda url: type("R", (), {"text": html})()) events = [e.finalize() for e in c.fetch()] assert len(events) == 3 ev = events[0] assert ev.source == "outaouais" assert ev.region == "Outaouais" # via tourist_region assert ev.start_date and ev.end_date assert ev.image.startswith("https://admin.tourismeoutaouais.com/") def test_lanaudiere_parses_fixture(monkeypatch): from sortika.connectors.lanaudiere import LanaudiereConnector c = LanaudiereConnector() _no_cache(monkeypatch, c) sitemap = (FIXTURES / "lanaudiere_sitemap.xml").read_text(encoding="utf-8") fiche = (FIXTURES / "lanaudiere_fiche.html").read_text(encoding="utf-8") responses = {True: sitemap} monkeypatch.setattr( c, "get", lambda url: type("R", (), { "text": sitemap if url.endswith(".xml") else fiche})()) events = [e.finalize() for e in c.fetch()] # 2 URLs françaises dans le sitemap (l'anglaise est exclue), même fiche # → dédupliquées par external_id distinct des slugs assert 1 <= len(events) <= 2 ev = events[0] assert ev.source == "lanaudiere" assert ev.title == "Festival Mémoire et Racines" assert ev.city == "Saint-Charles-Borromée" assert ev.region == "Lanaudière" assert (ev.start_date, ev.end_date) == ("2026-07-17", "2026-07-26") assert "tourinsoft" in ev.image def test_saguenaysorties_parses_fixture(monkeypatch): from sortika.connectors.saguenaysorties import SaguenaySortiesConnector c = SaguenaySortiesConnector() _no_cache(monkeypatch, c) liste = (FIXTURES / "saguenaysorties_liste.html").read_text(encoding="utf-8") detail = (FIXTURES / "saguenaysorties_detail.html").read_text(encoding="utf-8") monkeypatch.setattr( c, "_post", lambda url, data: detail if url.endswith("details") else (liste if data["offset"] == "0" else "")) events = [e.finalize() for e in c.fetch()] assert len(events) == 2 ev = events[0] assert ev.source == "saguenaysorties" assert ev.city and ev.venue assert ev.region == "Saguenay–Lac-Saint-Jean" assert ev.start_date == "2026-08-20" and ev.start_time == "08:30" assert ev.url.startswith("https://sorties.saguenay.ca/festivals?id=") assert "archéologues" in ev.description # détail ajax branché def test_bassaintlaurent_parses_fixture(monkeypatch): from sortika.connectors.bassaintlaurent import BasSaintLaurentConnector c = BasSaintLaurentConnector() _no_cache(monkeypatch, c) liste = (FIXTURES / "bassaintlaurent_liste.html").read_text(encoding="utf-8") fiche = (FIXTURES / "bassaintlaurent_fiche.html").read_text(encoding="utf-8") monkeypatch.setattr(c, "get", lambda url: type("R", (), {"text": liste})()) monkeypatch.setattr(c, "fetch_many", lambda urls, worker, max_workers=3: {u: c._parse_fiche(fiche) for u in urls}) events = [e.finalize() for e in c.fetch()] assert len(events) == 2 ev = events[0] assert ev.source == "bassaintlaurent" assert ev.region == "Bas-Saint-Laurent" assert ev.start_date and ev.end_date and ev.city assert ev.lat and 44.0 < ev.lat < 63.0 # GPS exact de la fiche def test_quebecanimee_parses_fixture(monkeypatch): from sortika.connectors.quebecanimee import QuebecAnimeeConnector c = QuebecAnimeeConnector() _no_cache(monkeypatch, c) html = (FIXTURES / "quebecanimee_liste.html").read_text(encoding="utf-8") monkeypatch.setattr(c, "get", lambda url: type("R", (), {"text": html})()) monkeypatch.setattr(c, "_enrich_fiches", lambda rows: None) events = [e.finalize() for e in c.fetch()] assert len(events) == 3 ev = events[0] assert ev.source == "quebecanimee" assert ev.city == "Québec" and ev.region == "Capitale-Nationale" assert ev.start_date and ev.start_date.startswith("2026") # data-prix « Gratuit » → gratuité réelle, jamais inventée assert any(e.is_free for e in events) def test_placedesarts_parses_fixture(monkeypatch): from sortika.connectors.placedesarts import PlaceDesArtsConnector c = PlaceDesArtsConnector() _no_cache(monkeypatch, c) fiche = (FIXTURES / "placedesarts_fiche.html").read_text(encoding="utf-8") liste = 'x' monkeypatch.setattr(c, "get", lambda url: type("R", (), {"text": liste})()) monkeypatch.setattr(c, "fetch_many", lambda paths, worker, max_workers=4: {p: c._parse_fiche(fiche) for p in paths}) events = [e.finalize() for e in c.fetch()] assert len(events) >= 3 # une fiche = N séances ev = events[0] assert ev.source == "placedesarts" assert ev.venue == "Théâtre Jean-Duceppe" assert ev.city == "Montréal" and ev.organizer == "Duceppe" assert ev.start_time and len(ev.start_time) == 5 # prix réels de la barre latérale (« 33 $ à 97,50 $ ») assert ev.price_min == 33.0 and ev.is_free is False # deux séances distinctes = deux dedup_key distincts (heure incluse) assert events[0].dedup_key() != events[1].dedup_key() def test_lhjmq_parses_fixture(monkeypatch): from sortika.connectors.lhjmq import LhjmqConnector c = LhjmqConnector() seasons = json.loads((FIXTURES / "lhjmq_seasons.json").read_text(encoding="utf-8")) sched = json.loads((FIXTURES / "lhjmq_schedule.json").read_text(encoding="utf-8")) monkeypatch.setattr(c, "get_json", lambda url: seasons if "view=seasons" in url else sched) monkeypatch.setattr(c, "_current_seasons", lambda: ["214"]) events = [e.finalize() for e in c.fetch()] # seuls les matchs à domicile AU QUÉBEC sont publiés (Terre-Neuve exclue) assert events and all(e.city != "Terre-Neuve" for e in events) ev = events[0] assert ev.source == "lhjmq" assert ev.title.startswith("LHJMQ : ") assert ev.venue and ev.start_date and ev.start_time assert "sport" in ev.categories assert ev.url.startswith("https://theqmjhl.ca/gamecentre/") def test_sallesjsonld_parses_fixture(monkeypatch): from sortika.connectors.sallesjsonld import SallesJsonLdConnector c = SallesJsonLdConnector() _no_cache(monkeypatch, c) liste = (FIXTURES / "sallesjsonld_liste.html").read_text(encoding="utf-8") fiche = (FIXTURES / "sallesjsonld_fiche.html").read_text(encoding="utf-8") salle = {"id": "grandtheatre", "nom": "Grand Théâtre de Québec", "listes": ["https://grandtheatre.qc.ca/programmation"], "motif": r"grandtheatre\.qc\.ca/programmation/[a-z0-9-]{3,}", "salle": "Grand Théâtre de Québec", "ville": "Québec"} monkeypatch.setattr(c, "_load_config", lambda: [salle]) monkeypatch.setattr(c, "get", lambda url: type("R", (), {"text": liste})()) monkeypatch.setattr(c, "fetch_many", lambda urls, worker, max_workers=3: {u: c._fiche_events(salle, u, fiche) for u in urls}) events = [e.finalize() for e in c.fetch()] assert len(events) == 1 ev = events[0] assert ev.source == "sallesjsonld" assert ev.venue and ev.city == "Québec" # le Grand Théâtre publie ses horodatages en UTC (+00:00) → convertis # en heure locale de Montréal (18:00 UTC = 14:00 EDT) assert ev.start_date == "2027-04-18" assert ev.start_time == "14:00" def test_venues_directory(tmp_path): """Annuaire de salles : apprentissage, matching prudent, application.""" from sortika import db, venues con = db.connect(tmp_path / "t.db") venues.ensure_schema(con) now = 0 # 2 événements géocodés par la source (même salle) + 2 sans GPS rows = [ ("a:1", "a", "1", "MTelus", "Montréal", 45.51, -73.56), ("a:2", "a", "2", "MTelus", "Montréal", 45.51, -73.56), ("b:1", "b", "1", "MTelus", "Montréal", None, None), # nom composé : le segment le plus long doit matcher l'annuaire ("b:2", "b", "2", "Théâtre Desjardins | MTelus", "Montréal", None, None), # ville en conflit : ne doit JAMAIS matcher ("b:3", "b", "3", "MTelus", "Québec", None, None), ] for uid, src, ext, venue, city, lat, lng in rows: con.execute( "INSERT INTO events (uid, source, external_id, title, venue, city," " lat, lng, active) VALUES (?,?,?,?,?,?,?,?,1)", (uid, src, ext, "T", venue, city, lat, lng)) con.commit() assert venues.learn(con) >= 1 stats = venues.apply(con) assert stats["events_filled"] == 2 got = con.execute("SELECT lat FROM events WHERE uid='b:1'").fetchone() assert got["lat"] == 45.51 conflict = con.execute("SELECT lat FROM events WHERE uid='b:3'").fetchone() assert conflict["lat"] is None # jamais de GPS inventé # alias : nom seul matché uniquement si UNIQUE dans l'annuaire con.execute("INSERT INTO venues (name, name_key, city, city_key, lat, lng," " origin) VALUES ('MTelus','mtelus','Trois-Rivières'," "'trois rivieres', 46.34, -72.54, 'osm')") con.commit() assert venues._match_row(con, "MTelus", "Laval") is None # ambigu → rien def test_venues_aliases(): from sortika.venues import name_aliases assert name_aliases("Salle Wilfrid-Pelletier, Place des Arts") == [ "salle wilfrid pelletier place des arts", "salle wilfrid pelletier", "place des arts"] assert name_aliases("") == []