| """Continuation lifecycle checks without GPUs, engines, or running subprocesses.""" |
| import json |
| from pathlib import Path |
| import sys |
| import tempfile |
| from types import SimpleNamespace |
| import unittest |
| from unittest.mock import Mock |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) |
| import continue_run |
|
|
|
|
| class NoSuchProcess(Exception): |
| pass |
|
|
|
|
| class ContinuationTests(unittest.TestCase): |
| def test_nonchild_zombie_does_not_wait_for_notebook_reaping(self): |
| process = Mock() |
| process.status.side_effect = ["running", "running", "zombie"] |
| process.cmdline.return_value = ["python", "scripts/train.py"] |
| process.is_running.return_value = True |
| psutil = SimpleNamespace(Process=Mock(return_value=process), |
| STATUS_ZOMBIE="zombie", NoSuchProcess=NoSuchProcess) |
| sleep = Mock() |
| continue_run.wait_for_training_process(123, psutil, sleep=sleep) |
| sleep.assert_called_once_with(2) |
| process.wait.assert_not_called() |
|
|
| def test_already_exited_training_and_pid_reuse(self): |
| psutil = SimpleNamespace(Process=Mock(side_effect=NoSuchProcess), |
| STATUS_ZOMBIE="zombie", NoSuchProcess=NoSuchProcess) |
| continue_run.wait_for_training_process(123, psutil, sleep=Mock()) |
| process = Mock() |
| process.status.return_value = "running" |
| process.cmdline.return_value = ["unrelated", "command"] |
| psutil.Process = Mock(return_value=process) |
| with self.assertRaisesRegex(RuntimeError, "not the training"): |
| continue_run.wait_for_training_process(123, psutil, sleep=Mock()) |
|
|
| def test_readiness_requires_live_current_pid_and_fresh_marker(self): |
| with tempfile.TemporaryDirectory() as folder: |
| root = Path(folder) |
| (root / "manifest.json").write_text("{}") |
| process = SimpleNamespace(pid=234, poll=lambda: None) |
| self.assertFalse(continue_run.evaluator_ready(root, process, 100)) |
| receipt = {"pid": 234, "ready_unix_seconds": 99, |
| "model_warmup_complete": True, "first_batch_processed": True} |
| marker = root / ".runtime-ready-234.json" |
| marker.write_text(json.dumps(receipt)) |
| self.assertFalse(continue_run.evaluator_ready(root, process, 100)) |
| receipt["ready_unix_seconds"] = 101 |
| marker.write_text(json.dumps(receipt)) |
| self.assertTrue(continue_run.evaluator_ready(root, process, 100)) |
| process.poll = lambda: 0 |
| self.assertFalse(continue_run.evaluator_ready(root, process, 100)) |
|
|
| def test_export_receipts_must_match_adapter_and_complete_file(self): |
| with tempfile.TemporaryDirectory() as folder: |
| root = Path(folder) |
| (root / "merged-fp16").mkdir() |
| (root / "gguf").mkdir() |
| merged = {"adapter_sha256": "exact-adapter", "tensor_count": 3} |
| (root / "merged-fp16/MERGE_COMPLETE.json").write_text(json.dumps(merged)) |
| model = root / "gguf/VibeGame-8B-Q4_K_M.gguf" |
| model.write_bytes(b"complete-file") |
| (root / "gguf/GGUF_COMPLETE.json").write_text(json.dumps({"source_merge": merged, "bytes": model.stat().st_size})) |
| self.assertTrue(continue_run.exports_complete(root, "exact-adapter")) |
| self.assertFalse(continue_run.exports_complete(root, "different-adapter")) |
| model.write_bytes(b"truncated") |
| self.assertFalse(continue_run.exports_complete(root, "exact-adapter")) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
