武阳
490 lines
20 KiB
Python
490 lines
20 KiB
Python
"""Integration tests for the local web GUI FastAPI server.
|
|
|
|
These tests exercise the JSON endpoints against a real :class:`AgentState`
|
|
without booting uvicorn, using ``fastapi.testclient.TestClient``. Slash
|
|
commands are dispatched locally inside :class:`LocalCodingAgent` and never
|
|
hit the network, so the chat endpoint can be exercised end-to-end against
|
|
``/help``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.api.server import AgentState, create_app
|
|
from src.agent_types import AgentRunResult
|
|
from src.session_store import StoredAgentSession, load_agent_session, save_agent_session
|
|
|
|
|
|
def _build_client(tmp: Path) -> tuple[TestClient, AgentState]:
|
|
state = AgentState(
|
|
cwd=tmp,
|
|
model='test-model',
|
|
base_url='http://127.0.0.1:8000/v1',
|
|
api_key='local-token',
|
|
allow_shell=False,
|
|
allow_write=False,
|
|
session_directory=tmp / 'sessions',
|
|
)
|
|
return TestClient(create_app(state)), state
|
|
|
|
|
|
class GuiServerTests(unittest.TestCase):
|
|
def test_root_serves_html(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.get('/')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('Claw Code', response.text)
|
|
|
|
def test_static_assets_served(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
self.assertEqual(client.get('/static/app.css').status_code, 200)
|
|
self.assertEqual(client.get('/static/app.js').status_code, 200)
|
|
|
|
def test_state_snapshot_round_trip(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.get('/api/state')
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload['model'], 'test-model')
|
|
self.assertFalse(payload['allow_shell'])
|
|
|
|
updated = client.post(
|
|
'/api/state',
|
|
json={'allow_shell': True, 'model': 'other-model'},
|
|
)
|
|
self.assertEqual(updated.status_code, 200)
|
|
data = updated.json()
|
|
self.assertTrue(data['allow_shell'])
|
|
self.assertEqual(data['model'], 'other-model')
|
|
|
|
def test_state_snapshot_can_scope_to_account(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.get('/api/state', params={'account_id': 'alice'})
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload['account_id'], 'alice')
|
|
self.assertIn('/accounts/alice/sessions', payload['session_directory'])
|
|
self.assertIn('/accounts/alice/sessions', payload['upload_directory'])
|
|
self.assertIn('/accounts/alice/python/.venv', payload['python_env_directory'])
|
|
self.assertTrue((Path(payload['python_env_directory']) / 'bin' / 'python').exists())
|
|
|
|
def test_state_update_is_scoped_to_account(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
updated = client.post(
|
|
'/api/state',
|
|
json={'account_id': 'alice', 'allow_shell': True, 'model': 'alice-model'},
|
|
)
|
|
self.assertEqual(updated.status_code, 200)
|
|
self.assertTrue(updated.json()['allow_shell'])
|
|
self.assertEqual(updated.json()['model'], 'alice-model')
|
|
|
|
alice = client.get('/api/state', params={'account_id': 'alice'}).json()
|
|
bob = client.get('/api/state', params={'account_id': 'bob'}).json()
|
|
default = client.get('/api/state').json()
|
|
self.assertEqual(alice['model'], 'alice-model')
|
|
self.assertTrue(alice['allow_shell'])
|
|
self.assertEqual(bob['model'], 'test-model')
|
|
self.assertFalse(bob['allow_shell'])
|
|
self.assertEqual(default['model'], 'test-model')
|
|
self.assertFalse(default['allow_shell'])
|
|
|
|
def test_run_locks_are_scoped_to_account(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
_, state = _build_client(Path(d))
|
|
self.assertIs(state.run_lock_for('alice'), state.run_lock_for('alice'))
|
|
self.assertIsNot(state.run_lock_for('alice'), state.run_lock_for('bob'))
|
|
self.assertIs(
|
|
state.run_lock_for('alice', 's1'),
|
|
state.run_lock_for('alice', 's1'),
|
|
)
|
|
self.assertIsNot(
|
|
state.run_lock_for('alice', 's1'),
|
|
state.run_lock_for('alice', 's2'),
|
|
)
|
|
|
|
def test_state_update_rejects_missing_cwd(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.post(
|
|
'/api/state',
|
|
json={'cwd': str(Path(d) / 'does-not-exist')},
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_slash_commands_listed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.get('/api/slash-commands')
|
|
self.assertEqual(response.status_code, 200)
|
|
commands = response.json()
|
|
self.assertTrue(commands)
|
|
primaries = {entry['primary'] for entry in commands}
|
|
self.assertIn('help', primaries)
|
|
|
|
def test_skills_listed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.get('/api/skills')
|
|
self.assertEqual(response.status_code, 200)
|
|
skills = response.json()
|
|
self.assertTrue(skills)
|
|
names = {entry['name'] for entry in skills}
|
|
self.assertIn('simplify', names)
|
|
|
|
def test_chat_runs_local_slash_command(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.post('/api/chat', json={'prompt': '/help'})
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload['turns'], 0)
|
|
self.assertEqual(payload['tool_calls'], 0)
|
|
self.assertIn('slash commands', payload['final_output'].lower())
|
|
self.assertIn('/help', payload['final_output'])
|
|
self.assertIn('total_tokens', payload['usage'])
|
|
|
|
def test_chat_uses_requested_session_directory(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, _ = _build_client(root)
|
|
response = client.post(
|
|
'/api/chat',
|
|
json={
|
|
'prompt': '/help',
|
|
'account_id': 'alice',
|
|
'session_id': 'thread-1',
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertIsNone(payload['session_id'])
|
|
|
|
def test_chat_persists_in_progress_session_before_agent_finishes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, state = _build_client(root)
|
|
agent = state.agent_for('alice', 'pending-1')
|
|
|
|
def fake_run(
|
|
prompt: str,
|
|
session_id: str | None = None,
|
|
*,
|
|
runtime_context: str | None = None,
|
|
event_sink: object | None = None,
|
|
) -> AgentRunResult:
|
|
stored = load_agent_session(
|
|
'pending-1',
|
|
directory=root / 'accounts' / 'alice' / 'sessions',
|
|
)
|
|
self.assertEqual(stored.session_id, 'pending-1')
|
|
self.assertEqual(stored.budget_state.get('status'), 'running')
|
|
self.assertEqual(stored.messages[-1]['role'], 'user')
|
|
self.assertEqual(stored.messages[-1]['content'], 'hello')
|
|
return AgentRunResult(
|
|
final_output='done',
|
|
turns=1,
|
|
tool_calls=0,
|
|
transcript=stored.messages
|
|
+ ({'role': 'assistant', 'content': 'done'},),
|
|
session_id=session_id,
|
|
)
|
|
|
|
original_run = agent.run
|
|
agent.run = fake_run # type: ignore[method-assign]
|
|
try:
|
|
response = client.post(
|
|
'/api/chat',
|
|
json={
|
|
'prompt': 'hello',
|
|
'account_id': 'alice',
|
|
'session_id': ' pending-1 ',
|
|
},
|
|
)
|
|
finally:
|
|
agent.run = original_run # type: ignore[method-assign]
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()['session_id'], 'pending-1')
|
|
|
|
def test_cancel_run_marks_latest_session_cancelled(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, state = _build_client(Path(d))
|
|
record = state.run_manager.start('alice', 'thread-1')
|
|
self.assertFalse(record.cancel_event.is_set())
|
|
|
|
response = client.post(
|
|
'/api/runs/cancel',
|
|
json={'account_id': 'alice', 'session_id': 'thread-1'},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json()['cancelled'])
|
|
self.assertTrue(record.cancel_event.is_set())
|
|
latest = client.get(
|
|
'/api/runs/latest',
|
|
params={'account_id': 'alice', 'session_id': 'thread-1'},
|
|
)
|
|
self.assertEqual(latest.status_code, 200)
|
|
self.assertEqual(latest.json()['status'], 'cancelled')
|
|
|
|
def test_latest_run_prefers_running_over_queued_for_session(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, state = _build_client(Path(d))
|
|
first = state.run_manager.start('alice', 'thread-1')
|
|
state.run_manager.update(first.run_id, status='running', stage='first stage')
|
|
second = state.run_manager.start('alice', 'thread-1')
|
|
self.assertEqual(second.status, 'queued')
|
|
|
|
latest = client.get(
|
|
'/api/runs/latest',
|
|
params={'account_id': 'alice', 'session_id': 'thread-1'},
|
|
)
|
|
|
|
self.assertEqual(latest.status_code, 200)
|
|
payload = latest.json()
|
|
self.assertEqual(payload['run_id'], first.run_id)
|
|
self.assertEqual(payload['status'], 'running')
|
|
self.assertEqual(payload['current_stage'], 'first stage')
|
|
|
|
def test_chat_uses_account_scoped_model_config(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
client.post(
|
|
'/api/state',
|
|
json={'account_id': 'alice', 'model': 'alice-model'},
|
|
)
|
|
|
|
response = client.post(
|
|
'/api/chat',
|
|
json={'prompt': '/status', 'account_id': 'alice'},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('- Model: alice-model', response.json()['final_output'])
|
|
|
|
bob = client.post(
|
|
'/api/chat',
|
|
json={'prompt': '/status', 'account_id': 'bob'},
|
|
)
|
|
self.assertEqual(bob.status_code, 200)
|
|
self.assertIn('- Model: test-model', bob.json()['final_output'])
|
|
|
|
def test_agent_session_store_uses_session_subdirectory(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
stored = StoredAgentSession(
|
|
session_id='thread-1',
|
|
model_config={'model': 'test-model'},
|
|
runtime_config={'cwd': str(root)},
|
|
system_prompt_parts=(),
|
|
user_context={},
|
|
system_context={},
|
|
messages=({'role': 'user', 'content': 'hello'},),
|
|
turns=1,
|
|
tool_calls=0,
|
|
usage={},
|
|
total_cost_usd=0.0,
|
|
file_history=(),
|
|
budget_state={},
|
|
plugin_state={},
|
|
scratchpad_directory=str(root / 'sessions' / 'thread-1' / 'scratchpad'),
|
|
)
|
|
path = save_agent_session(stored, directory=root / 'sessions')
|
|
self.assertEqual(path, root / 'sessions' / 'thread-1' / 'session.json')
|
|
loaded = load_agent_session('thread-1', directory=root / 'sessions')
|
|
self.assertEqual(loaded.session_id, 'thread-1')
|
|
|
|
def test_context_budget_reports_stored_session_prompt_size(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, _ = _build_client(root)
|
|
session_root = root / 'accounts' / 'alice' / 'sessions'
|
|
stored = StoredAgentSession(
|
|
session_id='thread-1',
|
|
model_config={'model': 'test-model'},
|
|
runtime_config={'cwd': str(root)},
|
|
system_prompt_parts=('system prompt',),
|
|
user_context={},
|
|
system_context={},
|
|
messages=({'role': 'user', 'content': 'hello'},),
|
|
turns=1,
|
|
tool_calls=0,
|
|
usage={},
|
|
total_cost_usd=0.0,
|
|
file_history=(),
|
|
budget_state={},
|
|
plugin_state={},
|
|
)
|
|
save_agent_session(stored, directory=session_root)
|
|
|
|
response = client.get(
|
|
'/api/context-budget',
|
|
params={'account_id': 'alice', 'session_id': 'thread-1'},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload['model'], 'test-model')
|
|
self.assertEqual(payload['context_window_tokens'], 128000)
|
|
self.assertGreater(payload['projected_input_tokens'], 0)
|
|
self.assertGreater(payload['soft_input_limit_tokens'], 0)
|
|
|
|
def test_chat_rejects_blank_prompt(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.post('/api/chat', json={'prompt': ' '})
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_chat_resume_unknown_session_returns_404(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.post(
|
|
'/api/chat',
|
|
json={'prompt': '/help', 'resume_session_id': 'missing'},
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_sessions_list_empty_when_directory_absent(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.get('/api/sessions')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json(), [])
|
|
|
|
def test_sessions_are_isolated_by_account(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, _ = _build_client(root)
|
|
alice_sessions = root / 'accounts' / 'alice' / 'sessions'
|
|
alice_sessions.mkdir(parents=True)
|
|
(alice_sessions / 's1.json').write_text(
|
|
json.dumps(
|
|
{
|
|
'session_id': 's1',
|
|
'turns': 1,
|
|
'tool_calls': 0,
|
|
'messages': [{'role': 'user', 'content': 'alice only'}],
|
|
}
|
|
),
|
|
encoding='utf-8',
|
|
)
|
|
|
|
alice = client.get('/api/sessions', params={'account_id': 'alice'})
|
|
bob = client.get('/api/sessions', params={'account_id': 'bob'})
|
|
self.assertEqual(alice.status_code, 200)
|
|
self.assertEqual(bob.status_code, 200)
|
|
self.assertEqual(alice.json()[0]['session_id'], 's1')
|
|
self.assertEqual(bob.json(), [])
|
|
|
|
def test_session_list_deduplicates_legacy_and_nested_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, _ = _build_client(root)
|
|
session_root = root / 'accounts' / 'alice' / 'sessions'
|
|
session_root.mkdir(parents=True)
|
|
legacy = {
|
|
'session_id': 'same-id',
|
|
'turns': 1,
|
|
'tool_calls': 0,
|
|
'messages': [{'role': 'user', 'content': 'legacy'}],
|
|
}
|
|
nested = {
|
|
'session_id': 'same-id',
|
|
'turns': 2,
|
|
'tool_calls': 1,
|
|
'title': '自动生成标题',
|
|
'messages': [{'role': 'user', 'content': 'nested'}],
|
|
}
|
|
(session_root / 'same-id.json').write_text(
|
|
json.dumps(legacy),
|
|
encoding='utf-8',
|
|
)
|
|
(session_root / 'same-id').mkdir()
|
|
(session_root / 'same-id' / 'session.json').write_text(
|
|
json.dumps(nested),
|
|
encoding='utf-8',
|
|
)
|
|
|
|
response = client.get('/api/sessions', params={'account_id': 'alice'})
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(len(payload), 1)
|
|
self.assertEqual(payload[0]['session_id'], 'same-id')
|
|
self.assertEqual(payload[0]['preview'], '自动生成标题')
|
|
|
|
def test_session_detail_404_when_missing(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.get('/api/sessions/nope')
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_delete_session_removes_nested_and_legacy_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, _ = _build_client(root)
|
|
session_root = root / 'accounts' / 'alice' / 'sessions'
|
|
nested = session_root / 'same-id'
|
|
nested.mkdir(parents=True)
|
|
(nested / 'session.json').write_text('{}', encoding='utf-8')
|
|
legacy = session_root / 'same-id.json'
|
|
legacy.write_text('{}', encoding='utf-8')
|
|
|
|
response = client.delete(
|
|
'/api/sessions/same-id',
|
|
params={'account_id': 'alice'},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json()['deleted'])
|
|
self.assertFalse(nested.exists())
|
|
self.assertFalse(legacy.exists())
|
|
|
|
missing = client.delete(
|
|
'/api/sessions/same-id',
|
|
params={'account_id': 'alice'},
|
|
)
|
|
self.assertEqual(missing.status_code, 404)
|
|
|
|
def test_update_session_title(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, _ = _build_client(root)
|
|
session_root = root / 'accounts' / 'alice' / 'sessions'
|
|
nested = session_root / 'same-id'
|
|
nested.mkdir(parents=True)
|
|
session_file = nested / 'session.json'
|
|
session_file.write_text(
|
|
json.dumps({'session_id': 'same-id', 'messages': []}),
|
|
encoding='utf-8',
|
|
)
|
|
|
|
response = client.patch(
|
|
'/api/sessions/same-id',
|
|
params={'account_id': 'alice'},
|
|
json={'title': ' 手动标题 '},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()['title'], '手动标题')
|
|
data = json.loads(session_file.read_text(encoding='utf-8'))
|
|
self.assertEqual(data['title'], '手动标题')
|
|
self.assertEqual(data['title_source'], 'manual')
|
|
|
|
def test_clear_runtime_state(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
response = client.post('/api/clear')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('model', response.json())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|