# ============================================================================= # Job·Ka — Groupe KA # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : jobka/fixtures.py # Rôle : Enregistrement/rejeu des réponses HTTP des sources pour les tests # hors ligne (`python run.py record ` puis pytest) # Créé : 2026-08-17 Modifié : 2026-08-17 # ============================================================================= """Enregistrement/rejeu des réponses HTTP des sources. `python run.py record ` capture toutes les réponses (HTML/JSON) dans tests/fixtures// + un instantané `expected.json` du parsing. Les tests (tests/test_connectors.py) rejouent ces fixtures hors-ligne : si un connecteur régresse, le test casse — sans toucher au site réel. """ from __future__ import annotations import hashlib import json import re from pathlib import Path import requests from requests.adapters import BaseAdapter, HTTPAdapter from .schema import JobPosting ROOT = Path(__file__).resolve().parent.parent FIXTURES_DIR = ROOT / "tests" / "fixtures" class MissingFixture(Exception): """Le connecteur a demandé une URL absente des fixtures enregistrées.""" # Secrets tiers présents dans le HTML public des sites sources (clés Google # Maps, jetons) : caviardés à l'enregistrement. _SECRET_RES = [ (re.compile(rb"AIza[0-9A-Za-z_-]{35}"), b"AIza_CLE_CAVIARDEE_JOBKA_000000000000000"), (re.compile(rb"(sk|pk)\.eyJ[A-Za-z0-9_.=-]+"), rb"\1.JETON_CAVIARDE_JOBKA"), ] def _redact(content: bytes) -> bytes: for rx, repl in _SECRET_RES: content = rx.sub(repl, content) return content def _req_key(method: str, url: str, body: bytes | str | None) -> str: if isinstance(body, str): body = body.encode("utf-8") h = hashlib.sha1() h.update(f"{method.upper()} {url} ".encode()) if body: h.update(body) return h.hexdigest()[:20] def _ext_for(content_type: str) -> str: if "json" in content_type: return ".json" if "html" in content_type: return ".html" return ".txt" # --------------------------------------------------------------------------- # Enregistrement # --------------------------------------------------------------------------- class _RecordAdapter(HTTPAdapter): def __init__(self, store: dict, out_dir: Path): super().__init__() self.store = store self.out_dir = out_dir def send(self, request, **kwargs): # type: ignore[override] resp = super().send(request, **kwargs) key = _req_key(request.method or "GET", request.url or "", request.body) ctype = resp.headers.get("Content-Type", "") fname = key + _ext_for(ctype) (self.out_dir / fname).write_bytes(_redact(resp.content)) self.store[key] = { "method": request.method, "url": request.url, "status": resp.status_code, "content_type": ctype, "file": fname, } if resp.headers.get("Location"): self.store[key]["location"] = resp.headers["Location"] return resp def record(source_id: str) -> dict: """Exécute le connecteur en enregistrant chaque réponse HTTP en fixture. Écrit tests/fixtures//{index.json, expected.json, *.html|json}. Retourne un résumé {source, requests, jobs}. """ from .connectors import CONNECTORS cls = CONNECTORS[source_id] out_dir = FIXTURES_DIR / source_id out_dir.mkdir(parents=True, exist_ok=True) for old in out_dir.iterdir(): old.unlink() inst = cls() inst.use_detail_cache = False # capturer aussi les pages détail store: dict = {} adapter = _RecordAdapter(store, out_dir) inst.session.mount("https://", adapter) inst.session.mount("http://", adapter) postings = inst.fetch() (out_dir / "index.json").write_text( json.dumps(store, ensure_ascii=False, indent=1), encoding="utf-8") (out_dir / "expected.json").write_text( json.dumps(snapshot(postings), ensure_ascii=False, indent=1), encoding="utf-8") return {"source": source_id, "requests": len(store), "jobs": len(postings)} def snapshot(postings: list[JobPosting]) -> dict: """Instantané compact et déterministe du parsing (champs bruts, pré-finalize). Les dates relatives normalisées dépendent du jour d'exécution : on ne snapshotte que les champs stables pour éviter les tests fragiles. """ rows = [] for p in sorted(postings, key=lambda x: x.uid): rows.append({ "uid": p.uid, "url": p.url, "title": p.title, "employer": p.employer, "city": p.city, "location_label": p.location_label, "salary_min": p.salary_min, "salary_max": p.salary_max, "salary_unit": p.salary_unit, "employment_type": p.employment_type, "ats": p.ats, "desc_len": len(p.description), }) return {"count": len(rows), "jobs": rows} # --------------------------------------------------------------------------- # Rejeu # --------------------------------------------------------------------------- class _ReplayAdapter(BaseAdapter): def __init__(self, index: dict, fixtures_dir: Path): super().__init__() self.index = index self.dir = fixtures_dir def send(self, request, **kwargs): # type: ignore[override] key = _req_key(request.method or "GET", request.url or "", request.body) meta = self.index.get(key) if meta is None: raise MissingFixture(f"{request.method} {request.url}") resp = requests.models.Response() resp.status_code = meta["status"] resp._content = (self.dir / meta["file"]).read_bytes() resp.headers["Content-Type"] = meta.get("content_type", "") if meta.get("location"): resp.headers["Location"] = meta["location"] resp.url = request.url or "" resp.request = request resp.encoding = "utf-8" return resp def close(self): pass def replay_connector(source_id: str): """Instancie le connecteur branché sur ses fixtures (aucun réseau).""" from .connectors import CONNECTORS cls = CONNECTORS[source_id] fdir = FIXTURES_DIR / source_id index = json.loads((fdir / "index.json").read_text(encoding="utf-8")) inst = cls() inst.request_delay = 0.0 # pas de politesse hors-ligne inst.use_detail_cache = False # tests indépendants de la BD adapter = _ReplayAdapter(index, fdir) inst.session.mount("https://", adapter) inst.session.mount("http://", adapter) return inst def recorded_sources() -> list[str]: """Sources ayant des fixtures enregistrées (pour paramétrer les tests).""" if not FIXTURES_DIR.exists(): return [] return sorted(p.name for p in FIXTURES_DIR.iterdir() if (p / "index.json").exists())