Spaces:
Running
Running
File size: 8,829 Bytes
febc39b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | """Offline tests of the MCP tools against the fake Cube.js backend.
Run: python -m unittest discover -s tests -v
"""
from __future__ import annotations
import json
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import src.api_client as api_client # noqa: E402
import src.cache as cache_mod # noqa: E402
import src.cube_resolver as resolver_mod # noqa: E402
from src.tools import ( # noqa: E402
find_territory,
get_indicator_details,
list_indicators,
query_indicator_data,
search_indicators,
)
from tests.fake_api import FakeCubeJsClient # noqa: E402
class ToolsTestCase(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
# Fresh singletons bound to the fake backend
api_client._client_instance = FakeCubeJsClient()
resolver_mod._resolver_instance = None
cache_mod._cache_instance = None
@property
def fake(self) -> FakeCubeJsClient:
return api_client._client_instance # type: ignore[return-value]
# ------------------------------------------------------------------ format
async def test_outputs_are_compact_json(self) -> None:
for raw in (
await list_indicators(),
await search_indicators("espace"),
await get_indicator_details("611"),
await query_indicator_data("611", "region", "93"),
):
self.assertNotIn("\n", raw)
self.assertIsInstance(json.loads(raw), dict)
# ------------------------------------------------------------------ list
async def test_list_all(self) -> None:
payload = json.loads(await list_indicators())
self.assertEqual(payload["count"], 168)
self.assertEqual(len(payload["indicators"]), 168)
self.assertNotIn("truncated", payload)
item = next(i for i in payload["indicators"] if i["id"] == 611)
self.assertEqual(item["mailles"], ["commune", "epci", "departement", "region"])
self.assertIn("Mieux se déplacer", payload["themes"])
async def test_list_filters_accent_insensitive(self) -> None:
a = json.loads(await list_indicators(thematique="deplacer"))
b = json.loads(await list_indicators(thematique="Déplacer"))
self.assertEqual(a["count"], b["count"])
self.assertEqual(a["count"], 58)
c = json.loads(await list_indicators(thematique="déplacer", maille="commune"))
self.assertLess(c["count"], a["count"])
self.assertTrue(all("commune" in i["mailles"] for i in c["indicators"]))
async def test_list_invalid_maille(self) -> None:
payload = json.loads(await list_indicators(maille="pays"))
self.assertIn("error", payload)
self.assertIn("valid_levels", payload)
# ------------------------------------------------------------------ search
async def test_search_accents_and_plural(self) -> None:
with_acc = json.loads(await search_indicators("émissions gaz effet de serre"))
without = json.loads(await search_indicators("emission gaz effet serre"))
self.assertTrue(with_acc["indicators"])
self.assertEqual(
[i["id"] for i in with_acc["indicators"]], [i["id"] for i in without["indicators"]]
)
async def test_search_libelle_first(self) -> None:
payload = json.loads(await search_indicators("consommation espace"))
ids = [i["id"] for i in payload["indicators"]]
self.assertIn(611, ids[:2])
self.assertIn(545, ids[:2])
async def test_search_capped(self) -> None:
payload = json.loads(await search_indicators("de"))
self.assertLessEqual(len(payload["indicators"]), 30)
self.assertTrue(payload["truncated"])
self.assertGreater(payload["total_count"], 30)
async def test_search_no_match(self) -> None:
payload = json.loads(await search_indicators("blockchain"))
self.assertEqual(payload["total_count"], 0)
self.assertIn("hint", payload)
# ------------------------------------------------------------------ details
async def test_details_sources_use_existing_dimensions(self) -> None:
payload = json.loads(await get_indicator_details("611"))
self.assertNotIn("sources_warning", payload)
self.assertEqual(payload["sources"][0]["producteur_source"], "Cerema")
self.assertNotIn("id_indicateur", payload["sources"][0])
self.assertEqual(payload["metadata"]["annees_disponibles"][0], "2009")
self.assertEqual(payload["available_cubes"]["commune"], "conso_enaf_com")
self.assertAlmostEqual(payload["metadata"]["completion"]["region"], 0.8947)
# the source query never asked for the non-existent `libelle` dimension
src_queries = [q for q in self.fake.queries if q["dimensions"][0].startswith("indicateur_x_source")]
self.assertTrue(src_queries)
self.assertNotIn("indicateur_x_source_metadata.libelle", src_queries[0]["dimensions"])
async def test_details_unknown_and_invalid(self) -> None:
self.assertIn("error", json.loads(await get_indicator_details("999999")))
self.assertIn("error", json.loads(await get_indicator_details("abc")))
# ------------------------------------------------------------------ query
async def test_query_one_territory(self) -> None:
payload = json.loads(await query_indicator_data("611", "region", "93"))
self.assertEqual(payload["unite"], "ha")
self.assertFalse(payload["truncated"])
self.assertEqual(payload["total_count"], 3)
row = payload["data"][0]
self.assertEqual(row["geocode"], "93")
self.assertIsInstance(row["valeur"], float) # converted from string
self.assertEqual(payload["annees_disponibles"][-1], "2023")
async def test_query_truncation_and_limit(self) -> None:
payload = json.loads(await query_indicator_data("611", "commune", year="2021"))
self.assertEqual(payload["total_count"], 100)
self.assertTrue(payload["truncated"])
self.assertIn("hint", payload)
big = json.loads(await query_indicator_data("611", "commune", year="2021", limit=500))
self.assertEqual(big["total_count"], 500)
self.assertTrue(big["truncated"])
small = json.loads(await query_indicator_data("611", "region", year="2021", limit=5))
self.assertEqual(small["total_count"], 5)
self.assertTrue(small["truncated"])
# limit is clamped, never rejected
clamped = json.loads(await query_indicator_data("611", "region", year="2021", limit=9999))
self.assertEqual(clamped["total_count"], 13)
self.assertFalse(clamped["truncated"])
async def test_query_wrong_level_and_hints(self) -> None:
payload = json.loads(await query_indicator_data("42", "commune"))
self.assertIn("error", payload)
self.assertEqual(payload["available_levels"], ["departement", "region"])
empty = json.loads(await query_indicator_data("611", "region", "93", year="1999"))
self.assertEqual(empty["total_count"], 0)
self.assertIn("1999 is not available", empty["hint"])
bad_code = json.loads(await query_indicator_data("611", "departement", "Rhône"))
self.assertIn("find_territory", bad_code["hint"])
# ------------------------------------------------------------------ territory
async def test_find_territory_region_local(self) -> None:
payload = json.loads(await find_territory("provence", "region"))
self.assertEqual(payload["territories"][0]["geocode"], "93")
self.assertFalse(self.fake.queries) # regions resolved without the API
async def test_find_territory_epci_and_commune(self) -> None:
payload = json.loads(await find_territory("Lyon"))
levels = {t["geographic_level"]: t for t in payload["territories"]}
self.assertEqual(levels["epci"]["geocode"], "200046977")
self.assertEqual(levels["commune"]["geocode"], "69123")
self.assertNotIn("departement", levels)
dpt = json.loads(await find_territory("rhône", "departement"))
self.assertEqual({t["geocode"] for t in dpt["territories"]}, {"13", "69"})
async def test_find_territory_invalid(self) -> None:
self.assertIn("error", json.loads(await find_territory("", "epci")))
self.assertIn("error", json.loads(await find_territory("Lyon", "pays")))
# ------------------------------------------------------------------ resilience
async def test_api_down_before_first_load(self) -> None:
self.fake.fail_meta = True
payload = json.loads(await search_indicators("espace"))
self.assertIn("error", payload)
self.assertIn("hint", payload)
if __name__ == "__main__":
unittest.main()
|