tests/fake_api.py

"""Fake Cube.js client for offline tests.

Reproduces the behaviour of the real Indicateurs API observed on the live Space
(tests/live_capture.json): metadata cube, source cube whose dimensions do NOT
include `libelle`, data cubes named {theme}_{maille} with measures `id_{n}` and
geographic dimensions geocode_*/libelle_*, string-typed numeric values, `limit`,
`order`, `equals` / `contains` / `set` filters. As on the real API, breakdown
columns (secteur, mode_transport…) and `date_mesure` are returned in every row
even when not requested, and a dimension-only query returns distinct rows.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from src.api_client import BadRequestError, CubeJsClient
from src.models import REGION_CODES

CAPTURE = json.loads((Path(__file__).parent / "live_capture.json").read_text(encoding="utf-8"))

META_DIMS = [
    "id", "libelle", "unite", "description", "methode_calcul", "annees_disponibles",
    "mailles_disponibles", "maille_mini_disponible", "couverture_geographique",
    "completion_region", "completion_departement", "completion_epci", "completion_commune",
    "thematique_fnv",
]
# As on the real API: no `libelle` dimension on the source cube.
SOURCE_DIMS = [
    "id_indicateur", "nom_source", "description", "producteur_source",
    "distributeur_source", "license_source", "lien_page", "date_derniere_extraction",
]
GEO_DIMS = {
    "commune": ["geocode_commune", "libelle_commune"],
    "epci": ["geocode_epci", "libelle_epci"],
    "departement": ["geocode_departement", "libelle_departement"],
    "region": ["geocode_region", "libelle_region"],
}
ALL_LEVELS = ["commune", "epci", "departement", "region"]
# cube -> (levels, indicator ids) — from the captured available_cubes
DATA_CUBES = {
    "conso_enaf_com": (ALL_LEVELS, [611, 545]),
    "stationnement_velo_habitant_com": (ALL_LEVELS, [94]),
    "surfaces_agricoles_com": (["commune"], [23]),
    "surface_bio_epci": (["epci", "region"], [23]),
    "surface_bio_dpt": (["departement", "region"], [23, 42]),
    "dechets_population_dpt": (["departement", "region"], [42]),
    "emission_ges_secteur_com": (ALL_LEVELS, [1]),
    "parts_modales_actifs_com": (ALL_LEVELS, [1053, 955]),
    # commune/EPCI indicator whose commune cube also carries département/région columns
    "climadiag_vague_chaleur_com": (ALL_LEVELS, [1298]),
    "climadiag_vague_chaleur_epci": (["epci"], [1298]),
}
YEARS = ["2020", "2021", "2022"]
CUBE_YEARS = {"climadiag_vague_chaleur_com": ["2005", "2030", "2050", "2100"],
              "climadiag_vague_chaleur_epci": ["2005", "2030", "2050", "2100"]}
# Breakdown columns, always returned by the API
CUBE_AXES = {
    "emission_ges_secteur_com": {"secteur": ["Agriculture", "Déchets", "Energie", "Industrie hors Energie",
                                             "Résidentiel", "Tertiaire", "Transports"]},
    "parts_modales_actifs_com": {"mode_transport": ["Marche", "Transports en commun", "Vélo (depuis 2017)", "Voiture"]},
}
EXTRA_MEASURES = {"climadiag_vague_chaleur_com": ["scenario_bas", "scenario_haut"],
                  "climadiag_vague_chaleur_epci": ["scenario_bas", "scenario_haut"]}
# Metadata not in the capture (axes, summable…), as returned by the real API
META_OVERRIDES: dict[int, dict[str, Any]] = {
    1: {"noms_axes": "secteur", "summable": True,
        "valeur_axes": CUBE_AXES["emission_ges_secteur_com"], "annees_disponibles": "2020, 2021, 2022"},
    1053: {"libelle": "Part d'actifs selon le mode de transport principalement utilisé pour aller travailler",
           "unite": "%", "noms_axes": "mode_transport", "summable": False,
           "valeur_axes": CUBE_AXES["parts_modales_actifs_com"], "mailles_disponibles": "commune, epci, departement, region",
           "annees_disponibles": "2020, 2021, 2022"},
    955: {"noms_axes": "", "summable": False, "unite": "%", "annees_disponibles": "2020, 2021, 2022"},
    1298: {"libelle": "Nombre annuel de jours en vague de chaleur", "unite": "jours", "summable": False,
           "mailles_disponibles": "commune, epci", "annees_disponibles": "2005, 2030, 2050, 2100",
           "completion_epci": 0.82},
}

# Synthetic territories (enough to exercise truncation on communes)
METRO_REGIONS = [code for code in REGION_CODES if not code.startswith("0")]
DEPARTEMENTS = dict.fromkeys(
    [(f"{i:02d}", f"Département {i:02d}") for i in range(1, 96) if i not in (13, 20, 22, 69)]
    + [("2A", "Corse-du-Sud"), ("2B", "Haute-Corse"), ("13", "Bouches-du-Rhône"), ("69", "Rhône"),
       ("22", "Côtes-d'Armor"), ("974", "La Réunion")]
)
EPCIS = {
    "200046977": "Métropole de Lyon",
    "200054781": "Métropole du Grand Paris",
    "200054807": "Métropole d'Aix-Marseille-Provence",
    "243100518": "Toulouse Métropole",
    "243400017": "Montpellier Méditerranée Métropole",
    "200069409": "Communauté d'agglomération Saint-Brieuc Armor Agglomération",
    "249740119": "Communauté intercommunale du Nord de La Réunion",
    **{f"2000{i:05d}": f"Communauté de communes n°{i}" for i in range(1, 60)},
}
DPT_REGION = {"75": "11", "93": "11", "13": "93", "69": "84", "31": "76", "34": "76",
              "22": "53", "35": "53", "56": "53", "974": "04"}
# commune: (label, epci, département)
COMMUNES: dict[str, tuple[str, str, str]] = {
    "75056": ("Paris", "200054781", "75"),
    "13055": ("Marseille", "200054807", "13"),
    "69123": ("Lyon", "200046977", "69"),
    "31555": ("Toulouse", "243100518", "31"),
    "34172": ("Montpellier", "243400017", "34"),
    "34057": ("Castelnau-le-Lez", "243400017", "34"),
    "22278": ("Saint-Brieuc", "200069409", "22"),
    "35258": ("Saint-Brieuc-des-Iffs", "200000001", "35"),
    "56208": ("Saint-Brieuc-de-Mauron", "200000002", "56"),
    "93066": ("Saint-Denis", "200054781", "93"),
    "97411": ("Saint-Denis", "249740119", "974"),
    **{f"{i:05d}": (f"Commune {i}", f"2000{i % 59 + 1:05d}", f"{i % 90 + 1:02d}") for i in range(1000, 1700)},
}


def _region_of(dpt: str) -> str:
    return DPT_REGION.get(dpt) or METRO_REGIONS[sum(map(ord, dpt)) % len(METRO_REGIONS)]


TERRITORIES: dict[str, list[tuple[str, str]]] = {
    "region": [(code, REGION_CODES[code]) for code in METRO_REGIONS],
    "departement": list(DEPARTEMENTS),
    "epci": list(EPCIS.items()),
    "commune": [(code, c[0]) for code, c in COMMUNES.items()],
}
DEPARTEMENT_LABELS = dict(TERRITORIES["departement"])


def _indicator_rows() -> list[dict[str, Any]]:
    """All indicators known from the capture, richest metadata when available."""
    rows: dict[int, dict[str, Any]] = {}
    for item in CAPTURE["list_all"]["indicators"]:
        rows[item["id"]] = dict(item)
    for key, det in CAPTURE["details"].items():
        rows[int(key)].update(det["metadata"])
    for ind_id, override in META_OVERRIDES.items():
        rows.setdefault(ind_id, {"id": ind_id}).update(override)
    return [rows[k] for k in sorted(rows)]


class FakeCubeJsClient(CubeJsClient):
    def __init__(self) -> None:
        super().__init__(token="fake-token")
        self.queries: list[dict[str, Any]] = []
        self.fail_meta = False

    async def get_meta(self) -> dict[str, Any]:
        if self.fail_meta:
            raise BadRequestError("meta unavailable")
        cubes = [
            {"name": "indicateur_metadata", "measures": [], "dimensions": [
                {"name": f"indicateur_metadata.{d}"} for d in META_DIMS]},
            {"name": "indicateur_x_source_metadata", "measures": [], "dimensions": [
                {"name": f"indicateur_x_source_metadata.{d}"} for d in SOURCE_DIMS]},
        ]
        for cube, (levels, ids) in DATA_CUBES.items():
            dims = ["date_mesure", "annee"] + [d for lv in levels for d in GEO_DIMS[lv]] + list(CUBE_AXES.get(cube, {}))
            measures = [f"id_{i}" for i in ids] + EXTRA_MEASURES.get(cube, [])
            cubes.append({
                "name": cube,
                "measures": [{"name": f"{cube}.{m}"} for m in measures],
                "dimensions": [{"name": f"{cube}.{d}"} for d in dims],
            })
        return {"cubes": cubes}

    async def load(self, query: dict[str, Any]) -> dict[str, Any]:
        self.queries.append(query)
        dims: list[str] = query.get("dimensions", [])
        measures: list[str] = query.get("measures", [])
        limit = int(query.get("limit", 10000))
        filters = query.get("filters", [])
        members = dims + measures
        cube = members[0].split(".")[0] if members else ""

        if cube == "indicateur_metadata":
            rows = [{f"indicateur_metadata.{k}": v for k, v in r.items()} for r in _indicator_rows()]
        elif cube == "indicateur_x_source_metadata":
            for d in dims:
                if d.split(".", 1)[1] not in SOURCE_DIMS:
                    raise BadRequestError(
                        f"Bad request to API. Details: {{'type': 'UserError', 'error': \"'{d.split('.', 1)[1]}' "
                        f"not found for path '{d}'\"}}"
                    )
            rows = [{
                "indicateur_x_source_metadata.id_indicateur": 611,
                "indicateur_x_source_metadata.nom_source": "fichiers_fonciers",
                "indicateur_x_source_metadata.description": "Fichiers fonciers (Cerema)",
                "indicateur_x_source_metadata.producteur_source": "Cerema",
                "indicateur_x_source_metadata.distributeur_source": "Cerema",
                "indicateur_x_source_metadata.license_source": "Licence ouverte",
                "indicateur_x_source_metadata.lien_page": "https://artificialisation.developpement-durable.gouv.fr",
                "indicateur_x_source_metadata.date_derniere_extraction": "2024-06-01",
            }]
        elif cube in DATA_CUBES and not measures and sum(d.split(".", 1)[1].startswith("geocode_") for d in dims) > 1:
            rows = self._hierarchy_rows(cube)
        elif cube in DATA_CUBES:
            rows = self._data_rows(cube, dims, measures)
        else:
            raise BadRequestError(f"Unknown cube {cube}")

        for f in filters:
            member, op, values = f["member"], f["operator"], [str(v) for v in f.get("values", [])]
            if op == "equals":
                rows = [r for r in rows if str(r.get(member)) in values]
            elif op == "contains":
                needle = values[0].lower()
                rows = [r for r in rows if needle in str(r.get(member, "")).lower()]
            elif op == "set":
                rows = [r for r in rows if r.get(member) is not None]
        for member, direction in reversed(list((query.get("order") or {}).items())):
            rows.sort(key=lambda r: str(r.get(member) or ""), reverse=direction == "desc")
        if dims and not measures:
            # Cube.js dedupes dimension-only queries
            seen, uniq = set(), []
            for r in rows:
                key = json.dumps([r.get(d) for d in dims], sort_keys=True, default=str)
                if key not in seen:
                    seen.add(key)
                    uniq.append({d: r.get(d) for d in dims})
            rows = uniq
        return {"data": rows[:limit]}

    def _hierarchy_rows(self, cube: str) -> list[dict[str, Any]]:
        """One row per commune with its EPCI, département and région (territory index)."""
        rows = []
        for code, (label, epci, dpt) in COMMUNES.items():
            rows.append({
                f"{cube}.geocode_commune": code, f"{cube}.libelle_commune": label,
                f"{cube}.geocode_epci": epci, f"{cube}.libelle_epci": EPCIS[epci],
                f"{cube}.geocode_departement": dpt,
                f"{cube}.libelle_departement": DEPARTEMENT_LABELS.get(dpt, f"Département {dpt}"),
                f"{cube}.geocode_region": _region_of(dpt),
            })
        return rows

    def _data_rows(self, cube: str, dims: list[str], measures: list[str]) -> list[dict[str, Any]]:
        levels = [lv for lv in GEO_DIMS if f"{cube}.{GEO_DIMS[lv][0]}" in dims]
        if not levels:
            # dimension `annee` only (latest-year lookup): one territory stands for all
            levels = [DATA_CUBES[cube][0][-1]]
        level = levels[0]
        real = CAPTURE["query_611_reg_93"]["data"] if (cube == "conso_enaf_com" and level == "region") else None
        axes = CUBE_AXES.get(cube, {})
        modalities: list[dict[str, str]] = [{}]
        for axis, values in axes.items():
            modalities = [{**m, axis: v} for m in modalities for v in values]
        rows = []
        for code, label in TERRITORIES[level]:
            for year in CUBE_YEARS.get(cube, YEARS):
                for modality in modalities:
                    row = {
                        f"{cube}.{GEO_DIMS[level][0]}": code,
                        f"{cube}.{GEO_DIMS[level][1]}": label,
                        f"{cube}.annee": year,
                        f"{cube}.date_mesure": f"{year}-01-01T00:00:00.000",
                        f"{cube}.date_mesure.year": f"{year}-01-01T00:00:00.000",
                        **{f"{cube}.{a}": v for a, v in modality.items()},
                    }
                    for m in measures:
                        value: Any = f"{(hash((code, year, m, tuple(modality.values()))) % 100000) / 10:.4f}"
                        if real and code == "93":
                            match = next((r for r in real if r["annee"] == year), None)
                            if match:
                                value = match["valeur"]
                        if m.endswith(".id_955") and (modality.get("mode_transport") != "Vélo (depuis 2017)" or year != "2022"):
                            value = None  # « - vélo » indicator: only the vélo modality, since 2017
                        if m.endswith(".id_611") and code == "13" and level == "departement":
                            value = None  # a territory without data at this level
                        row[m] = value
                    rows.append(row)
        return rows