"""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.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']) 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_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()