| """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: |
| 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() + " ") |
| 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() |
|
|
