# ============================================ # Projet : API-KA # Fichier : tests/test_collectors.py # Node : m3u96b # Author : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Date : 2026-08-16 # ============================================ """Tests des collecteurs : pipeline complet, déduplication, échecs et statut retried.""" from __future__ import annotations import datetime import pytest from sqlalchemy import func, select from src.collectors import COLLECTORS, build_collectors, get_collector from src.collectors.louka_collector import LoukaCollector from src.config import SERVICES from src.database.db import session_scope from src.database.models import DATA_MODELS, CollectionRun SAMPLE_PAYLOAD = [ {"id": 1, "nom": "enregistrement-a"}, {"id": 2, "nom": "enregistrement-b"}, ] def _count_rows(service: str, date_key: datetime.date) -> int: model = DATA_MODELS[service] with session_scope() as session: return ( session.execute( select(func.count()) .select_from(model) .where(model.source == service, model.date_key == date_key) ).scalar() or 0 ) def _last_run(service: str, date_key: datetime.date) -> CollectionRun | None: with session_scope() as session: return ( session.execute( select(CollectionRun) .where( CollectionRun.service == service, CollectionRun.date_key == date_key ) .order_by(CollectionRun.id.desc()) .limit(1) ) .scalars() .first() ) def test_registry_covers_all_services() -> None: """Chaque service KA a un collecteur enregistré.""" assert set(COLLECTORS) == set(SERVICES) assert len(build_collectors()) == 9 with pytest.raises(ValueError): get_collector("inconnu") def test_run_success(monkeypatch: pytest.MonkeyPatch) -> None: """Le pipeline complet insère les données et journalise un run success.""" date_key = datetime.date(2026, 8, 1) collector = LoukaCollector(attempts=1, delays=(0,)) monkeypatch.setattr(collector, "fetch", lambda: SAMPLE_PAYLOAD) result = collector.run(date_key=date_key) assert result["status"] == "success" assert result["records_count"] == 2 assert _count_rows("louka", date_key) == 2 run = _last_run("louka", date_key) assert run is not None assert run.status == "success" assert run.records_count == 2 assert run.node def test_run_deduplicates_by_checksum(monkeypatch: pytest.MonkeyPatch) -> None: """Une seconde collecte du même payload n'insère aucun doublon.""" date_key = datetime.date(2026, 8, 2) collector = LoukaCollector(attempts=1, delays=(0,)) monkeypatch.setattr(collector, "fetch", lambda: SAMPLE_PAYLOAD) first = collector.run(date_key=date_key) second = collector.run(date_key=date_key) assert first["records_count"] == 2 assert second["status"] == "success" assert second["records_count"] == 0 assert _count_rows("louka", date_key) == 2 def test_run_failure_after_retries(monkeypatch: pytest.MonkeyPatch) -> None: """Après épuisement des relances : run failed + message d'erreur journalisé.""" date_key = datetime.date(2026, 8, 3) collector = LoukaCollector(attempts=2, delays=(0, 0)) def broken() -> None: raise RuntimeError("source indisponible") monkeypatch.setattr(collector, "fetch", broken) result = collector.run(date_key=date_key) assert result["status"] == "failed" assert "source indisponible" in (result["error"] or "") run = _last_run("louka", date_key) assert run is not None assert run.status == "failed" assert run.records_count == 0 assert "source indisponible" in (run.error_message or "") def test_run_retried_status(monkeypatch: pytest.MonkeyPatch) -> None: """Un succès après relance est journalisé avec le statut retried.""" date_key = datetime.date(2026, 8, 4) collector = LoukaCollector(attempts=2, delays=(0, 0)) calls = {"count": 0} def flaky() -> list[dict]: calls["count"] += 1 if calls["count"] == 1: raise RuntimeError("transitoire") return SAMPLE_PAYLOAD monkeypatch.setattr(collector, "fetch", flaky) result = collector.run(date_key=date_key) assert result["status"] == "retried" assert result["records_count"] == 2 run = _last_run("louka", date_key) assert run is not None assert run.status == "retried" def test_validate_rejects_empty_payload() -> None: """Un payload vide ou None est rejeté par la validation.""" collector = LoukaCollector(attempts=1, delays=(0,)) with pytest.raises(ValueError): collector.validate(None) with pytest.raises(ValueError): collector.validate([]) def test_checksum_is_stable_and_order_insensitive() -> None: """Le checksum SHA-256 est canonique (indépendant de l'ordre des clés).""" a = LoukaCollector.checksum({"x": 1, "y": 2}) b = LoukaCollector.checksum({"y": 2, "x": 1}) c = LoukaCollector.checksum({"x": 1, "y": 3}) assert a == b assert a != c assert len(a) == 64 # ------------------------------------------------------------------- job-ka def test_jobka_pipeline_inserts_into_jobka_data( monkeypatch: pytest.MonkeyPatch, ) -> None: """Le collecteur job-ka insère dans jobka_data et journalise un run success.""" from src.collectors.jobka_collector import JobkaCollector date_key = datetime.date(2026, 8, 5) collector = JobkaCollector(attempts=1, delays=(0,)) monkeypatch.setattr( collector, "fetch", lambda: [ {"uid": "a1", "title": "Analyste", "employer": "Ville de Québec"}, {"uid": "b2", "title": "Technicien", "employer": "CHU de Québec"}, ], ) result = collector.run(date_key=date_key) assert result["service"] == "jobka" assert result["status"] == "success" assert result["records_count"] == 2 assert _count_rows("jobka", date_key) == 2 run = _last_run("jobka", date_key) assert run is not None assert run.status == "success" def test_jobka_fetch_paginates_offset_until_total( monkeypatch: pytest.MonkeyPatch, ) -> None: """fetch() pagine /api/jobs (limit/offset, clé "jobs") jusqu'au total annoncé.""" from src.collectors import base_collector from src.collectors.jobka_collector import JobkaCollector total = 7 jobs = [{"uid": f"job-{i}", "title": f"Poste {i}"} for i in range(total)] calls: list[dict] = [] class FakeResponse: def __init__(self, offset: int, limit: int) -> None: self._batch = jobs[offset : offset + limit] def raise_for_status(self) -> None: return None def json(self) -> dict: return {"total": total, "count": len(self._batch), "jobs": self._batch} class FakeClient: def __init__(self, *args, **kwargs) -> None: pass def __enter__(self) -> "FakeClient": return self def __exit__(self, *exc) -> None: return None def get(self, url: str, params: dict) -> FakeResponse: calls.append({"url": url, **params}) return FakeResponse(params["offset"], params["limit"]) monkeypatch.setattr(base_collector.httpx, "Client", FakeClient) collector = JobkaCollector(attempts=1, delays=(0,)) monkeypatch.setattr(type(collector), "page_size", 3) items = collector.fetch() assert [i["uid"] for i in items] == [f"job-{i}" for i in range(total)] # 3 pages : offsets 0, 3, 6 — l'arrêt vient du total annoncé par l'API. assert [c["offset"] for c in calls] == [0, 3, 6] assert all(c["limit"] == 3 for c in calls) def test_jobka_registered_with_expected_settings() -> None: """job-ka est enregistré dans le registre avec la bonne configuration.""" from src.collectors.jobka_collector import JobkaCollector assert COLLECTORS["jobka"] is JobkaCollector collector = get_collector("jobka") assert collector.service == "jobka" assert collector.items_key == "jobs" assert collector.pagination == "offset" assert DATA_MODELS["jobka"].__tablename__ == "jobka_data"