File size: 3,620 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 | """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()
|