mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-07 16:11:46 +02:00
160 lines
8.2 KiB
Python
160 lines
8.2 KiB
Python
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
from lifecycle import Lifecycle, run_test_cases
|
|
from owned_processes import OwnedProcesses, ProcessInspectionUnavailable, process_table, command_tokens
|
|
from qualification_observer import ActiveInspectionFailed, QualificationInspection, wait_for_owned_command
|
|
|
|
|
|
class QualificationObserverTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = 10
|
|
self.owner = SimpleNamespace(stop_deadline=None, observe=Mock(return_value={}), owned={})
|
|
self.inspection = QualificationInspection(self.owner, 190, clock=lambda: self.now)
|
|
|
|
def test_normal_budget_five_seconds_then_remaining_command_window(self):
|
|
with patch('qualification_observer.process_table', return_value={}) as table:
|
|
self.inspection.table()
|
|
table.assert_called_once_with(timeout=5)
|
|
self.now = 188.75
|
|
self.inspection.table()
|
|
self.assertEqual(table.call_args.kwargs['timeout'], 1.25)
|
|
|
|
def test_three_second_inspection_is_admitted_once_without_retry(self):
|
|
def table(*, timeout):
|
|
self.assertEqual(timeout, 5)
|
|
self.now += 3
|
|
return {123: {'started': 'fixture'}}
|
|
with patch('qualification_observer.process_table', side_effect=table) as call:
|
|
self.assertEqual(self.inspection.table(), {123: {'started': 'fixture'}})
|
|
self.assertEqual(call.call_count, 1)
|
|
|
|
def test_cleanup_uses_its_own_existing_deadline_after_active_expiry(self):
|
|
self.now = 191
|
|
self.owner.stop_deadline = 291
|
|
with patch('qualification_observer.command_tokens', return_value=['fixture']) as argv:
|
|
self.inspection.argv(123)
|
|
argv.assert_called_once_with(123, timeout=5)
|
|
self.now = 290.75
|
|
self.inspection.argv(123)
|
|
self.assertEqual(argv.call_args.kwargs['timeout'], .25)
|
|
|
|
def test_exhausted_active_deadline_never_starts_inspection(self):
|
|
self.now = 190
|
|
with patch('qualification_observer.process_table') as table:
|
|
with self.assertRaises(ActiveInspectionFailed): self.inspection.table()
|
|
table.assert_not_called()
|
|
|
|
def test_exhausted_cleanup_deadline_preserves_cleanup_failure_type(self):
|
|
self.owner.stop_deadline = self.now
|
|
with patch('qualification_observer.command_tokens') as argv:
|
|
with self.assertRaises(ProcessInspectionUnavailable): self.inspection.argv(123)
|
|
argv.assert_not_called()
|
|
|
|
def test_active_table_and_argv_failures_are_fatal_without_retry(self):
|
|
for operation, method, args, error in (
|
|
('process_table', self.inspection.table, (), subprocess.TimeoutExpired('ps', 5)),
|
|
('command_tokens', self.inspection.argv, (123,), self.argv_timeout()),
|
|
):
|
|
with self.subTest(operation=operation), patch('qualification_observer.' + operation, side_effect=error) as call:
|
|
with self.assertRaises(ActiveInspectionFailed): method(*args)
|
|
self.assertEqual(call.call_count, 1)
|
|
|
|
@staticmethod
|
|
def argv_timeout():
|
|
error = ProcessInspectionUnavailable()
|
|
error.__cause__ = subprocess.TimeoutExpired('ps', 5)
|
|
return error
|
|
|
|
def inspect_argv_race(self, after, error):
|
|
root_row = {'parent': 1, 'started': 'root-identity', 'status': 'sleeping', 'executable': '/fixture/Python'}
|
|
child_row = {'parent': 123, 'started': 'child-identity', 'status': 'sleeping', 'executable': '/fixture/node'}
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
private = Path(tmp).resolve(); private.chmod(0o700)
|
|
owner = OwnedProcesses(private/'journal.json', private, private, private/'sidecar.cjs')
|
|
inspection = QualificationInspection(owner, 190, clock=lambda: self.now)
|
|
owner.table = inspection.table; owner.argv = inspection.argv
|
|
next_rows = {123: root_row}
|
|
if after == 'same': next_rows[124] = child_row
|
|
elif after == 'reused': next_rows[124] = {**child_row, 'started': 'different-identity'}
|
|
with patch('qualification_observer.process_table', side_effect=[{123: root_row, 124: child_row}, next_rows]) as table:
|
|
with patch('qualification_observer.command_tokens', side_effect=error) as argv:
|
|
inspection.observe(123)
|
|
self.assertEqual(table.call_count, 2)
|
|
self.assertEqual(argv.call_count, 1)
|
|
self.assertNotIn(124, owner.owned)
|
|
|
|
def test_completed_empty_argv_after_natural_exit_does_not_invent_failure(self):
|
|
self.inspect_argv_race('gone', ProcessInspectionUnavailable())
|
|
|
|
def test_completed_ps_exit_one_after_pid_reuse_does_not_admit_new_identity(self):
|
|
error = ProcessInspectionUnavailable()
|
|
error.__cause__ = subprocess.CalledProcessError(1, 'ps')
|
|
self.inspect_argv_race('reused', error)
|
|
|
|
def test_same_identity_with_unresolved_argv_is_fatal(self):
|
|
with self.assertRaisesRegex(ActiveInspectionFailed, 'remains unresolved'):
|
|
self.inspect_argv_race('same', ProcessInspectionUnavailable())
|
|
|
|
def test_argv_timeout_never_gets_natural_exit_recheck(self):
|
|
with patch('qualification_observer.process_table') as table:
|
|
with patch('qualification_observer.command_tokens', side_effect=self.argv_timeout()) as argv:
|
|
with self.assertRaises(ActiveInspectionFailed): self.inspection.argv(123)
|
|
self.assertEqual(argv.call_count, 1)
|
|
table.assert_not_called()
|
|
|
|
def test_cleanup_retains_original_inspection_exception(self):
|
|
self.owner.stop_deadline = 110
|
|
error = ProcessInspectionUnavailable()
|
|
with patch('qualification_observer.command_tokens', side_effect=error):
|
|
with self.assertRaises(ProcessInspectionUnavailable) as caught: self.inspection.argv(123)
|
|
self.assertIs(caught.exception, error)
|
|
|
|
def test_poll_cadence_and_final_observation(self):
|
|
child = SimpleNamespace(pid=123, poll=Mock(side_effect=[None, None, 0]))
|
|
sleep = Mock()
|
|
wait_for_owned_command(child, self.inspection, sleep=sleep)
|
|
self.assertEqual([c.args for c in sleep.call_args_list], [(.5,), (.5,)])
|
|
self.assertEqual(self.owner.observe.call_count, 3)
|
|
|
|
def test_observation_consuming_window_stops_without_sleep_or_next_poll(self):
|
|
child = SimpleNamespace(pid=123, poll=Mock(return_value=None))
|
|
self.owner.observe.side_effect = lambda pid: setattr(self, 'now', 190)
|
|
sleep = Mock()
|
|
with self.assertRaises(ActiveInspectionFailed): wait_for_owned_command(child, self.inspection, sleep=sleep)
|
|
sleep.assert_not_called()
|
|
self.assertEqual(child.poll.call_count, 1)
|
|
|
|
def test_failed_observation_still_retires_and_cannot_launch_next_test(self):
|
|
lifecycle = Lifecycle()
|
|
launched = []
|
|
stop = Mock()
|
|
child = SimpleNamespace(pid=123, poll=Mock(return_value=None))
|
|
# Match observe(pid)'s callback signature while using the actual adapter.
|
|
self.owner.observe.side_effect = lambda pid: self.inspection.table()
|
|
def run(command, label):
|
|
lifecycle.begin(); launched.append(label)
|
|
try: wait_for_owned_command(child, self.inspection, sleep=Mock())
|
|
finally: stop(child.pid); lifecycle.confirmed()
|
|
with patch('qualification_observer.process_table', side_effect=subprocess.TimeoutExpired('ps', 5)) as table:
|
|
with self.assertRaises(ActiveInspectionFailed):
|
|
run_test_cases([('startup', []), ('sdk', [])], run, lifecycle)
|
|
self.assertEqual(table.call_count, 1)
|
|
stop.assert_called_once_with(123)
|
|
self.assertEqual(launched, ['startup'])
|
|
self.assertTrue(lifecycle.safe)
|
|
|
|
def test_shared_process_inspection_defaults_remain_two_seconds(self):
|
|
with patch('owned_processes.subprocess.check_output', return_value='') as ps:
|
|
process_table()
|
|
self.assertEqual(ps.call_args.kwargs['timeout'], 2)
|
|
with patch('owned_processes.subprocess.check_output', return_value='fixture') as ps:
|
|
command_tokens(123)
|
|
self.assertEqual(ps.call_args.kwargs['timeout'], 2)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|