| """Data contract tests; these never start Godot or require network access.""" |
| import importlib.util |
| import json |
| from pathlib import Path |
| import sys |
| import tempfile |
| import unittest |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT / "src")) |
| from vibegame.data import (Deduper, canonical_output, has_conflicting_header, |
| identify_license, read_jsonl, sha256, write_jsonl) |
|
|
| spec = importlib.util.spec_from_file_location("build_dataset", ROOT / "scripts/build_dataset.py") |
| builder = importlib.util.module_from_spec(spec) |
| spec.loader.exec_module(builder) |
|
|
|
|
| def bundle(extra=None): |
| files = {"project.godot": '[application]\nrun/main_scene="res://main.tscn"', |
| "main.tscn": '[gd_scene format=3]\n[node name="Main" type="Node2D"]', |
| "main.gd": "extends Node2D\n"} |
| files.update(extra or {}) |
| return json.dumps({"files": files}) |
|
|
|
|
| class DataTests(unittest.TestCase): |
| def test_canonical_output_has_required_files(self): |
| canonical, files = canonical_output(bundle()) |
| self.assertEqual(set(json.loads(canonical)), {"files"}) |
| self.assertEqual(len(files), 3) |
|
|
| def test_refuse_traversal_and_assets(self): |
| for path in ["../escape.gd", "/escape.gd", ".godot/cache.gd", "a/../bad.gd", "C:\\bad.gd", "art.png"]: |
| with self.subTest(path=path), self.assertRaises(ValueError): |
| canonical_output(bundle({path: "payload"})) |
|
|
| def test_refuse_missing_project_or_extra_top_level(self): |
| for output in ['{"files":{"main.gd":"extends Node"}}', '{"files":{},"explanation":"x"}']: |
| with self.assertRaises(ValueError): |
| canonical_output(output) |
|
|
| def test_license_filter_rejects_mixed_license(self): |
| mit = ('Permission is hereby granted, free of charge to anyone. ' |
| 'The above copyright notice and this permission notice. ' |
| 'THE SOFTWARE IS PROVIDED AS IS.') |
| self.assertEqual(identify_license(mit), "MIT") |
| self.assertIsNone(identify_license(mit + " Art assets CC-BY-NC-SA 4.0")) |
| self.assertIsNone(identify_license("See website for license")) |
| self.assertTrue(has_conflicting_header("# SPDX-License-Identifier: GPL-3.0-only\nextends Node")) |
| self.assertTrue(has_conflicting_header("# SPDX-License-Identifier: MIT OR GPL-3.0-only")) |
| self.assertFalse(has_conflicting_header("# SPDX-License-Identifier: MIT\nextends Node")) |
| self.assertTrue(has_conflicting_header("# SPDX-License-Identifier: Apache-2.0", expected_license="MIT")) |
| self.assertTrue(has_conflicting_header("# Released under the Apache License", expected_license="MIT")) |
| self.assertFalse(has_conflicting_header('var label = "GPL"')) |
|
|
| def test_dedupe_catches_whitespace_and_near_copies(self): |
| deduper = Deduper(threshold=.75, width=3) |
| original = "make an arcade game where the player catches a ball and scores ten points to win" |
| deduper.add("first", original) |
| self.assertEqual(deduper.match(original.upper() + " "), "first") |
| self.assertEqual(deduper.match(original + " please"), "first") |
| self.assertIsNone(deduper.match("Solve a warehouse crate puzzle by pushing boxes onto goals")) |
|
|
| def test_eval_frozen_unique_and_unlabeled(self): |
| rows = builder.frozen_eval_rows() |
| self.assertEqual(len(rows), 500) |
| self.assertEqual(len({row["id"] for row in rows}), 500) |
| self.assertEqual(len({row["sha256"] for row in rows}), 500) |
| self.assertEqual(len({row["family"] for row in rows}), 25) |
| self.assertTrue(all("output" not in row for row in rows)) |
| self.assertTrue(all(sha256(row["instruction"]) == row["sha256"] for row in rows)) |
|
|
| def test_freeze_refuses_modified_benchmark(self): |
| with tempfile.TemporaryDirectory() as directory: |
| root = Path(directory) |
| builder.freeze_eval(root) |
| builder.freeze_eval(root) |
| target = root / "eval_prompts.jsonl" |
| rows = list(read_jsonl(target)) |
| rows[0]["instruction"] = "Changed benchmark" |
| write_jsonl(target, rows) |
| with self.assertRaises(ValueError): |
| builder.freeze_eval(root) |
|
|
| def test_split_no_duplicated_rows_and_order_independent(self): |
| rows = [{"instruction": f"unique prompt {i}", "output": f"project {i}"} for i in range(10)] |
| train, dev = builder.split_records(rows) |
| self.assertEqual((len(train), len(dev)), (9, 1)) |
| self.assertEqual((train, dev), builder.split_records(list(reversed(rows)))) |
| self.assertFalse({row["instruction"] for row in train} & {row["instruction"] for row in dev}) |
| with self.assertRaises(ValueError): |
| builder.split_records(rows[:1]) |
|
|
| def test_cache_requires_hashes_runtime_and_exact_sources(self): |
| from vibegame.project import project_sha256 |
| _, files = canonical_output(bundle()) |
| with tempfile.TemporaryDirectory() as directory: |
| target = Path(directory) |
| (target / "project").mkdir() |
| for name, source in files.items(): |
| (target / "project" / name).write_text(source) |
| result = {"compile_success": True, "clean_run_60s": True, "run_seconds": 60, |
| "godot_binary_sha256": "test-binary", "project_sha256": project_sha256(files)} |
| for phase in ["import", "runtime"]: |
| (target / f"{phase}.log").write_text("test fixture\n") |
| result[phase] = {"success": True, "wall_seconds": 60.1, |
| "log_sha256": sha256("test fixture\n")} |
| (target / "validation.json").write_text(json.dumps(result)) |
| self.assertIsNotNone(builder.successful_cached_validation(target, files, "test-binary", 60)) |
| self.assertIsNone(builder.successful_cached_validation(target, files, "different-binary", 60)) |
| self.assertIsNone(builder.successful_cached_validation(target, files, "test-binary", 61)) |
| (target / "runtime.log").write_text("changed fixture") |
| self.assertIsNone(builder.successful_cached_validation(target, files, "test-binary", 60)) |
|
|
| def test_teacher_prompts_are_not_blind_eval(self): |
| target = ROOT / "data/teacher_prompts.jsonl" |
| if not target.exists(): |
| self.skipTest("teacher prompts not included") |
| prompts = Deduper(threshold=.78, width=3) |
| for row in builder.frozen_eval_rows(): |
| prompts.add(row["id"], row["instruction"]) |
| rows = list(read_jsonl(target)) |
| self.assertEqual(len(rows), 100) |
| for row in rows: |
| self.assertEqual(set(row), {"id", "instruction"}) |
| self.assertIsNone(prompts.match(row["instruction"]), row["id"]) |
| prompts.add(row["id"], row["instruction"]) |
|
|
| def test_collected_reference_integrity(self): |
| target = ROOT / "data/reference_gd.jsonl" |
| if not target.exists(): |
| self.skipTest("optional reference corpus not downloaded") |
| for row in read_jsonl(target): |
| self.assertIn(row["license"], {"MIT", "Apache-2.0", "CC0-1.0"}) |
| self.assertTrue(row["path"].endswith(".gd")) |
| self.assertEqual(len(row["commit"]), 40) |
| self.assertEqual(row["sha256"], sha256(row["code"])) |
| self.assertEqual(row["use"], "reference_only_not_supervised_training") |
| self.assertFalse(has_conflicting_header(row["code"], expected_license=row["license"])) |
| for license_path in row["license_files"]: |
| self.assertTrue((ROOT / "data" / license_path).is_file()) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
