#!/usr/bin/env python3 """ Unit tests for session-recovery.py
Tests: J.26.1.8 Track: J.26 - Session Recovery Automation Author: Claude (Opus 4.6) """
import json import os import signal import sys import tempfile import unittest from datetime import datetime, timedelta from pathlib import Path from unittest.mock import MagicMock, patch
Add scripts directory to path
sys.path.insert(0, str(Path(file).parent.parent))
Load session-recovery.py as a module (hyphenated filename)
from importlib.util import module_from_spec, spec_from_file_location
spec = spec_from_file_location( "session_recovery", Path(file).parent.parent / "session-recovery.py" ) session_recovery = module_from_spec(spec) spec.loader.exec_module(session_recovery)
ProcessInfo = session_recovery.ProcessInfo SessionInfo = session_recovery.SessionInfo SessionRecovery = session_recovery.SessionRecovery
class TestProcessInfo(unittest.TestCase): """Tests for ProcessInfo dataclass."""
def _make(self, cpu_time='0:00'):
return ProcessInfo(pid=1, tty='s001', cpu_time=cpu_time,
mem_percent=1.0, command='claude', user='test')
def test_cpu_minutes_mm_ss(self):
self.assertAlmostEqual(self._make('5:30').cpu_minutes, 5.5)
def test_cpu_minutes_hh_mm_ss(self):
self.assertAlmostEqual(self._make('1:30:00').cpu_minutes, 90.0)
def test_cpu_minutes_zero(self):
self.assertAlmostEqual(self._make('0:00').cpu_minutes, 0.0)
def test_cpu_minutes_invalid(self):
self.assertAlmostEqual(self._make('invalid').cpu_minutes, 0.0)
class TestSessionInfo(unittest.TestCase): """Tests for SessionInfo dataclass."""
def _make(self, uuid='abc', size=500, age_min=0):
return SessionInfo(
uuid=uuid, path=Path('/tmp/test.jsonl'),
project_hash='proj1', size_bytes=size,
modified=datetime.now() - timedelta(minutes=age_min)
)
def test_size_human_bytes(self):
self.assertEqual(self._make(size=500).size_human, '500B')
def test_size_human_kb(self):
self.assertEqual(self._make(size=5120).size_human, '5.0KB')
def test_size_human_mb(self):
self.assertEqual(self._make(size=5 * 1024 * 1024).size_human, '5.0MB')
def test_resume_command(self):
self.assertEqual(self._make(uuid='test-uuid-123').resume_command,
'claude --resume test-uuid-123')
def test_age_minutes(self):
s = self._make(age_min=15)
self.assertGreater(s.age_minutes, 14)
self.assertLess(s.age_minutes, 16)
class TestDetectClaudeProcesses(unittest.TestCase): """Tests for SessionRecovery.detect_claude_processes()."""
def _make_recovery(self):
r = SessionRecovery.__new__(SessionRecovery)
r.current_tty = 's009'
return r
@patch('subprocess.run')
def test_parses_ps_output(self, mock_run):
mock_run.return_value = MagicMock(
stdout=(
"USER PID %CPU %MEM VSZ RSS TT STAT STARTED TIME COMMAND\n"
"hal 12345 1.2 0.5 12345678 12345 s009 S+ 10:30AM 5:23 claude\n"
"hal 67890 0.3 0.2 9876543 9876 s013 S+ 11:00AM 2:10 claude\n"
)
)
processes = self._make_recovery().detect_claude_processes()
self.assertEqual(len(processes), 2)
self.assertEqual(processes[0].pid, 12345)
self.assertEqual(processes[1].pid, 67890)
@patch('subprocess.run')
def test_excludes_grep(self, mock_run):
mock_run.return_value = MagicMock(
stdout=(
"USER PID %CPU %MEM VSZ RSS TT STAT STARTED TIME COMMAND\n"
"hal 12345 1.2 0.5 12345678 12345 s009 S+ 10:30AM 5:23 claude\n"
"hal 99999 0.0 0.0 1234567 1234 s010 S+ 10:30AM 0:00 grep claude\n"
)
)
processes = self._make_recovery().detect_claude_processes()
self.assertEqual(len(processes), 1)
@patch('subprocess.run')
def test_handles_no_processes(self, mock_run):
mock_run.return_value = MagicMock(
stdout="USER PID %CPU %MEM VSZ RSS TT STAT STARTED TIME COMMAND\n"
)
processes = self._make_recovery().detect_claude_processes()
self.assertEqual(len(processes), 0)
class TestDetectHungSessions(unittest.TestCase): """Tests for SessionRecovery.detect_hung_sessions()."""
def _make_recovery(self, current_tty='s009', timeout=10):
r = SessionRecovery.__new__(SessionRecovery)
r.timeout_minutes = timeout
r.current_tty = current_tty
return r
def _proc(self, pid=100, tty='s013', cpu='5:00'):
return ProcessInfo(pid=pid, tty=tty, cpu_time=cpu,
mem_percent=1.0, command='claude', user='test')
def _session(self, uuid='old', age_min=30):
return SessionInfo(uuid=uuid, path=Path('/tmp/test.jsonl'),
project_hash='proj1', size_bytes=1000,
modified=datetime.now() - timedelta(minutes=age_min))
def test_detects_hung(self):
r = self._make_recovery()
hung = r.detect_hung_sessions([self._proc()], [self._session()])
self.assertEqual(len(hung), 1)
self.assertEqual(hung[0].uuid, 'old')
def test_ignores_recent(self):
r = self._make_recovery()
hung = r.detect_hung_sessions([self._proc()], [self._session(age_min=2)])
self.assertEqual(len(hung), 0)
def test_ignores_idle_cpu(self):
r = self._make_recovery()
hung = r.detect_hung_sessions([self._proc(cpu='0:00')], [self._session()])
self.assertEqual(len(hung), 0)
def test_excludes_current_tty(self):
r = self._make_recovery()
hung = r.detect_hung_sessions([self._proc(tty='s009')], [self._session()])
self.assertEqual(len(hung), 0)
class TestExportSession(unittest.TestCase): """Tests for SessionRecovery.export_session()."""
def test_creates_lossless_copy(self):
with tempfile.TemporaryDirectory() as tmpdir:
source = Path(tmpdir) / 'source.jsonl'
source.write_text('{"type":"test"}\n')
session = SessionInfo(uuid='test-uuid', path=source,
project_hash='proj1', size_bytes=16,
modified=datetime.now())
r = SessionRecovery.__new__(SessionRecovery)
r.export_dir = Path(tmpdir) / 'exports'
result = r.export_session(session)
self.assertIsNotNone(result)
self.assertTrue(result.exists())
self.assertIn('LOSSLESS', result.name)
self.assertIn('test-uuid', result.name)
self.assertEqual(result.read_text(), '{"type":"test"}\n')
def test_returns_none_on_missing_source(self):
with tempfile.TemporaryDirectory() as tmpdir:
session = SessionInfo(uuid='missing', path=Path('/nonexistent/f.jsonl'),
project_hash='proj1', size_bytes=0,
modified=datetime.now())
r = SessionRecovery.__new__(SessionRecovery)
r.export_dir = Path(tmpdir) / 'exports'
self.assertIsNone(r.export_session(session))
class TestKillProcess(unittest.TestCase): """Tests for SessionRecovery.kill_process()."""
@patch('os.kill')
def test_prevents_killing_current_session(self, mock_kill):
r = SessionRecovery.__new__(SessionRecovery)
r.current_tty = 's009'
with patch.object(r, 'detect_claude_processes', return_value=[
ProcessInfo(pid=123, tty='s009', cpu_time='1:00',
mem_percent=1.0, command='claude', user='test')
]):
self.assertFalse(r.kill_process(123))
mock_kill.assert_not_called()
@patch('os.kill')
def test_kills_other_session(self, mock_kill):
r = SessionRecovery.__new__(SessionRecovery)
r.current_tty = 's009'
with patch.object(r, 'detect_claude_processes', return_value=[
ProcessInfo(pid=456, tty='s013', cpu_time='1:00',
mem_percent=1.0, command='claude', user='test')
]):
self.assertTrue(r.kill_process(456))
mock_kill.assert_called_once_with(456, signal.SIGTERM)
@patch('os.kill')
def test_force_uses_sigkill(self, mock_kill):
r = SessionRecovery.__new__(SessionRecovery)
r.current_tty = 's009'
with patch.object(r, 'detect_claude_processes', return_value=[
ProcessInfo(pid=456, tty='s013', cpu_time='1:00',
mem_percent=1.0, command='claude', user='test')
]):
self.assertTrue(r.kill_process(456, force=True))
mock_kill.assert_called_once_with(456, signal.SIGKILL)
class TestGetAllSessions(unittest.TestCase): """Tests for SessionRecovery.get_all_sessions()."""
def test_discovers_sessions(self):
with tempfile.TemporaryDirectory() as tmpdir:
proj = Path(tmpdir) / 'projects' / 'hash123'
proj.mkdir(parents=True)
(proj / 'session-1.jsonl').write_text('{"t":1}\n')
(proj / 'session-2.jsonl').write_text('{"t":2}\n')
r = SessionRecovery.__new__(SessionRecovery)
r.claude_dir = Path(tmpdir) / 'projects'
sessions = r.get_all_sessions()
self.assertEqual(len(sessions), 2)
uuids = {s.uuid for s in sessions}
self.assertIn('session-1', uuids)
self.assertIn('session-2', uuids)
def test_handles_missing_dir(self):
r = SessionRecovery.__new__(SessionRecovery)
r.claude_dir = Path('/nonexistent/dir')
self.assertEqual(len(r.get_all_sessions()), 0)
def test_sorted_by_mtime(self):
with tempfile.TemporaryDirectory() as tmpdir:
proj = Path(tmpdir) / 'projects' / 'hash1'
proj.mkdir(parents=True)
old = proj / 'old.jsonl'
old.write_text('old\n')
# Set mtime to 1 hour ago
os.utime(old, (old.stat().st_atime, old.stat().st_mtime - 3600))
new = proj / 'new.jsonl'
new.write_text('new\n')
r = SessionRecovery.__new__(SessionRecovery)
r.claude_dir = Path(tmpdir) / 'projects'
sessions = r.get_all_sessions()
self.assertEqual(sessions[0].uuid, 'new')
self.assertEqual(sessions[1].uuid, 'old')
class TestGenerateResumeCommand(unittest.TestCase): """Tests for SessionRecovery.generate_resume_command()."""
def test_format(self):
r = SessionRecovery.__new__(SessionRecovery)
session = SessionInfo(uuid='abc-def-123', path=Path('/tmp/t.jsonl'),
project_hash='p1', size_bytes=100,
modified=datetime.now())
self.assertEqual(r.generate_resume_command(session),
'claude --resume abc-def-123')
class TestMapTtyToSession(unittest.TestCase): """Tests for SessionRecovery.map_tty_to_session()."""
def test_maps_tty_to_most_recent(self):
r = SessionRecovery.__new__(SessionRecovery)
proc = ProcessInfo(pid=100, tty='s013', cpu_time='1:00',
mem_percent=1.0, command='claude', user='test')
sessions = [
SessionInfo(uuid='newest', path=Path('/tmp/n.jsonl'),
project_hash='p1', size_bytes=100,
modified=datetime.now())
]
result = r.map_tty_to_session('s013', [proc], sessions)
self.assertIsNotNone(result)
self.assertEqual(result.uuid, 'newest')
self.assertEqual(result.process.pid, 100)
def test_returns_none_for_unknown_tty(self):
r = SessionRecovery.__new__(SessionRecovery)
proc = ProcessInfo(pid=100, tty='s013', cpu_time='1:00',
mem_percent=1.0, command='claude', user='test')
result = r.map_tty_to_session('s999', [proc], [])
self.assertIsNone(result)
if name == 'main': unittest.main()