| 1 | """GitGoodBench importer tests against synthetic CSV/JSONL exports (no network).
|
| 2 |
|
| 3 | Covers the REAL published schema (id, name, default_branch, license,
|
| 4 | stargazers, created_at, topics, programming_language, scenario, sample_type,
|
| 5 | project_size, difficulty), the file_commit_chain two-scenario expansion, the
|
| 6 | legacy liberal fallback, and exit-code behavior.
|
| 7 | """
|
| 8 |
|
| 9 | from __future__ import annotations
|
| 10 |
|
| 11 | import csv
|
| 12 | import json
|
| 13 | import subprocess
|
| 14 | import sys
|
| 15 | import tempfile
|
| 16 | import unittest
|
| 17 | from pathlib import Path
|
| 18 |
|
| 19 | SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "import_gitgoodbench.py"
|
| 20 |
|
| 21 | REAL_COLUMNS = [
|
| 22 | "id",
|
| 23 | "name",
|
| 24 | "default_branch",
|
| 25 | "license",
|
| 26 | "stargazers",
|
| 27 | "created_at",
|
| 28 | "topics",
|
| 29 | "programming_language",
|
| 30 | "scenario",
|
| 31 | "sample_type",
|
| 32 | "project_size",
|
| 33 | "difficulty",
|
| 34 | ]
|
| 35 |
|
| 36 | CITATION = "lindenbauer-etal-2025-gitgoodbench (REALM 2025, doi:10.18653/v1/2025.realm-1.19)"
|
| 37 | SOURCE_DATASETS = [
|
| 38 | "JetBrains/git_good_bench",
|
| 39 | "JetBrains/git_good_bench-lite",
|
| 40 | "JetBrains/git_good_bench-train",
|
| 41 | ]
|
| 42 |
|
| 43 | MERGE_SCENARIO = {
|
| 44 | "merge_commit_hash": "abc123",
|
| 45 | "parents": ["p1", "p2"],
|
| 46 | "files": ["src/a.py", "src/b.py"],
|
| 47 | }
|
| 48 | CHAIN_SCENARIO = {
|
| 49 | "first_commit": "1111111",
|
| 50 | "last_commit": "2222222",
|
| 51 | "file": "src/x.py",
|
| 52 | "times_file_changed": 5,
|
| 53 | }
|
| 54 |
|
| 55 |
|
| 56 | def run_importer(*args: str) -> subprocess.CompletedProcess:
|
| 57 | return subprocess.run(
|
| 58 | [sys.executable, str(SCRIPT), *args],
|
| 59 | capture_output=True,
|
| 60 | text=True,
|
| 61 | check=False,
|
| 62 | )
|
| 63 |
|
| 64 |
|
| 65 | def write_real_csv(path: Path, rows: list[dict]) -> None:
|
| 66 | with path.open("w", newline="", encoding="utf-8") as fh:
|
| 67 | writer = csv.DictWriter(fh, fieldnames=REAL_COLUMNS)
|
| 68 | writer.writeheader()
|
| 69 | for row in rows:
|
| 70 | writer.writerow(row)
|
| 71 |
|
| 72 |
|
| 73 | def real_row(**overrides) -> dict:
|
| 74 | row = {
|
| 75 | "id": "sample-1",
|
| 76 | "name": "octo/widgets",
|
| 77 | "default_branch": "main",
|
| 78 | "license": "MIT",
|
| 79 | "stargazers": "42",
|
| 80 | "created_at": "2020-01-01T00:00:00Z",
|
| 81 | "topics": "git;tools",
|
| 82 | "programming_language": "Python",
|
| 83 | "scenario": json.dumps(MERGE_SCENARIO),
|
| 84 | "sample_type": "merge",
|
| 85 | "project_size": "small",
|
| 86 | "difficulty": "easy",
|
| 87 | }
|
| 88 | row.update(overrides)
|
| 89 | return row
|
| 90 |
|
| 91 |
|
| 92 | class RealSchemaImportTest(unittest.TestCase):
|
| 93 | def test_real_schema_csv_merge_chain_unparseable_and_mirror_missing(self) -> None:
|
| 94 | with tempfile.TemporaryDirectory() as tmp:
|
| 95 | root = Path(tmp)
|
| 96 | mirror_root = root / "mirrors"
|
| 97 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 98 | csv_path = root / "ggb.csv"
|
| 99 | write_real_csv(
|
| 100 | csv_path,
|
| 101 | [
|
| 102 | real_row(id="m1", sample_type="merge"),
|
| 103 | real_row(
|
| 104 | id="c1",
|
| 105 | sample_type="file_commit_chain",
|
| 106 | scenario=json.dumps(CHAIN_SCENARIO),
|
| 107 | difficulty="medium",
|
| 108 | ),
|
| 109 | real_row(id="u1", sample_type="merge", scenario="{not valid json"),
|
| 110 | real_row(id="m2", sample_type="merge", name="acme/missing"),
|
| 111 | ],
|
| 112 | )
|
| 113 | out_path = root / "scenarios.json"
|
| 114 | proc = run_importer(
|
| 115 | "--csv", str(csv_path),
|
| 116 | "--mirror-root", str(mirror_root),
|
| 117 | "--out", str(out_path),
|
| 118 | )
|
| 119 |
|
| 120 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 121 | document = json.loads(out_path.read_text())
|
| 122 |
|
| 123 | self.assertEqual(document["source"], "gitgoodbench")
|
| 124 | self.assertEqual(document["license"], "Apache-2.0")
|
| 125 | self.assertEqual(document["citation"], CITATION)
|
| 126 | self.assertEqual(document["source_datasets"], SOURCE_DATASETS)
|
| 127 | # m1 + c1--rebase + c1--commits ready; u1 + m2 skipped.
|
| 128 | self.assertEqual(document["counts"], {"ready": 3, "skipped": 2})
|
| 129 |
|
| 130 | by_id = {item["scenario_id"]: item for item in document["scenarios"]}
|
| 131 | self.assertEqual(
|
| 132 | set(by_id),
|
| 133 | {"m1", "c1--rebase", "c1--commits", "u1", "m2"},
|
| 134 | )
|
| 135 |
|
| 136 | merge = by_id["m1"]
|
| 137 | self.assertEqual(merge["status"], "ready")
|
| 138 | self.assertEqual(merge["scenario_type"], "merge_conflict_resolution")
|
| 139 | self.assertEqual(merge["sample_type"], "merge")
|
| 140 | self.assertEqual(merge["repo"], "octo/widgets")
|
| 141 | self.assertEqual(merge["commit"], "abc123") # liberal merge_commit_hash pick
|
| 142 | self.assertEqual(merge["source_scenario"], MERGE_SCENARIO) # verbatim
|
| 143 | self.assertEqual(merge["oracle"], {"mode": "strict-em", "normalization": None})
|
| 144 | self.assertIsNone(merge["oracle_note"])
|
| 145 | self.assertEqual(
|
| 146 | merge["metadata"],
|
| 147 | {"difficulty": "easy", "programming_language": "Python", "default_branch": "main"},
|
| 148 | )
|
| 149 | self.assertTrue(merge["mirror_path"].endswith("octo__widgets"))
|
| 150 |
|
| 151 | rebase = by_id["c1--rebase"]
|
| 152 | commits = by_id["c1--commits"]
|
| 153 | for chain in (rebase, commits):
|
| 154 | self.assertEqual(chain["status"], "ready")
|
| 155 | self.assertEqual(chain["sample_type"], "file_commit_chain")
|
| 156 | self.assertEqual(chain["source_scenario"], CHAIN_SCENARIO)
|
| 157 | self.assertIsNone(chain["oracle"])
|
| 158 | self.assertEqual(chain["oracle_note"], "oracle_defined_at_runtime:tree_equivalence")
|
| 159 | self.assertEqual(chain["metadata"]["difficulty"], "medium")
|
| 160 | self.assertEqual(rebase["scenario_type"], "interactive_rebase")
|
| 161 | self.assertEqual(commits["scenario_type"], "iterative_commit")
|
| 162 |
|
| 163 | unparseable = by_id["u1"]
|
| 164 | self.assertEqual(unparseable["status"], "skipped")
|
| 165 | self.assertEqual(unparseable["skip_reason"], "scenario_json_unparseable")
|
| 166 | self.assertIsNone(unparseable["scenario_type"])
|
| 167 | self.assertEqual(unparseable["sample_type"], "merge")
|
| 168 | self.assertIsNone(unparseable["source_scenario"])
|
| 169 |
|
| 170 | missing = by_id["m2"]
|
| 171 | self.assertEqual(missing["status"], "skipped")
|
| 172 | self.assertEqual(missing["skip_reason"], "mirror_missing:acme/missing")
|
| 173 | self.assertIsNone(missing["mirror_path"])
|
| 174 | self.assertIsNone(missing["oracle"])
|
| 175 | # Source scenario is still preserved verbatim on honest skips.
|
| 176 | self.assertEqual(missing["source_scenario"], MERGE_SCENARIO)
|
| 177 |
|
| 178 | def test_chain_expansion_is_exactly_two_scenarios(self) -> None:
|
| 179 | with tempfile.TemporaryDirectory() as tmp:
|
| 180 | root = Path(tmp)
|
| 181 | mirror_root = root / "mirrors"
|
| 182 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 183 | csv_path = root / "ggb.csv"
|
| 184 | write_real_csv(
|
| 185 | csv_path,
|
| 186 | [real_row(id="c9", sample_type="file_commit_chain", scenario=json.dumps(CHAIN_SCENARIO))],
|
| 187 | )
|
| 188 | out_path = root / "scenarios.json"
|
| 189 | proc = run_importer(
|
| 190 | "--csv", str(csv_path),
|
| 191 | "--mirror-root", str(mirror_root),
|
| 192 | "--out", str(out_path),
|
| 193 | )
|
| 194 |
|
| 195 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 196 | document = json.loads(out_path.read_text())
|
| 197 |
|
| 198 | self.assertEqual(len(document["scenarios"]), 2)
|
| 199 | ids = [item["scenario_id"] for item in document["scenarios"]]
|
| 200 | self.assertEqual(ids, ["c9--rebase", "c9--commits"])
|
| 201 |
|
| 202 | def test_raw_huggingface_csv_python_literal_scenario_imports(self) -> None:
|
| 203 | with tempfile.TemporaryDirectory() as tmp:
|
| 204 | root = Path(tmp)
|
| 205 | mirror_root = root / "mirrors"
|
| 206 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 207 | csv_path = root / "ggb.csv"
|
| 208 | with csv_path.open("w", newline="", encoding="utf-8") as fh:
|
| 209 | writer = csv.DictWriter(fh, fieldnames=["", *REAL_COLUMNS])
|
| 210 | writer.writeheader()
|
| 211 | row = real_row(id="hf1", scenario=repr(MERGE_SCENARIO))
|
| 212 | writer.writerow({"": "899", **row})
|
| 213 | out_path = root / "scenarios.json"
|
| 214 | proc = run_importer(
|
| 215 | "--csv", str(csv_path),
|
| 216 | "--mirror-root", str(mirror_root),
|
| 217 | "--out", str(out_path),
|
| 218 | )
|
| 219 |
|
| 220 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 221 | document = json.loads(out_path.read_text())
|
| 222 |
|
| 223 | self.assertEqual(document["counts"], {"ready": 1, "skipped": 0})
|
| 224 | scenario = document["scenarios"][0]
|
| 225 | self.assertEqual(scenario["scenario_id"], "hf1")
|
| 226 | self.assertEqual(scenario["status"], "ready")
|
| 227 | self.assertEqual(scenario["source_scenario"], MERGE_SCENARIO)
|
| 228 | self.assertEqual(scenario["commit"], "abc123")
|
| 229 |
|
| 230 | def test_scalar_scenario_values_skip_on_both_parse_paths(self) -> None:
|
| 231 | # A scalar parses as valid JSON ("123") or a valid Python literal
|
| 232 | # ("123,") but carries no scenario data: both must skip, not go ready.
|
| 233 | with tempfile.TemporaryDirectory() as tmp:
|
| 234 | root = Path(tmp)
|
| 235 | mirror_root = root / "mirrors"
|
| 236 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 237 | csv_path = root / "ggb.csv"
|
| 238 | write_real_csv(
|
| 239 | csv_path,
|
| 240 | [
|
| 241 | real_row(id="j1", sample_type="merge", scenario="123"),
|
| 242 | real_row(id="p1", sample_type="merge", scenario="'just a string'"),
|
| 243 | ],
|
| 244 | )
|
| 245 | out_path = root / "scenarios.json"
|
| 246 | proc = run_importer(
|
| 247 | "--csv", str(csv_path),
|
| 248 | "--mirror-root", str(mirror_root),
|
| 249 | "--out", str(out_path),
|
| 250 | )
|
| 251 |
|
| 252 | self.assertEqual(proc.returncode, 3, proc.stderr)
|
| 253 | document = json.loads(out_path.read_text())
|
| 254 |
|
| 255 | self.assertEqual(document["counts"], {"ready": 0, "skipped": 2})
|
| 256 | for scenario in document["scenarios"]:
|
| 257 | self.assertEqual(scenario["status"], "skipped")
|
| 258 | self.assertEqual(scenario["skip_reason"], "scenario_json_unparseable")
|
| 259 |
|
| 260 | def test_jsonl_input_variant(self) -> None:
|
| 261 | with tempfile.TemporaryDirectory() as tmp:
|
| 262 | root = Path(tmp)
|
| 263 | mirror_root = root / "mirrors"
|
| 264 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 265 | jsonl_path = root / "ggb.jsonl"
|
| 266 | rows = [
|
| 267 | real_row(id="j1", sample_type="merge"),
|
| 268 | real_row(id="j2", sample_type="file_commit_chain", scenario=json.dumps(CHAIN_SCENARIO)),
|
| 269 | ]
|
| 270 | jsonl_path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
|
| 271 | out_path = root / "scenarios.json"
|
| 272 | proc = run_importer(
|
| 273 | "--csv", str(jsonl_path),
|
| 274 | "--mirror-root", str(mirror_root),
|
| 275 | "--out", str(out_path),
|
| 276 | )
|
| 277 |
|
| 278 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 279 | document = json.loads(out_path.read_text())
|
| 280 |
|
| 281 | self.assertEqual(document["counts"], {"ready": 3, "skipped": 0})
|
| 282 | by_id = {item["scenario_id"]: item for item in document["scenarios"]}
|
| 283 | self.assertEqual(set(by_id), {"j1", "j2--rebase", "j2--commits"})
|
| 284 | self.assertEqual(by_id["j1"]["scenario_type"], "merge_conflict_resolution")
|
| 285 | self.assertEqual(by_id["j1"]["commit"], "abc123")
|
| 286 | self.assertEqual(by_id["j2--rebase"]["scenario_type"], "interactive_rebase")
|
| 287 | self.assertEqual(by_id["j2--commits"]["scenario_type"], "iterative_commit")
|
| 288 |
|
| 289 | def test_missing_id_falls_back_to_row_index(self) -> None:
|
| 290 | with tempfile.TemporaryDirectory() as tmp:
|
| 291 | root = Path(tmp)
|
| 292 | mirror_root = root / "mirrors"
|
| 293 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 294 | csv_path = root / "ggb.csv"
|
| 295 | write_real_csv(csv_path, [real_row(id="")])
|
| 296 | out_path = root / "scenarios.json"
|
| 297 | proc = run_importer(
|
| 298 | "--csv", str(csv_path),
|
| 299 | "--mirror-root", str(mirror_root),
|
| 300 | "--out", str(out_path),
|
| 301 | )
|
| 302 |
|
| 303 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 304 | document = json.loads(out_path.read_text())
|
| 305 |
|
| 306 | self.assertEqual(document["scenarios"][0]["scenario_id"], "gitgoodbench-row-0")
|
| 307 |
|
| 308 |
|
| 309 | class LegacyFallbackImportTest(unittest.TestCase):
|
| 310 | def test_legacy_csv_mixed_rows_still_import(self) -> None:
|
| 311 | with tempfile.TemporaryDirectory() as tmp:
|
| 312 | root = Path(tmp)
|
| 313 | mirror_root = root / "mirrors"
|
| 314 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 315 | csv_path = root / "lite.csv"
|
| 316 | csv_path.write_text(
|
| 317 | "id,repo,scenario_type,commit\n"
|
| 318 | "s1,octo/widgets,merge_conflict,abc123\n"
|
| 319 | "s2,acme/missing,interactive_rebase,def456\n"
|
| 320 | "s3,acme/also-missing,iterative_commit,fed789\n"
|
| 321 | "s4,octo/widgets,bogus_type,aaa000\n"
|
| 322 | )
|
| 323 | out_path = root / "scenarios.json"
|
| 324 | proc = run_importer(
|
| 325 | "--csv", str(csv_path),
|
| 326 | "--mirror-root", str(mirror_root),
|
| 327 | "--out", str(out_path),
|
| 328 | )
|
| 329 |
|
| 330 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 331 | document = json.loads(out_path.read_text())
|
| 332 |
|
| 333 | self.assertEqual(document["source"], "gitgoodbench")
|
| 334 | self.assertEqual(document["license"], "Apache-2.0")
|
| 335 | self.assertEqual(document["citation"], CITATION)
|
| 336 | self.assertEqual(document["source_datasets"], SOURCE_DATASETS)
|
| 337 | self.assertEqual(document["oracle_mode"], "strict-em")
|
| 338 | self.assertEqual(document["counts"], {"ready": 1, "skipped": 3})
|
| 339 |
|
| 340 | by_id = {item["scenario_id"]: item for item in document["scenarios"]}
|
| 341 | self.assertEqual(len(by_id), 4)
|
| 342 |
|
| 343 | ready = by_id["s1"]
|
| 344 | self.assertEqual(ready["status"], "ready")
|
| 345 | self.assertEqual(ready["scenario_type"], "merge_conflict_resolution")
|
| 346 | self.assertEqual(ready["repo"], "octo/widgets")
|
| 347 | self.assertEqual(ready["commit"], "abc123")
|
| 348 | self.assertIsNone(ready["skip_reason"])
|
| 349 | self.assertIsNone(ready["sample_type"])
|
| 350 | self.assertTrue(ready["mirror_path"].endswith("octo__widgets"))
|
| 351 | self.assertEqual(ready["oracle"], {"mode": "strict-em", "normalization": None})
|
| 352 |
|
| 353 | rebase = by_id["s2"]
|
| 354 | self.assertEqual(rebase["status"], "skipped")
|
| 355 | self.assertEqual(rebase["scenario_type"], "interactive_rebase")
|
| 356 | self.assertEqual(rebase["skip_reason"], "mirror_missing:acme/missing")
|
| 357 | self.assertIsNone(rebase["mirror_path"])
|
| 358 | self.assertIsNone(rebase["oracle"])
|
| 359 |
|
| 360 | iterative = by_id["s3"]
|
| 361 | self.assertEqual(iterative["status"], "skipped")
|
| 362 | self.assertEqual(iterative["scenario_type"], "iterative_commit")
|
| 363 | self.assertEqual(iterative["skip_reason"], "mirror_missing:acme/also-missing")
|
| 364 |
|
| 365 | bogus = by_id["s4"]
|
| 366 | self.assertEqual(bogus["status"], "skipped")
|
| 367 | self.assertIsNone(bogus["scenario_type"])
|
| 368 | self.assertEqual(bogus["skip_reason"], "unrecognized_scenario_type:bogus_type")
|
| 369 |
|
| 370 | def test_legacy_ready_rebase_gets_runtime_oracle_note(self) -> None:
|
| 371 | with tempfile.TemporaryDirectory() as tmp:
|
| 372 | root = Path(tmp)
|
| 373 | mirror_root = root / "mirrors"
|
| 374 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 375 | csv_path = root / "lite.csv"
|
| 376 | csv_path.write_text(
|
| 377 | "id,repo,scenario_type,commit\n"
|
| 378 | "r1,octo/widgets,interactive_rebase,abc123\n"
|
| 379 | )
|
| 380 | out_path = root / "scenarios.json"
|
| 381 | proc = run_importer(
|
| 382 | "--csv", str(csv_path),
|
| 383 | "--mirror-root", str(mirror_root),
|
| 384 | "--out", str(out_path),
|
| 385 | )
|
| 386 |
|
| 387 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 388 | document = json.loads(out_path.read_text())
|
| 389 |
|
| 390 | scenario = document["scenarios"][0]
|
| 391 | self.assertEqual(scenario["status"], "ready")
|
| 392 | self.assertIsNone(scenario["oracle"])
|
| 393 | self.assertEqual(scenario["oracle_note"], "oracle_defined_at_runtime:tree_equivalence")
|
| 394 |
|
| 395 | def test_normalized_oracle_mode_recorded(self) -> None:
|
| 396 | with tempfile.TemporaryDirectory() as tmp:
|
| 397 | root = Path(tmp)
|
| 398 | mirror_root = root / "mirrors"
|
| 399 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 400 | csv_path = root / "lite.csv"
|
| 401 | csv_path.write_text(
|
| 402 | "scenario_id,repository,task_type,merge_commit_sha\n"
|
| 403 | "n1,octo/widgets,merge,abc123\n"
|
| 404 | )
|
| 405 | out_path = root / "scenarios.json"
|
| 406 | proc = run_importer(
|
| 407 | "--csv", str(csv_path),
|
| 408 | "--mirror-root", str(mirror_root),
|
| 409 | "--out", str(out_path),
|
| 410 | "--oracle", "normalized",
|
| 411 | )
|
| 412 |
|
| 413 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 414 | document = json.loads(out_path.read_text())
|
| 415 |
|
| 416 | self.assertEqual(document["oracle_mode"], "normalized")
|
| 417 | scenario = document["scenarios"][0]
|
| 418 | self.assertEqual(scenario["status"], "ready")
|
| 419 | self.assertEqual(
|
| 420 | scenario["oracle"],
|
| 421 | {"mode": "normalized", "normalization": "whitespace_and_blank_lines"},
|
| 422 | )
|
| 423 |
|
| 424 | def test_normalized_oracle_mode_recorded_on_canonical_merge(self) -> None:
|
| 425 | with tempfile.TemporaryDirectory() as tmp:
|
| 426 | root = Path(tmp)
|
| 427 | mirror_root = root / "mirrors"
|
| 428 | (mirror_root / "octo__widgets").mkdir(parents=True)
|
| 429 | csv_path = root / "ggb.csv"
|
| 430 | write_real_csv(csv_path, [real_row(id="m1")])
|
| 431 | out_path = root / "scenarios.json"
|
| 432 | proc = run_importer(
|
| 433 | "--csv", str(csv_path),
|
| 434 | "--mirror-root", str(mirror_root),
|
| 435 | "--out", str(out_path),
|
| 436 | "--oracle", "normalized",
|
| 437 | )
|
| 438 |
|
| 439 | self.assertEqual(proc.returncode, 0, proc.stderr)
|
| 440 | document = json.loads(out_path.read_text())
|
| 441 |
|
| 442 | scenario = document["scenarios"][0]
|
| 443 | self.assertEqual(scenario["status"], "ready")
|
| 444 | self.assertEqual(
|
| 445 | scenario["oracle"],
|
| 446 | {"mode": "normalized", "normalization": "whitespace_and_blank_lines"},
|
| 447 | )
|
| 448 |
|
| 449 |
|
| 450 | class ExitCodeTest(unittest.TestCase):
|
| 451 | def test_missing_csv_exits_3_with_skip_json(self) -> None:
|
| 452 | with tempfile.TemporaryDirectory() as tmp:
|
| 453 | root = Path(tmp)
|
| 454 | missing = root / "nope.csv"
|
| 455 | proc = run_importer(
|
| 456 | "--csv", str(missing),
|
| 457 | "--mirror-root", str(root / "mirrors"),
|
| 458 | "--out", str(root / "scenarios.json"),
|
| 459 | )
|
| 460 |
|
| 461 | self.assertEqual(proc.returncode, 3)
|
| 462 | payload = json.loads(proc.stdout)
|
| 463 | self.assertEqual(payload["status"], "skipped")
|
| 464 | self.assertEqual(payload["reason"], f"dataset_missing:{missing}")
|
| 465 |
|
| 466 | def test_all_skipped_exits_3(self) -> None:
|
| 467 | with tempfile.TemporaryDirectory() as tmp:
|
| 468 | root = Path(tmp)
|
| 469 | mirror_root = root / "mirrors"
|
| 470 | mirror_root.mkdir()
|
| 471 | csv_path = root / "lite.csv"
|
| 472 | csv_path.write_text(
|
| 473 | "id,repo,scenario_type,commit\n"
|
| 474 | "s1,acme/missing,merge_conflict,abc123\n"
|
| 475 | "s2,acme/missing,nonsense,def456\n"
|
| 476 | )
|
| 477 | out_path = root / "scenarios.json"
|
| 478 | proc = run_importer(
|
| 479 | "--csv", str(csv_path),
|
| 480 | "--mirror-root", str(mirror_root),
|
| 481 | "--out", str(out_path),
|
| 482 | )
|
| 483 |
|
| 484 | self.assertEqual(proc.returncode, 3, proc.stderr)
|
| 485 | document = json.loads(out_path.read_text())
|
| 486 |
|
| 487 | self.assertEqual(document["counts"], {"ready": 0, "skipped": 2})
|
| 488 |
|
| 489 | def test_all_skipped_real_schema_exits_3(self) -> None:
|
| 490 | with tempfile.TemporaryDirectory() as tmp:
|
| 491 | root = Path(tmp)
|
| 492 | mirror_root = root / "mirrors"
|
| 493 | mirror_root.mkdir()
|
| 494 | csv_path = root / "ggb.csv"
|
| 495 | write_real_csv(
|
| 496 | csv_path,
|
| 497 | [real_row(id="m1", name="acme/missing"),
|
| 498 | real_row(id="u1", scenario="not json")],
|
| 499 | )
|
| 500 | out_path = root / "scenarios.json"
|
| 501 | proc = run_importer(
|
| 502 | "--csv", str(csv_path),
|
| 503 | "--mirror-root", str(mirror_root),
|
| 504 | "--out", str(out_path),
|
| 505 | )
|
| 506 |
|
| 507 | self.assertEqual(proc.returncode, 3, proc.stderr)
|
| 508 | document = json.loads(out_path.read_text())
|
| 509 |
|
| 510 | self.assertEqual(document["counts"], {"ready": 0, "skipped": 2})
|
| 511 |
|
| 512 | def test_missing_required_args_is_usage_error(self) -> None:
|
| 513 | proc = run_importer("--csv", "whatever.csv")
|
| 514 | self.assertEqual(proc.returncode, 2)
|
| 515 |
|
| 516 |
|
| 517 | if __name__ == "__main__":
|
| 518 | unittest.main()
|