| 1 | #!/usr/bin/env python3 |
| 2 | """Offline tests for scripts/catalog_models_dev.py (#4117).""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import hashlib |
| 7 | import json |
| 8 | import os |
| 9 | import subprocess |
| 10 | import sys |
| 11 | import tempfile |
| 12 | import unittest |
| 13 | from pathlib import Path |
| 14 | |
| 15 | ROOT = Path(__file__).resolve().parents[1] |
| 16 | SCRIPT = ROOT / "scripts" / "catalog_models_dev.py" |
| 17 | SEED = ROOT / "crates" / "config" / "assets" / "models_dev.bundled.json" |
| 18 | |
| 19 | |
| 20 | class CatalogModelsDevScriptTests(unittest.TestCase): |
| 21 | def test_snapshot_check_validates_offline_seed(self) -> None: |
| 22 | proc = subprocess.run( |
| 23 | [sys.executable, str(SCRIPT), "snapshot", "--check", str(SEED)], |
| 24 | cwd=ROOT, |
| 25 | capture_output=True, |
| 26 | text=True, |
| 27 | check=False, |
| 28 | ) |
| 29 | self.assertEqual(proc.returncode, 0, proc.stderr) |
| 30 | self.assertIn("ok:", proc.stdout) |
| 31 | self.assertIn("providers=", proc.stdout) |
| 32 | |
| 33 | def test_scrub_drops_api_key_fields(self) -> None: |
| 34 | # Import helpers without network. |
| 35 | sys.path.insert(0, str(ROOT / "scripts")) |
| 36 | import catalog_models_dev as mod # type: ignore |
| 37 | |
| 38 | dirty = { |
| 39 | "models": {}, |
| 40 | "providers": { |
| 41 | "deepseek": { |
| 42 | "api_key": "sk-should-never-persist", |
| 43 | "models": {"deepseek-v4-pro": {"id": "deepseek-v4-pro"}}, |
| 44 | } |
| 45 | }, |
| 46 | "token": "nope", |
| 47 | } |
| 48 | clean = mod.strip_sensitive_fields(dirty) |
| 49 | self.assertNotIn("token", clean) |
| 50 | self.assertNotIn("api_key", clean["providers"]["deepseek"]) |
| 51 | self.assertIn("models", clean["providers"]["deepseek"]) |
| 52 | |
| 53 | def test_ensure_shape_rejects_empty_object(self) -> None: |
| 54 | sys.path.insert(0, str(ROOT / "scripts")) |
| 55 | import catalog_models_dev as mod # type: ignore |
| 56 | |
| 57 | with self.assertRaises(SystemExit): |
| 58 | mod.ensure_models_dev_shape({}, "test") |
| 59 | |
| 60 | def test_public_document_drops_api_key(self) -> None: |
| 61 | sys.path.insert(0, str(ROOT / "scripts")) |
| 62 | import catalog_models_dev as mod # type: ignore |
| 63 | |
| 64 | dirty = { |
| 65 | "models": {}, |
| 66 | "providers": {"deepseek": {"api_key": "sk-x", "models": {}}}, |
| 67 | "token": "nope", |
| 68 | } |
| 69 | clean = mod.public_models_dev_document(dirty) |
| 70 | self.assertNotIn("token", clean) |
| 71 | self.assertNotIn("api_key", clean["providers"]["deepseek"]) |
| 72 | |
| 73 | def test_refresh_write_cache_is_rejected_without_writing(self) -> None: |
| 74 | with tempfile.TemporaryDirectory() as td: |
| 75 | source = Path(td) / "catalog.json" |
| 76 | target = Path(td) / "cache.json" |
| 77 | source.write_text( |
| 78 | json.dumps({"models": {}, "providers": {}, "api_key": "sk-nope"}), |
| 79 | encoding="utf-8", |
| 80 | ) |
| 81 | env = os.environ.copy() |
| 82 | env["CODEWHALE_MODELS_DEV_PATH"] = str(source) |
| 83 | |
| 84 | proc = subprocess.run( |
| 85 | [ |
| 86 | sys.executable, |
| 87 | str(SCRIPT), |
| 88 | "refresh", |
| 89 | "--write-cache", |
| 90 | str(target), |
| 91 | ], |
| 92 | cwd=ROOT, |
| 93 | capture_output=True, |
| 94 | text=True, |
| 95 | check=False, |
| 96 | env=env, |
| 97 | ) |
| 98 | |
| 99 | self.assertNotEqual(proc.returncode, 0) |
| 100 | self.assertIn("disk writes are intentionally unsupported", proc.stderr) |
| 101 | self.assertFalse(target.exists(), "refresh must remain dry-run only") |
| 102 | |
| 103 | def test_public_limit_value_never_echoes_tokens(self) -> None: |
| 104 | sys.path.insert(0, str(ROOT / "scripts")) |
| 105 | import catalog_models_dev as mod # type: ignore |
| 106 | |
| 107 | self.assertEqual(mod.public_limit_value(128000), "128000") |
| 108 | self.assertEqual(mod.public_limit_value(None), "null") |
| 109 | self.assertEqual(mod.public_limit_value("sk-this-is-a-token"), "redacted") |
| 110 | self.assertEqual(mod.public_limit_value({"authorization": "Bearer secret"}), "redacted") |
| 111 | self.assertEqual(mod.public_limit_value(True), "redacted") |
| 112 | |
| 113 | def test_public_source_label_strips_query_string(self) -> None: |
| 114 | sys.path.insert(0, str(ROOT / "scripts")) |
| 115 | import catalog_models_dev as mod # type: ignore |
| 116 | |
| 117 | self.assertEqual( |
| 118 | mod.public_source_label("url:https://models.dev/catalog.json?token=sk-leak"), |
| 119 | "url:https://models.dev/catalog.json", |
| 120 | ) |
| 121 | self.assertEqual(mod.public_source_label("file:/tmp/catalog.json"), "file:/tmp/catalog.json") |
| 122 | |
| 123 | def test_drift_command_is_gone(self) -> None: |
| 124 | proc = run_script("drift") |
| 125 | self.assertNotEqual(proc.returncode, 0) |
| 126 | |
| 127 | |
| 128 | SPEC_FIXTURE = """ |
| 129 | [source] |
| 130 | url = "https://models.dev/catalog.json" |
| 131 | |
| 132 | [meta] |
| 133 | role = "NOT a competing source of truth; live Models.dev wins." |
| 134 | |
| 135 | [[canonical]] |
| 136 | key = "demo-pro" |
| 137 | upstream = "vendor/demo-pro" |
| 138 | |
| 139 | [[providers]] |
| 140 | id = "moonshot" |
| 141 | upstream = "moonshotai" |
| 142 | name = "Moonshot" |
| 143 | env = ["MOONSHOT_API_KEY"] |
| 144 | default = "kimi-k3" |
| 145 | models = [ |
| 146 | "kimi-k3", |
| 147 | { id = "GLM-5.2", base_model = "demo-pro" }, |
| 148 | { id = "kimi-plan", from = "moonshotai-plan", upstream_id = "kimi-k3-plan" }, |
| 149 | { id = "kimi-old", curated = true }, |
| 150 | ] |
| 151 | |
| 152 | [[curated]] |
| 153 | provider = "moonshot" |
| 154 | id = "kimi-old" |
| 155 | reason = "upstream dropped it; still served" |
| 156 | |
| 157 | [curated.row] |
| 158 | name = "Kimi Old" |
| 159 | limit = { context = 1000 } |
| 160 | """ |
| 161 | |
| 162 | UPSTREAM_FIXTURE = { |
| 163 | "models": { |
| 164 | "vendor/demo-pro": { |
| 165 | "id": "vendor/demo-pro", |
| 166 | "name": "Demo Pro", |
| 167 | "limit": {"context": 1000, "output": 100}, |
| 168 | "benchmarks": [{"name": "x"}], |
| 169 | } |
| 170 | }, |
| 171 | "providers": { |
| 172 | "moonshotai": { |
| 173 | "id": "moonshotai", |
| 174 | "api_key": "sk-provider-level-secret", |
| 175 | "models": { |
| 176 | "kimi-k3": { |
| 177 | "id": "kimi-k3", |
| 178 | "limit": {"context": 1048576, "output": 131072}, |
| 179 | "cost": {"input": 3, "output": 15, "tiers": [{"input": 6}]}, |
| 180 | "modalities": {"input": ["text", "image"], "output": ["text"]}, |
| 181 | "description": "not carried", |
| 182 | "client_secret": "sk-row-secret", |
| 183 | }, |
| 184 | "glm-5.2": {"id": "glm-5.2", "reasoning": True}, |
| 185 | "kimi-new": {"id": "kimi-new"}, |
| 186 | }, |
| 187 | }, |
| 188 | "moonshotai-plan": {"id": "moonshotai-plan", "models": {"kimi-k3-plan": {"id": "kimi-k3-plan"}}}, |
| 189 | }, |
| 190 | } |
| 191 | |
| 192 | |
| 193 | def run_script(*args: str, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: |
| 194 | return subprocess.run( |
| 195 | [sys.executable, str(SCRIPT), *args], |
| 196 | cwd=ROOT, |
| 197 | capture_output=True, |
| 198 | text=True, |
| 199 | check=False, |
| 200 | env=env, |
| 201 | ) |
| 202 | |
| 203 | |
| 204 | class SeedGeneratorTests(unittest.TestCase): |
| 205 | def setUp(self) -> None: |
| 206 | self.tmp = tempfile.TemporaryDirectory() |
| 207 | self.dir = Path(self.tmp.name) |
| 208 | self.spec = self.dir / "spec.toml" |
| 209 | self.lock = self.dir / "lock.json" |
| 210 | self.out = self.dir / "seed.json" |
| 211 | self.upstream = self.dir / "upstream.json" |
| 212 | self.corrections = self.dir / "corrections.json" |
| 213 | self.spec.write_text(SPEC_FIXTURE, encoding="utf-8") |
| 214 | self.upstream.write_text(json.dumps(UPSTREAM_FIXTURE), encoding="utf-8") |
| 215 | self.corrections.write_text( |
| 216 | json.dumps( |
| 217 | { |
| 218 | "revision": "t", |
| 219 | "reviewed": {"revision": "fixture"}, |
| 220 | "models": [{"provider": "moonshot", "id": "kimi-k3", "max_output": 131072, "reason": "r"}], |
| 221 | } |
| 222 | ), |
| 223 | encoding="utf-8", |
| 224 | ) |
| 225 | |
| 226 | def tearDown(self) -> None: |
| 227 | self.tmp.cleanup() |
| 228 | |
| 229 | def env(self) -> dict[str, str]: |
| 230 | env = os.environ.copy() |
| 231 | env["CODEWHALE_MODELS_DEV_PATH"] = str(self.upstream) |
| 232 | return env |
| 233 | |
| 234 | def lock_cmd(self, *extra: str) -> subprocess.CompletedProcess[str]: |
| 235 | return run_script( |
| 236 | "seed", "lock", "--spec", str(self.spec), "--lock", str(self.lock), |
| 237 | "--corrections", str(self.corrections), *extra, env=self.env(), |
| 238 | ) |
| 239 | |
| 240 | def render_cmd(self, *extra: str) -> subprocess.CompletedProcess[str]: |
| 241 | return run_script( |
| 242 | "seed", "render", "--spec", str(self.spec), "--lock", str(self.lock), |
| 243 | "--out", str(self.out), "--corrections", str(self.corrections), *extra, |
| 244 | ) |
| 245 | |
| 246 | def test_dry_run_writes_nothing(self) -> None: |
| 247 | proc = self.lock_cmd("--dry-run") |
| 248 | self.assertEqual(proc.returncode, 0, proc.stderr) |
| 249 | self.assertFalse(self.lock.exists()) |
| 250 | self.assertIn("dry-run", proc.stdout) |
| 251 | |
| 252 | def test_lock_keeps_allowlisted_fields_scrubs_secrets_and_pins_the_document(self) -> None: |
| 253 | proc = self.lock_cmd() |
| 254 | self.assertEqual(proc.returncode, 0, proc.stderr) |
| 255 | text = self.lock.read_text(encoding="utf-8") |
| 256 | self.assertNotIn("sk-", text) |
| 257 | self.assertNotIn("description", text) |
| 258 | self.assertNotIn("tiers", text) |
| 259 | self.assertNotIn("benchmarks", text) |
| 260 | lock = json.loads(text) |
| 261 | expected = hashlib.sha256(self.upstream.read_bytes()).hexdigest() |
| 262 | self.assertEqual(lock["source"]["sha256"], expected) |
| 263 | rows = lock["providers"]["moonshotai"] |
| 264 | # Case-insensitive match keeps upstream's id in the lock. |
| 265 | self.assertIn("glm-5.2", rows) |
| 266 | self.assertNotIn("kimi-new", rows, "only referenced rows are pinned") |
| 267 | self.assertIn("kimi-k3-plan", lock["providers"]["moonshotai-plan"]) |
| 268 | self.assertIn("kimi-new", proc.stdout, "new upstream models are reported") |
| 269 | # A correction whose value upstream now states is reported as stale. |
| 270 | self.assertIn("moonshot/kimi-k3: max_output 131072 equals upstream", proc.stdout) |
| 271 | |
| 272 | def test_render_is_deterministic_and_maps_ids(self) -> None: |
| 273 | self.assertEqual(self.lock_cmd().returncode, 0) |
| 274 | self.assertEqual(self.render_cmd().returncode, 0) |
| 275 | first = self.out.read_bytes() |
| 276 | self.assertEqual(self.render_cmd().returncode, 0) |
| 277 | self.assertEqual(first, self.out.read_bytes()) |
| 278 | seed = json.loads(first) |
| 279 | models = seed["providers"]["moonshot"]["models"] |
| 280 | self.assertEqual(list(models), ["kimi-k3", "GLM-5.2", "kimi-plan", "kimi-old"]) |
| 281 | self.assertEqual(models["GLM-5.2"]["id"], "GLM-5.2", "Codewhale wire id is kept") |
| 282 | self.assertEqual(models["GLM-5.2"]["base_model"], "demo-pro") |
| 283 | self.assertTrue(models["kimi-k3"]["default"]) |
| 284 | self.assertNotIn("default", models["GLM-5.2"]) |
| 285 | self.assertEqual(models["kimi-plan"]["id"], "kimi-plan") |
| 286 | self.assertEqual(models["kimi-old"]["limit"], {"context": 1000}) |
| 287 | self.assertEqual(seed["models"]["demo-pro"]["id"], "demo-pro") |
| 288 | self.assertIn("1 canonical", seed["_meta"]["coverage"]) |
| 289 | self.assertIn(seed["_meta"]["role"], SPEC_FIXTURE) |
| 290 | |
| 291 | def test_check_fails_with_a_diff_after_a_hand_edit(self) -> None: |
| 292 | self.assertEqual(self.lock_cmd().returncode, 0) |
| 293 | self.assertEqual(self.render_cmd().returncode, 0) |
| 294 | self.assertEqual(self.render_cmd("--check").returncode, 0) |
| 295 | edited = self.out.read_text(encoding="utf-8").replace("131072", "131073", 1) |
| 296 | self.out.write_text(edited, encoding="utf-8") |
| 297 | proc = self.render_cmd("--check") |
| 298 | self.assertNotEqual(proc.returncode, 0) |
| 299 | self.assertIn("131073", proc.stdout) |
| 300 | self.assertIn("seed render", proc.stderr) |
| 301 | |
| 302 | def test_lock_refuses_missing_rows_and_curated_rows_upstream_now_lists(self) -> None: |
| 303 | upstream = json.loads(json.dumps(UPSTREAM_FIXTURE)) |
| 304 | del upstream["providers"]["moonshotai"]["models"]["glm-5.2"] |
| 305 | upstream["providers"]["moonshotai"]["models"]["kimi-old"] = {"id": "kimi-old"} |
| 306 | self.upstream.write_text(json.dumps(upstream), encoding="utf-8") |
| 307 | proc = self.lock_cmd() |
| 308 | self.assertNotEqual(proc.returncode, 0) |
| 309 | self.assertIn("does not list GLM-5.2", proc.stderr) |
| 310 | self.assertIn("kimi-old: curated, but upstream", proc.stderr) |
| 311 | self.assertFalse(self.lock.exists()) |
| 312 | |
| 313 | def test_render_refuses_a_curated_row_present_in_the_lock(self) -> None: |
| 314 | self.assertEqual(self.lock_cmd().returncode, 0) |
| 315 | lock = json.loads(self.lock.read_text(encoding="utf-8")) |
| 316 | lock["providers"]["moonshotai"]["kimi-old"] = {"id": "kimi-old"} |
| 317 | self.lock.write_text(json.dumps(lock), encoding="utf-8") |
| 318 | proc = self.render_cmd() |
| 319 | self.assertNotEqual(proc.returncode, 0) |
| 320 | self.assertIn("curated row also present in the lock", proc.stderr) |
| 321 | |
| 322 | def test_spec_refuses_value_overrides_and_bad_defaults(self) -> None: |
| 323 | for broken, message in [ |
| 324 | (SPEC_FIXTURE.replace('"kimi-k3",', '{ id = "kimi-k3", limit = 5 },', 1), "never restates"), |
| 325 | (SPEC_FIXTURE.replace('default = "kimi-k3"', 'default = "nope"'), "exactly one default"), |
| 326 | (SPEC_FIXTURE.replace("curated = true", "base_model = \"x\""), "not marked curated"), |
| 327 | ]: |
| 328 | self.spec.write_text(broken, encoding="utf-8") |
| 329 | proc = self.render_cmd() |
| 330 | self.assertNotEqual(proc.returncode, 0) |
| 331 | self.assertIn(message, proc.stderr) |
| 332 | |
| 333 | |
| 334 | class CommittedSeedTests(unittest.TestCase): |
| 335 | def test_committed_seed_is_the_rendered_seed(self) -> None: |
| 336 | proc = run_script("seed", "render", "--check") |
| 337 | self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) |
| 338 | |
| 339 | def test_reviewed_source_is_exactly_embedded_without_mutating_lock(self) -> None: |
| 340 | data = json.loads(SEED.read_text()) |
| 341 | source = json.loads((ROOT / "crates/config/assets/catalog_corrections.json").read_text()) |
| 342 | self.assertEqual(data["_reviewed"], source["reviewed"]) |
| 343 | self.assertIn("fetched 2026-09-26", data["_meta"]["upstream"]) |
| 344 | self.assertEqual(len(data["_reviewed"]["public_models"]), 78) |
| 345 | |
| 346 | def test_reviewed_malformed_and_duplicate_public_facts_are_refused(self) -> None: |
| 347 | sys.path.insert(0, str(ROOT / "scripts")) |
| 348 | import catalog_models_dev as mod |
| 349 | data = json.loads((ROOT / "crates/config/assets/catalog_corrections.json").read_text())["reviewed"] |
| 350 | for mutate in [ |
| 351 | lambda row: row.pop("revision"), |
| 352 | lambda row: row["intrinsic"]["kimi-k3"].update(context_window=0), |
| 353 | lambda row: row["intrinsic"]["kimi-k3"].update(max_output=True), |
| 354 | lambda row: row["public_models"].append(row["public_models"][0]), |
| 355 | lambda row: row["compatibility_aliases"].update(hostile={"name": "bad\x1bvalue"}), |
| 356 | lambda row: row["numeric_refs"]["KIMI_K3_CONTEXT_WINDOW_TOKENS"].update(model="missing-private-name"), |
| 357 | ]: |
| 358 | fixture = json.loads(json.dumps(data)) |
| 359 | mutate(fixture) |
| 360 | with self.assertRaises(SystemExit): |
| 361 | mod.validate_reviewed(fixture) |
| 362 | |
| 363 | |
| 364 | if __name__ == "__main__": |
| 365 | unittest.main() |
| 366 |