58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
import os
|
|
import signal
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from grm import call # noqa: E402
|
|
|
|
CHILD = ("import json, sys, time\n"
|
|
"p = sys.stdin.buffer.read().decode('utf-8')\n"
|
|
"print(json.dumps({'type': 'system', 'subtype': 'init', 'session_id': p}), flush=True)\n"
|
|
"if p == 'sleep': time.sleep(30)\n"
|
|
"print(json.dumps({'type': 'result', 'session_id': p}), flush=True)\n")
|
|
|
|
|
|
class RunClaudeTest(unittest.TestCase):
|
|
def _run(self, d, prompt):
|
|
pids = []
|
|
raw = Path(d) / "j.jsonl"
|
|
out = call.run_claude([sys.executable, "-c", CHILD], prompt, dict(os.environ), d,
|
|
raw, pids.append)
|
|
return out, raw, pids
|
|
|
|
def test_tee_and_parse(self):
|
|
"""B1: prompt goes in on stdin; the stream is parsed and teed to the raw file."""
|
|
with tempfile.TemporaryDirectory() as d:
|
|
(ss, code, stopped), raw, pids = self._run(d, "héllo")
|
|
self.assertEqual((code, stopped, len(pids)), (0, False, 1))
|
|
self.assertEqual(ss.session_id, "héllo")
|
|
self.assertIsNotNone(ss.result)
|
|
self.assertEqual(len(raw.read_bytes().splitlines()), 2)
|
|
|
|
def test_sigterm_stops_child(self):
|
|
"""B4: SIGTERM stops the child; stopped is set and the partial raw file is kept."""
|
|
with tempfile.TemporaryDirectory() as d:
|
|
timer = threading.Timer(1.5, signal.raise_signal, (signal.SIGTERM,))
|
|
t0 = time.time()
|
|
timer.start()
|
|
try:
|
|
(ss, code, stopped), raw, pids = self._run(d, "sleep")
|
|
finally:
|
|
timer.cancel()
|
|
self.assertTrue(stopped)
|
|
self.assertIsNotNone(code) # the child has exited
|
|
self.assertLess(time.time() - t0, 20)
|
|
self.assertIsNone(ss.result)
|
|
self.assertEqual(ss.session_id, "sleep")
|
|
self.assertIn(b'"init"', raw.read_bytes())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|