spb/job-ka

1# =============================================================================2# Job·Ka — Groupe KA3# Auteur  : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : jobka/fixtures.py6# Rôle    : Enregistrement/rejeu des réponses HTTP des sources pour les tests7#           hors ligne (`python run.py record <source>` puis pytest)8# Créé    : 2026-08-17   Modifié : 2026-08-179# =============================================================================10"""Enregistrement/rejeu des réponses HTTP des sources.1112`python run.py record <source>` capture toutes les réponses (HTML/JSON) dans13tests/fixtures/<source>/ + un instantané `expected.json` du parsing. Les tests14(tests/test_connectors.py) rejouent ces fixtures hors-ligne : si un connecteur15régresse, le test casse — sans toucher au site réel.16"""17from __future__ import annotations1819import hashlib20import json21import re22from pathlib import Path2324import requests25from requests.adapters import BaseAdapter, HTTPAdapter2627from .schema import JobPosting2829ROOT = Path(__file__).resolve().parent.parent30FIXTURES_DIR = ROOT / "tests" / "fixtures"313233class MissingFixture(Exception):34    """Le connecteur a demandé une URL absente des fixtures enregistrées."""353637# Secrets tiers présents dans le HTML public des sites sources (clés Google38# Maps, jetons) : caviardés à l'enregistrement.39_SECRET_RES = [40    (re.compile(rb"AIza[0-9A-Za-z_-]{35}"), b"AIza_CLE_CAVIARDEE_JOBKA_000000000000000"),41    (re.compile(rb"(sk|pk)\.eyJ[A-Za-z0-9_.=-]+"), rb"\1.JETON_CAVIARDE_JOBKA"),42]434445def _redact(content: bytes) -> bytes:46    for rx, repl in _SECRET_RES:47        content = rx.sub(repl, content)48    return content495051def _req_key(method: str, url: str, body: bytes | str | None) -> str:52    if isinstance(body, str):53        body = body.encode("utf-8")54    h = hashlib.sha1()55    h.update(f"{method.upper()} {url} ".encode())56    if body:57        h.update(body)58    return h.hexdigest()[:20]596061def _ext_for(content_type: str) -> str:62    if "json" in content_type:63        return ".json"64    if "html" in content_type:65        return ".html"66    return ".txt"676869# ---------------------------------------------------------------------------70# Enregistrement71# ---------------------------------------------------------------------------7273class _RecordAdapter(HTTPAdapter):74    def __init__(self, store: dict, out_dir: Path):75        super().__init__()76        self.store = store77        self.out_dir = out_dir7879    def send(self, request, **kwargs):  # type: ignore[override]80        resp = super().send(request, **kwargs)81        key = _req_key(request.method or "GET", request.url or "", request.body)82        ctype = resp.headers.get("Content-Type", "")83        fname = key + _ext_for(ctype)84        (self.out_dir / fname).write_bytes(_redact(resp.content))85        self.store[key] = {86            "method": request.method, "url": request.url,87            "status": resp.status_code, "content_type": ctype, "file": fname,88        }89        if resp.headers.get("Location"):90            self.store[key]["location"] = resp.headers["Location"]91        return resp929394def record(source_id: str) -> dict:95    """Exécute le connecteur en enregistrant chaque réponse HTTP en fixture.9697    Écrit tests/fixtures/<source>/{index.json, expected.json, *.html|json}.98    Retourne un résumé {source, requests, jobs}.99    """100    from .connectors import CONNECTORS101    cls = CONNECTORS[source_id]102    out_dir = FIXTURES_DIR / source_id103    out_dir.mkdir(parents=True, exist_ok=True)104    for old in out_dir.iterdir():105        old.unlink()106107    inst = cls()108    inst.use_detail_cache = False        # capturer aussi les pages détail109    store: dict = {}110    adapter = _RecordAdapter(store, out_dir)111    inst.session.mount("https://", adapter)112    inst.session.mount("http://", adapter)113114    postings = inst.fetch()115    (out_dir / "index.json").write_text(116        json.dumps(store, ensure_ascii=False, indent=1), encoding="utf-8")117    (out_dir / "expected.json").write_text(118        json.dumps(snapshot(postings), ensure_ascii=False, indent=1),119        encoding="utf-8")120    return {"source": source_id, "requests": len(store), "jobs": len(postings)}121122123def snapshot(postings: list[JobPosting]) -> dict:124    """Instantané compact et déterministe du parsing (champs bruts, pré-finalize).125126    Les dates relatives normalisées dépendent du jour d'exécution : on ne127    snapshotte que les champs stables pour éviter les tests fragiles.128    """129    rows = []130    for p in sorted(postings, key=lambda x: x.uid):131        rows.append({132            "uid": p.uid, "url": p.url, "title": p.title,133            "employer": p.employer, "city": p.city,134            "location_label": p.location_label,135            "salary_min": p.salary_min, "salary_max": p.salary_max,136            "salary_unit": p.salary_unit,137            "employment_type": p.employment_type, "ats": p.ats,138            "desc_len": len(p.description),139        })140    return {"count": len(rows), "jobs": rows}141142143# ---------------------------------------------------------------------------144# Rejeu145# ---------------------------------------------------------------------------146147class _ReplayAdapter(BaseAdapter):148    def __init__(self, index: dict, fixtures_dir: Path):149        super().__init__()150        self.index = index151        self.dir = fixtures_dir152153    def send(self, request, **kwargs):  # type: ignore[override]154        key = _req_key(request.method or "GET", request.url or "", request.body)155        meta = self.index.get(key)156        if meta is None:157            raise MissingFixture(f"{request.method} {request.url}")158        resp = requests.models.Response()159        resp.status_code = meta["status"]160        resp._content = (self.dir / meta["file"]).read_bytes()161        resp.headers["Content-Type"] = meta.get("content_type", "")162        if meta.get("location"):163            resp.headers["Location"] = meta["location"]164        resp.url = request.url or ""165        resp.request = request166        resp.encoding = "utf-8"167        return resp168169    def close(self):170        pass171172173def replay_connector(source_id: str):174    """Instancie le connecteur branché sur ses fixtures (aucun réseau)."""175    from .connectors import CONNECTORS176    cls = CONNECTORS[source_id]177    fdir = FIXTURES_DIR / source_id178    index = json.loads((fdir / "index.json").read_text(encoding="utf-8"))179180    inst = cls()181    inst.request_delay = 0.0             # pas de politesse hors-ligne182    inst.use_detail_cache = False        # tests indépendants de la BD183    adapter = _ReplayAdapter(index, fdir)184    inst.session.mount("https://", adapter)185    inst.session.mount("http://", adapter)186    return inst187188189def recorded_sources() -> list[str]:190    """Sources ayant des fixtures enregistrées (pour paramétrer les tests)."""191    if not FIXTURES_DIR.exists():192        return []193    return sorted(p.name for p in FIXTURES_DIR.iterdir()194                  if (p / "index.json").exists())195