"""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`, `equals` and `contains` filters. """ 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"], } # cube -> (levels, indicator ids) — from the captured available_cubes DATA_CUBES = { "conso_enaf_com": (["commune", "epci", "departement", "region"], [611, 545]), "stationnement_velo_habitant_com": (["commune", "epci", "departement", "region"], [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]), } YEARS = ["2020", "2021", "2022"] # Synthetic territories (enough to exercise truncation on communes) TERRITORIES: dict[str, list[tuple[str, str]]] = { "region": list(REGION_CODES.items()), "departement": [(f"{i:02d}", f"Département {i:02d}") for i in range(1, 96) if i != 20] + [("2A", "Corse-du-Sud"), ("2B", "Haute-Corse"), ("13", "Bouches-du-Rhône"), ("69", "Rhône")], "epci": [ ("200046977", "Métropole de Lyon"), ("200054781", "Métropole du Grand Paris"), ("200054807", "Métropole d'Aix-Marseille-Provence"), ("243100518", "Toulouse Métropole"), ] + [(f"2000{i:05d}", f"Communauté de communes n°{i}") for i in range(1, 60)], "commune": [("75056", "Paris"), ("13055", "Marseille"), ("69123", "Lyon"), ("31555", "Toulouse")] + [(f"{i:05d}", f"Commune {i}") for i in range(1000, 1700)], } TERRITORIES["departement"] = list(dict.fromkeys(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"]) 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 = ["annee"] + [d for lv in levels for d in GEO_DIMS[lv]] cubes.append({ "name": cube, "measures": [{"name": f"{cube}.id_{i}"} for i in ids], "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: 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["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()] if dims and not measures: # Cube.js dedupes dimension-only queries seen, uniq = set(), [] for r in rows: key = tuple(r.get(d) for d in dims) 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 _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: return [] level = levels[0] real = CAPTURE["query_611_reg_93"]["data"] if (cube == "conso_enaf_com" and level == "region") else None rows = [] for code, label in TERRITORIES[level]: for year in YEARS: row = { f"{cube}.{GEO_DIMS[level][0]}": code, f"{cube}.{GEO_DIMS[level][1]}": label, f"{cube}.annee": year, } for m in measures: value = f"{(hash((code, year, m)) % 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"] row[m] = value rows.append(row) return rows