File size: 10,394 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 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 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 | """Offline recovery tests: tiny synthetic evidence, no CUDA, network, or Godot."""
import hashlib
import json
import sys
import tempfile
import unittest
from concurrent.futures import Future
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
import evaluate
import validate_saved as recovery
from vibegame.project import project_sha256, write_project
FILES = {"project.godot": 'config_version=5\n[application]\nrun/main_scene="res://main.tscn"\n',
"main.tscn": '[gd_scene load_steps=2 format=3]\n[ext_resource type="Script" path="res://main.gd" id="1"]\n[node name="Game" type="Node2D"]\nscript=ExtResource("1")\n',
"main.gd": "extends Node2D\nfunc _ready():\n\tqueue_redraw()\n"}
def save(path, value):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value) + "\n")
def evidence(files, destination, godot, seconds, *, clean=True):
destination.mkdir(parents=True, exist_ok=True)
write_project(files, destination / "project")
result = {"compile_success": True, "clean_run": clean, "clean_run_60s": clean,
"run_seconds": seconds, "project_sha256": project_sha256(files),
"godot_binary_sha256": recovery.file_hash(Path(godot)), "error": None}
for name in ("import", "runtime"):
log = destination / (name + ".log")
log.write_text("Offline test fixture, not actual engine evidence\n")
result[name] = {"success": clean if name == "runtime" else True,
"log_sha256": recovery.file_hash(log), "wall_seconds": seconds,
"required_seconds": seconds, "startup_seconds": 0.1}
save(destination / "validation.json", result)
return result
class ImmediatePool:
def __init__(self, **kwargs):
self.kwargs = kwargs
def __enter__(self):
return self
def __exit__(self, *args):
return False
def submit(self, function, *args):
future = Future()
try:
future.set_result(function(*args))
except Exception as error: # noqa: BLE001 - emulate Future exception delivery
future.set_exception(error)
return future
class SavedValidationTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name).resolve()
self.evaluation = self.root / "evaluation"
(self.evaluation / "items").mkdir(parents=True)
self.godot = self.root / "godot-fixture"
self.godot.write_bytes(b"never executed")
self.prompts = self.root / "prompts.jsonl"
self.rows = [{"id": f"test-{i}", "instruction": f"Offline synthetic test prompt {i}"} for i in range(500)]
self.prompts.write_text("".join(json.dumps(row) + "\n" for row in self.rows))
self.manifest = {"expected_count": 500, "run_seconds": 60, "smoke_test": False,
"sample_indices": [0, 200, 400], "workers": 4, "batch_size": 8,
"prompts_sha256": recovery.file_hash(self.prompts), "runtime_sha256": recovery.file_hash(self.godot),
"generation": {"backend": "unsloth", "adapter": {"sha256": "unchanged"}}}
save(self.evaluation / "manifest.json", self.manifest)
def tearDown(self):
self.temp.cleanup()
def item(self, index, *, output=None, generation=True):
item = self.evaluation / "items" / f"{index:04d}"
row = self.rows[index]
save(item / "attempt.json", {"index": index, "prompt_id": row["id"], "instruction": row["instruction"]})
if generation:
output = json.dumps({"files": FILES}) if output is None else output
save(item / "generation.json", {"output": output, "output_sha256": hashlib.sha256(output.encode()).hexdigest(),
"generation_seconds": 99, "generation_call_seconds": 101, "generation_batch_size": 8})
return item
def plan(self):
return recovery.prepare(self.evaluation, self.prompts, self.godot)
def run_recovery(self, plan=None):
with patch.object(recovery, "preflight") as preflight, patch.object(recovery, "ProcessPoolExecutor", ImmediatePool), \
patch.object(evaluate, "make_generator", side_effect=AssertionError("No model may load")), \
patch.object(evaluate, "validate_project", side_effect=evidence) as engine:
summary = recovery.recover(plan or self.plan(), self.godot)
return summary, engine, preflight
def test_resume_saved_only_reuses_completed_worker_and_preserves_original_bytes(self):
completed = self.item(0)
save(completed / "result.json", {"index": 0, "prompt_id": "test-0", "compile_success": False, "clean_run_60s": False})
worker_done = self.item(1)
evidence(FILES, worker_done / "check", self.godot, 60)
interrupted = self.item(2)
(interrupted / "check").mkdir()
malformed = self.item(3, output="not valid JSON")
no_output = self.item(4, generation=False)
fixed_paths = [self.evaluation / "manifest.json", completed / "result.json", worker_done / "check/validation.json"]
fixed_paths += list(self.evaluation.glob("items/*/generation.json")) + list(self.evaluation.glob("items/*/attempt.json"))
before = {path: path.read_bytes() for path in fixed_paths}
summary, engine, _ = self.run_recovery()
self.assertEqual(engine.call_count, 1)
self.assertEqual(engine.call_args.args[1], interrupted / "check-resume-001")
self.assertTrue((interrupted / "check/interruption.json").exists())
self.assertEqual(summary["recorded_prompts"], 4)
self.assertEqual(summary["pending_prompts"], 496)
self.assertEqual(summary["status"], "incomplete")
self.assertIsNone(summary["compile_success_rate"])
self.assertIsNone(summary["clean_run_60s_rate"])
self.assertFalse((no_output / "result.json").exists())
self.assertFalse((self.evaluation / "items/0005").exists())
self.assertIn("ProjectError", recovery.read_json(malformed / "result.json")["error"])
for item in (worker_done, interrupted):
self.assertIsNone(recovery.read_json(item / "result.json")["prompt_to_launch_seconds"])
self.assertEqual(recovery.read_json(interrupted / "result.json")["interrupted_validation_attempts"], 1)
self.assertEqual(before, {path: path.read_bytes() for path in fixed_paths})
summary, engine, _ = self.run_recovery()
self.assertEqual(engine.call_count, 0)
self.assertEqual(summary["recorded_prompts"], 4)
def test_wrong_original_runtime_and_prompt_file_rejected_before_validation(self):
self.item(0)
for path in (self.godot, self.prompts):
original = path.read_bytes()
path.write_bytes(original + b"changed")
with self.assertRaises(ValueError):
self.plan()
path.write_bytes(original)
self.assertFalse((self.evaluation / "items/0000/result.json").exists())
def test_modified_generation_hash_and_attempt_identity_rejected(self):
item = self.item(0)
for name, key, value in (("generation.json", "output", "modified"), ("attempt.json", "prompt_id", "different")):
path = item / name
original = path.read_bytes()
record = recovery.read_json(path)
record[key] = value
save(path, record)
with self.assertRaises(ValueError):
self.plan()
path.write_bytes(original)
def test_completed_worker_requires_source_runtime_and_log_fingerprints(self):
item = self.item(0)
evidence(FILES, item / "check", self.godot, 60)
for path in (item / "check/import.log", item / "check/runtime.log", item / "check/project/main.gd"):
original = path.read_bytes()
path.write_bytes(original + b"changed")
with self.assertRaises(ValueError):
self.plan()
path.write_bytes(original)
path = item / "check/validation.json"
record = recovery.read_json(path)
record["godot_binary_sha256"] = "f" * 64
save(path, record)
with self.assertRaises(ValueError):
self.plan()
self.assertFalse((item / "result.json").exists())
def test_multiple_completed_checks_and_symlink_are_rejected(self):
item = self.item(0)
evidence(FILES, item / "check", self.godot, 60)
evidence(FILES, item / "check-resume-001", self.godot, 60)
with self.assertRaises(ValueError):
self.plan()
another = self.item(1)
outside = self.root / "outside.json"
(another / "generation.json").replace(outside)
(another / "generation.json").symlink_to(outside)
with self.assertRaises(ValueError):
self.plan()
def test_manifest_change_after_preparation_stops_before_writes(self):
self.item(0)
plan = self.plan()
path = self.evaluation / "manifest.json"
path.write_text(path.read_text() + " ")
with self.assertRaises(ValueError):
self.run_recovery(plan)
self.assertFalse((self.evaluation / "items/0000/result.json").exists())
def test_saved_raw_change_after_preparation_stops_without_new_result(self):
item = self.item(0)
plan = self.plan()
path = item / "generation.json"
path.write_text(path.read_text() + " ") # Same JSON/output, different raw bytes.
with self.assertRaises(ValueError):
self.run_recovery(plan)
self.assertFalse((item / "result.json").exists())
def test_incomplete_generation_is_preserved_and_never_resampled(self):
item = self.item(0)
(item / "generation.json").write_bytes(b'{"output":')
raw = (item / "generation.json").read_bytes()
summary, engine, _ = self.run_recovery()
self.assertEqual(engine.call_count, 0)
self.assertEqual(summary["recorded_prompts"], 1)
self.assertEqual((item / "generation.json").read_bytes(), raw)
self.assertIn("persistence interrupted", recovery.read_json(item / "result.json")["error"])
if __name__ == "__main__":
unittest.main()
|