fraujulian/MyAiConfiguration
1
0 Fork 0
Code Issues Pull Requests Actions Activity

test(clients): add smoke test timeouts and required-client mode

Browse Source 
Read client output on a thread with a deadline so a silent or exited client
fails instead of hanging, and fail on a missing CLI when
AI_CONFIG_REQUIRE_CLIENTS=1.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Julian lechnerandClaude Opus 5.5 committed 2026-10-01 11:02:05 +02:00
1 parent c14101e60d
commit 82391bdfbf
1 file changed
+60 -23
Download Patch File Download Diff File Expand all files Collapse all files
scripts/tests/test-client-smoke.py
+60 -23
View File
@@ -4,21 +4,67 @@ Each test installs a generated package into a temporary home that already holds
foreign settings, then asks the real client CLI what it discovered. Claude runs
with an invalid API key and is stopped at its init event; Codex is queried
through `codex app-server`. No model request completes.
Set AI_CONFIG_REQUIRE_CLIENTS=1 to fail instead of skip when a client CLI is missing.
"""
import json
import os
from pathlib import Path
import queue
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import unittest
ROOT = Path(__file__).resolve().parent.parent.parent
SHELL = 'powershell' if os.name == 'nt' else 'bash'
MARKER_SCRIPT = "import sys; open(sys.argv[1], 'w').write('ok')"
REQUIRE_CLIENTS = os.environ.get('AI_CONFIG_REQUIRE_CLIENTS') == '1'
INSTALL_TIMEOUT = 300
OUTPUT_TIMEOUT = 90
def client_cli(test, name):
path = shutil.which(name)
if path:
return path
if REQUIRE_CLIENTS:
test.fail(f'{name} CLI unavailable but AI_CONFIG_REQUIRE_CLIENTS=1')
test.skipTest(f'{name} CLI unavailable')
class LineReader:
"""Read process output lines on a thread so every wait has a deadline."""
def __init__(self, stream):
self.lines = queue.Queue()
threading.Thread(target=self._pump, args=(stream,), daemon=True).start()
def _pump(self, stream):
for line in stream:
self.lines.put(line)
self.lines.put(None)
def next(self, what):
try:
line = self.lines.get(timeout=OUTPUT_TIMEOUT)
except queue.Empty:
raise AssertionError(f'no output for {OUTPUT_TIMEOUT}s while waiting for {what}') from None
if line is None:
raise AssertionError(f'client exited while waiting for {what}')
return line
def stop(process):
process.kill()
for stream in (process.stdin, process.stdout):
if stream:
stream.close()
process.wait(timeout=30)
def frontmatter_names(directory, pattern):
@@ -56,7 +102,7 @@ def install(client, home):
'while IFS="|" read -r source destination; do '
'sync_managed_destination "$source" "$destination" smoke "$destination" bash "pwsh -NoProfile -File" false true >/dev/null; done')
args = ['bash', '-c', script]
result = subprocess.run(args, capture_output=True, text=True)
result = subprocess.run(args, capture_output=True, text=True, timeout=INSTALL_TIMEOUT)
if result.returncode:
raise RuntimeError(f"install failed: {result.stdout}{result.stderr}")
@@ -67,14 +113,14 @@ class ClientSmokeTests(unittest.TestCase):
self.home = self.base / 'home'
self.work = self.base / 'work'
self.work.mkdir(parents=True)
subprocess.run(['git', 'init', '-q'], cwd=self.work, check=True)
subprocess.run(['git', 'init', '-q'], cwd=self.work, check=True, timeout=60)
self.marker = self.base / 'foreign-hook.txt'
def tearDown(self):
shutil.rmtree(self.base, ignore_errors=True)
@unittest.skipUnless(shutil.which('claude'), 'Claude CLI unavailable')
def test_claude_discovers_package_and_keeps_foreign_settings(self):
claude = client_cli(self, 'claude')
package = ROOT / f'generated/claude-{SHELL}'
config = self.home / '.claude'
config.mkdir(parents=True)
@@ -82,23 +128,17 @@ class ClientSmokeTests(unittest.TestCase):
(config / 'settings.json').write_text(json.dumps({'foreignSetting': 'keep', 'hooks': {'SessionStart': [foreign_hook]}}))
install('Claude', self.home)
process = subprocess.Popen([shutil.which('claude'), '-p', 'smoke', '--output-format', 'stream-json', '--verbose', '--max-turns', '1'],
process = subprocess.Popen([claude, '-p', 'smoke', '--output-format', 'stream-json', '--verbose', '--max-turns', '1'],
cwd=self.work, env=clean_env(CLAUDE_CONFIG_DIR=str(config), ANTHROPIC_API_KEY='sk-ant-invalid-smoke'),
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, encoding='utf-8')
events = []
try:
for line in process.stdout:
events.append(json.loads(line))
if events[-1].get('subtype') == 'init':
break
reader = LineReader(process.stdout)
while not events or events[-1].get('subtype') != 'init':
events.append(json.loads(reader.next('the Claude init event')))
finally:
process.kill()
for stream in (process.stdin, process.stdout):
if stream:
stream.close()
process.wait(timeout=30)
init = next((event for event in events if event.get('subtype') == 'init'), None)
self.assertIsNotNone(init, 'Claude emitted no init event')
stop(process)
init = events[-1]
with self.subTest('skills'):
self.assertEqual(sorted(frontmatter_names(package / 'skills', 'SKILL.md') - set(init['skills'])), [])
@@ -112,8 +152,8 @@ class ClientSmokeTests(unittest.TestCase):
with self.subTest('foreign settings'):
self.assertEqual(json.loads((config / 'settings.json').read_text(encoding='utf-8-sig')).get('foreignSetting'), 'keep')
@unittest.skipUnless(shutil.which('codex'), 'Codex CLI unavailable')
def test_codex_discovers_package_and_keeps_foreign_settings(self):
codex = client_cli(self, 'codex')
package = ROOT / f'generated/codex-{SHELL}'
codex_home = self.home / '.codex'
codex_home.mkdir(parents=True)
@@ -123,15 +163,16 @@ class ClientSmokeTests(unittest.TestCase):
f'[[hooks.SessionStart]]\n[[hooks.SessionStart.hooks]]\ntype = "command"\ncommand = {command}\ntimeout = 10\n')
install('Codex', self.home)
process = subprocess.Popen([shutil.which('codex'), 'app-server'], cwd=self.work,
process = subprocess.Popen([codex, 'app-server'], cwd=self.work,
env=clean_env(CODEX_HOME=str(codex_home), OPENAI_API_KEY='sk-invalid-smoke'),
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, encoding='utf-8')
reader = LineReader(process.stdout)
def call(request_id, method, params):
process.stdin.write(json.dumps({'jsonrpc': '2.0', 'id': request_id, 'method': method, 'params': params}) + '\n')
process.stdin.flush()
while True:
message = json.loads(process.stdout.readline())
message = json.loads(reader.next(method))
if message.get('id') == request_id:
self.assertNotIn('error', message, method)
return message['result']
@@ -146,11 +187,7 @@ class ClientSmokeTests(unittest.TestCase):
hooks = call(4, 'hooks/list', {'cwds': [str(self.work)]})['data'][0]['hooks']
config = call(5, 'config/read', {})['config']
finally:
process.kill()
for stream in (process.stdin, process.stdout):
if stream:
stream.close()
process.wait(timeout=30)
stop(process)
isolated = {skill['name'] for skill in skills if Path(skill['path']).resolve().is_relative_to(skills_root.resolve())}
self.assertEqual(sorted(frontmatter_names(package / 'skills', 'SKILL.md') - isolated), [], 'skills not discovered')