144 lines
5.7 KiB
Python
144 lines
5.7 KiB
Python
"""100% compliance test for Allegro Commit-or-Abort (M2, Epic #842)."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import cycle_guard as cg
|
|
|
|
|
|
class TestCycleGuard(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmpdir = tempfile.TemporaryDirectory()
|
|
self.state_path = os.path.join(self.tmpdir.name, "cycle_state.json")
|
|
cg.STATE_PATH = self.state_path
|
|
|
|
def tearDown(self):
|
|
self.tmpdir.cleanup()
|
|
cg.STATE_PATH = cg.DEFAULT_STATE
|
|
|
|
def test_load_empty_state(self):
|
|
state = cg.load_state(self.state_path)
|
|
self.assertEqual(state["status"], "complete")
|
|
self.assertIsNone(state["cycle_id"])
|
|
|
|
def test_start_cycle(self):
|
|
state = cg.start_cycle("M2: Commit-or-Abort", path=self.state_path)
|
|
self.assertEqual(state["status"], "in_progress")
|
|
self.assertEqual(state["target"], "M2: Commit-or-Abort")
|
|
self.assertIsNotNone(state["cycle_id"])
|
|
|
|
def test_start_slice_requires_in_progress(self):
|
|
with self.assertRaises(RuntimeError):
|
|
cg.start_slice("test", path=self.state_path)
|
|
|
|
def test_slice_lifecycle(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("gather", path=self.state_path)
|
|
state = cg.load_state(self.state_path)
|
|
self.assertEqual(len(state["slices"]), 1)
|
|
self.assertEqual(state["slices"][0]["name"], "gather")
|
|
self.assertEqual(state["slices"][0]["status"], "in_progress")
|
|
|
|
cg.end_slice(status="complete", artifact="artifact.txt", path=self.state_path)
|
|
state = cg.load_state(self.state_path)
|
|
self.assertEqual(state["slices"][0]["status"], "complete")
|
|
self.assertEqual(state["slices"][0]["artifact"], "artifact.txt")
|
|
self.assertIsNotNone(state["slices"][0]["ended_at"])
|
|
|
|
def test_commit_cycle(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("work", path=self.state_path)
|
|
cg.end_slice(path=self.state_path)
|
|
proof = {"files": ["a.py"]}
|
|
state = cg.commit_cycle(proof=proof, path=self.state_path)
|
|
self.assertEqual(state["status"], "complete")
|
|
self.assertEqual(state["proof"], proof)
|
|
self.assertIsNotNone(state["completed_at"])
|
|
|
|
def test_commit_without_in_progress_fails(self):
|
|
with self.assertRaises(RuntimeError):
|
|
cg.commit_cycle(path=self.state_path)
|
|
|
|
def test_abort_cycle(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("work", path=self.state_path)
|
|
state = cg.abort_cycle("manual abort", path=self.state_path)
|
|
self.assertEqual(state["status"], "aborted")
|
|
self.assertEqual(state["abort_reason"], "manual abort")
|
|
self.assertIsNotNone(state["aborted_at"])
|
|
self.assertEqual(state["slices"][-1]["status"], "aborted")
|
|
|
|
def test_slice_timeout_true(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("work", path=self.state_path)
|
|
# Manually backdate slice start to 11 minutes ago
|
|
state = cg.load_state(self.state_path)
|
|
old = (datetime.now(timezone.utc) - timedelta(minutes=11)).isoformat()
|
|
state["slices"][0]["started_at"] = old
|
|
cg.save_state(state, self.state_path)
|
|
self.assertTrue(cg.check_slice_timeout(max_minutes=10, path=self.state_path))
|
|
|
|
def test_slice_timeout_false(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("work", path=self.state_path)
|
|
self.assertFalse(cg.check_slice_timeout(max_minutes=10, path=self.state_path))
|
|
|
|
def test_resume_or_abort_keeps_fresh_cycle(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
state = cg.resume_or_abort(path=self.state_path)
|
|
self.assertEqual(state["status"], "in_progress")
|
|
|
|
def test_resume_or_abort_aborts_stale_cycle(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
# Backdate start to 31 minutes ago
|
|
state = cg.load_state(self.state_path)
|
|
old = (datetime.now(timezone.utc) - timedelta(minutes=31)).isoformat()
|
|
state["started_at"] = old
|
|
cg.save_state(state, self.state_path)
|
|
state = cg.resume_or_abort(path=self.state_path)
|
|
self.assertEqual(state["status"], "aborted")
|
|
self.assertIn("crash recovery", state["abort_reason"])
|
|
|
|
def test_slice_duration_minutes(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("work", path=self.state_path)
|
|
# Backdate by 5 minutes
|
|
state = cg.load_state(self.state_path)
|
|
old = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat()
|
|
state["slices"][0]["started_at"] = old
|
|
cg.save_state(state, self.state_path)
|
|
mins = cg.slice_duration_minutes(path=self.state_path)
|
|
self.assertAlmostEqual(mins, 5.0, delta=0.5)
|
|
|
|
def test_cli_resume_prints_status(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
rc = cg.main(["resume"])
|
|
self.assertEqual(rc, 0)
|
|
|
|
def test_cli_check_timeout(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("work", path=self.state_path)
|
|
state = cg.load_state(self.state_path)
|
|
old = (datetime.now(timezone.utc) - timedelta(minutes=11)).isoformat()
|
|
state["slices"][0]["started_at"] = old
|
|
cg.save_state(state, self.state_path)
|
|
rc = cg.main(["check"])
|
|
self.assertEqual(rc, 1)
|
|
|
|
def test_cli_check_ok(self):
|
|
cg.start_cycle("test", path=self.state_path)
|
|
cg.start_slice("work", path=self.state_path)
|
|
rc = cg.main(["check"])
|
|
self.assertEqual(rc, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|