File size: 7,720 Bytes
ae36cfb | 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 | """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()
|