| """Opt-in SDK/Daemon lifecycle test; isolated from the production Daemon/device.""" |
| import json |
| import os |
| from pathlib import Path |
| import re |
| import sys |
| import tempfile |
| import time |
| import unittest |
| from unittest.mock import patch |
| from urllib.request import urlopen |
|
|
| from watcherobot.cli import ensure_runtime, stop_runtime, _request_json |
|
|
|
|
| @unittest.skipUnless(os.environ.get('WATCHER_CODEX_RUN_INTEGRATION') == '1', 'Set WATCHER_CODEX_RUN_INTEGRATION=1 for isolated managed-runtime test') |
| class ManagedRuntimeTests(unittest.TestCase): |
| def test_real_daemon_start_three_chats_three_rounds_restart_stop(self): |
| with tempfile.TemporaryDirectory(prefix='codex-sdk-acceptance-') as temporary: |
| base = Path(temporary) |
| codex = base / 'codex-fixture' |
| sessions = codex / 'sessions' |
| sessions.mkdir(parents=True) |
| paths = [] |
| for n in range(3): |
| path = sessions / f'rollout-fixture-{n}.jsonl' |
| path.write_text(json.dumps({'type': 'session_meta', 'payload': {'id': f'fixture-{n}', 'cwd': 'C:/fixture'}}) + '\n', encoding='utf-8') |
| paths.append(path) |
| (codex / 'session_index.jsonl').write_text(''.join(json.dumps({'id': f'fixture-{n}', 'thread_name': f'圆环任务{n + 1}'}, ensure_ascii=False) + '\n' for n in range(3)), encoding='utf-8') |
| |
| |
| settings = {'WATCHER_RUNTIME_STATE_ROOT': str(base / 'runtime'), 'WATCHER_RUNTIME_INSTANCE_ROOT': str(base / 'instance'), 'WATCHER_RUNTIME_CONTROL_PORT': '0', 'WATCHER_CODEX_NO_BROWSER': '1', 'WATCHER_CODEX_DATA_HOME': str(codex)} |
| with patch.dict(os.environ, settings): |
| runtime, reused = ensure_runtime(state_root=base / 'runtime', ephemeral_ports=True) |
| self.assertFalse(reused) |
| try: |
| app_root = Path(__file__).resolve().parents[1] |
| def control(path, payload=None): |
| return _request_json(runtime.control_url, path, method='POST', payload=payload, timeout=15) |
| control('/daemon/application/select', {'application_dir': str(app_root), 'launcher': {'kind': 'python', 'executable': sys.executable}}) |
| control('/daemon/application/start') |
| log_path = base / 'runtime' / 'logs' / 'applications' / 'com.orulink.codex_supervisor.jsonl' |
| def wait_preview(): |
| end = time.monotonic() + 12 |
| while time.monotonic() < end: |
| text = log_path.read_text(encoding='utf-8') if log_path.exists() else '' |
| urls = re.findall(r'Codex Supervisor preview: (http://127\.0\.0\.1:\d+)', text) |
| if urls: |
| try: |
| with urlopen(urls[-1] + '/api/health', timeout=1) as response: |
| if json.load(response)['ok']: |
| return urls[-1] |
| except OSError: |
| pass |
| time.sleep(.1) |
| self.fail('Managed Application did not expose a healthy preview') |
| url = wait_preview() |
| def wait_state(expected): |
| end = time.monotonic() + 8 |
| while time.monotonic() < end: |
| with urlopen(url + '/api/state', timeout=2) as response: |
| state = json.load(response) |
| if len(state['sessions']) == 3 and all(item['status'] == expected for item in state['sessions']): |
| self.assertNotEqual(state['device']['state'], 'connected') |
| return state |
| time.sleep(.1) |
| self.fail('Three-chat lifecycle did not converge') |
| for round_number in range(1, 4): |
| for action, expected in (('task_started', 'running'), ('task_complete', 'waiting')): |
| for path in paths: |
| payload = {'type': action, 'turn_id': f'round-{round_number}', 'started_at' if action == 'task_started' else 'completed_at': time.time()} |
| with path.open('a', encoding='utf-8') as output: |
| output.write(json.dumps({'type': 'event_msg', 'payload': payload}) + '\n') |
| state = wait_state(expected) |
| self.assertEqual([item['roundCountToday'] for item in state['sessions']], [round_number] * 3) |
| control('/daemon/application/restart') |
| url = wait_preview() |
| self.assertEqual(len(wait_state('waiting')['sessions']), 3) |
| control('/daemon/application/stop') |
| status = _request_json(runtime.control_url, '/daemon/status')['application']['state'] |
| self.assertEqual(status, 'ended') |
| print('Managed SDK acceptance: 3 chats x 3 rounds, restart and graceful stop passed (device offline).') |
| finally: |
| stop_runtime() |
| time.sleep(.5) |
|
|
|
|
| if __name__ == '__main__': |
| unittest.main() |
|
|
