mirror of
https://github.com/saymrwulf/BraiinsRatchet.git
synced 2026-07-29 19:52:28 +00:00
125 lines
4.8 KiB
Python
125 lines
4.8 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
import sqlite3
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from braiins_ratchet.lifecycle import (
|
|
close_manual_position,
|
|
get_lifecycle_status,
|
|
init_lifecycle_db,
|
|
list_manual_positions,
|
|
open_manual_position,
|
|
recover_stale_active_watch,
|
|
render_manual_positions,
|
|
render_lifecycle_status,
|
|
render_supervisor_plan,
|
|
_sync_recent_watch_cooldown,
|
|
)
|
|
|
|
|
|
class LifecycleTests(unittest.TestCase):
|
|
def test_lifecycle_tables_initialize_and_status_defaults(self) -> None:
|
|
conn = sqlite3.connect(":memory:")
|
|
|
|
init_lifecycle_db(conn)
|
|
status = get_lifecycle_status(conn)
|
|
|
|
self.assertEqual(status.phase, "idle")
|
|
self.assertIsNone(status.next_action_utc)
|
|
self.assertIn("no lifecycle state", status.message)
|
|
|
|
def test_lifecycle_status_renders_countdown(self) -> None:
|
|
conn = sqlite3.connect(":memory:")
|
|
init_lifecycle_db(conn)
|
|
next_action = datetime.now(UTC) + timedelta(minutes=5)
|
|
conn.execute("INSERT INTO lifecycle_state (key, value) VALUES (?, ?)", ("phase", "cooldown"))
|
|
conn.execute(
|
|
"INSERT INTO lifecycle_state (key, value) VALUES (?, ?)",
|
|
("next_action_utc", next_action.isoformat(timespec="seconds")),
|
|
)
|
|
conn.commit()
|
|
|
|
text = render_lifecycle_status(conn)
|
|
|
|
self.assertIn("Phase: cooldown", text)
|
|
self.assertIn("Countdown:", text)
|
|
|
|
def test_supervisor_plan_states_monitor_only_resume_contract(self) -> None:
|
|
text = render_supervisor_plan()
|
|
|
|
self.assertIn("Resume persisted lifecycle state", text)
|
|
self.assertIn("Restart ./scripts/ratchet supervise", text)
|
|
self.assertIn("never places", text)
|
|
|
|
def test_manual_position_open_blocks_lifecycle_until_closed(self) -> None:
|
|
conn = sqlite3.connect(":memory:")
|
|
init_lifecycle_db(conn)
|
|
|
|
position_id = open_manual_position(
|
|
conn,
|
|
venue="braiins",
|
|
description="manual long bid",
|
|
expected_maturity_utc="2026-04-30T00:00:00+00:00",
|
|
payload={"spend_btc": "0.0001"},
|
|
)
|
|
|
|
status = get_lifecycle_status(conn)
|
|
active = list_manual_positions(conn, status="active")
|
|
self.assertEqual(status.phase, "manual_exposure_active")
|
|
self.assertEqual(len(active), 1)
|
|
self.assertEqual(active[0].id, position_id)
|
|
self.assertIn("manual long bid", render_manual_positions(conn))
|
|
|
|
self.assertTrue(close_manual_position(conn, position_id))
|
|
self.assertEqual(list_manual_positions(conn, status="active"), [])
|
|
self.assertEqual(get_lifecycle_status(conn).phase, "idle")
|
|
|
|
def test_recent_watch_report_synchronizes_supervisor_cooldown(self) -> None:
|
|
conn = sqlite3.connect(":memory:")
|
|
init_lifecycle_db(conn)
|
|
completed_watch = SimpleNamespace(
|
|
report_path="reports/run-example.md",
|
|
remaining_minutes=42,
|
|
earliest_action_utc="2026-04-28T15:41:51+00:00",
|
|
)
|
|
operator_state = SimpleNamespace(completed_watch=completed_watch)
|
|
|
|
with patch("braiins_ratchet.lifecycle.get_operator_state", return_value=operator_state):
|
|
wait_seconds = _sync_recent_watch_cooldown(conn)
|
|
|
|
status = get_lifecycle_status(conn)
|
|
self.assertEqual(wait_seconds, 42 * 60)
|
|
self.assertEqual(status.phase, "cooldown")
|
|
self.assertEqual(status.next_action_utc, "2026-04-28T15:41:51+00:00")
|
|
self.assertIn("recent watch report", status.message)
|
|
|
|
def test_recover_stale_active_watch_writes_partial_report_and_cooldown(self) -> None:
|
|
conn = sqlite3.connect(":memory:")
|
|
init_lifecycle_db(conn)
|
|
active_watch = SimpleNamespace(
|
|
read_text=lambda encoding: (
|
|
'{"pid": 123456, "run_id": "run-crashed", '
|
|
'"started_utc": "2026-04-29T08:48:06+00:00", '
|
|
'"planned_cycles": 24, "interval_seconds": 300}'
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("braiins_ratchet.lifecycle.ACTIVE_WATCH", active_watch),
|
|
patch("braiins_ratchet.lifecycle._pid_exists", return_value=False),
|
|
patch("braiins_ratchet.lifecycle.finish_experiment", return_value="reports/run-crashed.md") as finish,
|
|
):
|
|
report = recover_stale_active_watch(conn)
|
|
|
|
status = get_lifecycle_status(conn)
|
|
self.assertEqual(report, "reports/run-crashed.md")
|
|
self.assertEqual(status.phase, "cooldown")
|
|
self.assertEqual(status.last_run_id, "run-crashed")
|
|
self.assertIn("watch recovered after engine crash", status.message)
|
|
finish.assert_called_once()
|
|
self.assertEqual(finish.call_args.kwargs["status"], "recovered_after_crash")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|