wuyang6
882 lines
36 KiB
Python
882 lines
36 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 unittest.mock import patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.api.server import (
|
|
AgentState,
|
|
create_app,
|
|
_derive_initial_session_title,
|
|
_ensure_session_title,
|
|
_generate_session_title,
|
|
_runtime_event_stage,
|
|
_save_in_progress_session,
|
|
_sanitize_stored_session_for_resume,
|
|
)
|
|
from src.agent_types import AgentRunResult, AssistantTurn
|
|
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',
|
|
timeout_seconds=120.0,
|
|
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_update_accepts_provider_prefixed_model(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, _ = _build_client(Path(d))
|
|
updated = client.post(
|
|
'/api/state',
|
|
json={'model': 'azure_openai/gpt-4o-mini'},
|
|
)
|
|
self.assertEqual(updated.status_code, 200)
|
|
self.assertEqual(updated.json()['model'], 'azure_openai/gpt-4o-mini')
|
|
|
|
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:
|
|
root = Path(d)
|
|
skill_dir = root / 'skills' / 'project-skill'
|
|
skill_dir.mkdir(parents=True)
|
|
(skill_dir / 'SKILL.md').write_text(
|
|
(
|
|
'---\n'
|
|
'name: project-skill\n'
|
|
'description: Project skill.\n'
|
|
'---\n'
|
|
'Project skill body.\n'
|
|
),
|
|
encoding='utf-8',
|
|
)
|
|
client, _ = _build_client(root)
|
|
response = client.get('/api/skills')
|
|
self.assertEqual(response.status_code, 200)
|
|
skills = response.json()
|
|
self.assertTrue(skills)
|
|
by_name = {entry['name']: entry for entry in skills}
|
|
self.assertIn('project-skill', by_name)
|
|
self.assertTrue(by_name['project-skill']['enabled'])
|
|
self.assertTrue(by_name['project-skill']['configurable'])
|
|
self.assertNotIn('simplify', by_name)
|
|
|
|
def test_skill_preference_toggle_updates_prompt_filter(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
skill_dir = root / 'skills' / 'project-skill'
|
|
skill_dir.mkdir(parents=True)
|
|
(skill_dir / 'SKILL.md').write_text(
|
|
(
|
|
'---\n'
|
|
'name: project-skill\n'
|
|
'description: Project skill.\n'
|
|
'---\n'
|
|
'Project skill body.\n'
|
|
),
|
|
encoding='utf-8',
|
|
)
|
|
client, state = _build_client(root)
|
|
|
|
response = client.patch(
|
|
'/api/skills',
|
|
json={'skill': 'project-skill', 'enabled': False},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
skill = {
|
|
entry['name']: entry
|
|
for entry in response.json()
|
|
}['project-skill']
|
|
self.assertFalse(skill['enabled'])
|
|
self.assertNotIn('project-skill', state.agent_for().runtime_config.enabled_skill_names)
|
|
self.assertIn('verify', state.agent_for().runtime_config.enabled_skill_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_latest_run_includes_runtime_event_history(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
client, state = _build_client(Path(d))
|
|
record = state.run_manager.start('alice', 'thread-1')
|
|
state.run_manager.update(record.run_id, status='running')
|
|
state.run_manager.record_event(
|
|
record.run_id,
|
|
{
|
|
'type': 'tool_start',
|
|
'tool_name': 'python_exec',
|
|
'tool_call_id': 'call-1',
|
|
'arguments': {'code': 'print(1)'},
|
|
'assistant_content': '我用 Python 分析一下。',
|
|
},
|
|
)
|
|
state.run_manager.record_event(
|
|
record.run_id,
|
|
{
|
|
'type': 'tool_result',
|
|
'tool_name': 'python_exec',
|
|
'tool_call_id': 'call-1',
|
|
'ok': True,
|
|
'metadata': {'output_preview': 'exit_code=0'},
|
|
},
|
|
)
|
|
|
|
latest = client.get(
|
|
'/api/runs/latest',
|
|
params={'account_id': 'alice', 'session_id': 'thread-1'},
|
|
)
|
|
|
|
self.assertEqual(latest.status_code, 200)
|
|
events = latest.json()['events']
|
|
self.assertEqual([event['type'] for event in events], ['tool_start', 'tool_result'])
|
|
self.assertEqual(events[0]['tool_name'], 'python_exec')
|
|
self.assertEqual(events[1]['metadata']['output_preview'], 'exit_code=0')
|
|
|
|
def test_cancel_run_marks_streaming_session_tail_cancelled(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, state = _build_client(root)
|
|
session_root = root / 'accounts' / 'alice' / 'sessions'
|
|
save_agent_session(
|
|
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'},
|
|
{
|
|
'role': 'assistant',
|
|
'content': '我先看看文件',
|
|
'tool_calls': [
|
|
{
|
|
'id': 'call-1',
|
|
'function': {'name': 'list_dir', 'arguments': '{}'},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
'role': 'tool',
|
|
'content': '',
|
|
'tool_call_id': 'call-1',
|
|
'state': 'streaming',
|
|
'metadata': {'phase': 'starting'},
|
|
},
|
|
),
|
|
turns=1,
|
|
tool_calls=1,
|
|
usage={},
|
|
total_cost_usd=0.0,
|
|
file_history=(),
|
|
budget_state={'status': 'running'},
|
|
plugin_state={},
|
|
),
|
|
directory=session_root,
|
|
)
|
|
state.run_manager.start('alice', 'thread-1')
|
|
|
|
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'])
|
|
stored = load_agent_session('thread-1', directory=session_root)
|
|
self.assertEqual(stored.messages[-1]['role'], 'assistant')
|
|
self.assertEqual(stored.messages[-1]['stop_reason'], 'cancelled')
|
|
self.assertEqual(
|
|
stored.messages[-1]['metadata']['status'],
|
|
'cancelled',
|
|
)
|
|
self.assertNotEqual(stored.messages[-1].get('state'), 'streaming')
|
|
|
|
def test_latest_run_reports_interrupted_for_orphaned_streaming_session(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
client, _ = _build_client(root)
|
|
session_root = root / 'accounts' / 'alice' / 'sessions'
|
|
save_agent_session(
|
|
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'},
|
|
{
|
|
'role': 'tool',
|
|
'content': '',
|
|
'tool_call_id': 'call-1',
|
|
'state': 'streaming',
|
|
'metadata': {'phase': 'starting'},
|
|
},
|
|
),
|
|
turns=1,
|
|
tool_calls=1,
|
|
usage={},
|
|
total_cost_usd=0.0,
|
|
file_history=(),
|
|
budget_state={'status': 'running'},
|
|
plugin_state={},
|
|
),
|
|
directory=session_root,
|
|
)
|
|
|
|
response = client.get(
|
|
'/api/runs/latest',
|
|
params={'account_id': 'alice', 'session_id': 'thread-1'},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()['status'], 'interrupted')
|
|
|
|
def test_resume_sanitizer_removes_ui_run_status_messages(self) -> None:
|
|
stored = StoredAgentSession(
|
|
session_id='thread-1',
|
|
model_config={'model': 'test-model'},
|
|
runtime_config={},
|
|
system_prompt_parts=('system prompt',),
|
|
user_context={},
|
|
system_context={},
|
|
messages=(
|
|
{'role': 'user', 'content': 'hello'},
|
|
{
|
|
'role': 'assistant',
|
|
'content': '上一次任务已中断，后台没有正在执行的进程。',
|
|
'state': 'final',
|
|
'stop_reason': 'interrupted',
|
|
'metadata': {'kind': 'run_status', 'status': 'interrupted'},
|
|
},
|
|
{'role': 'user', 'content': 'continue'},
|
|
),
|
|
display_messages=(
|
|
{'role': 'user', 'content': 'hello'},
|
|
{
|
|
'role': 'assistant',
|
|
'content': '上一次任务已中断，后台没有正在执行的进程。',
|
|
'state': 'final',
|
|
'stop_reason': 'interrupted',
|
|
'metadata': {'kind': 'run_status', 'status': 'interrupted'},
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'content': 'pending',
|
|
'metadata': {'status': 'running', 'placeholder': True},
|
|
},
|
|
),
|
|
turns=1,
|
|
tool_calls=0,
|
|
usage={},
|
|
total_cost_usd=0.0,
|
|
file_history=(),
|
|
budget_state={'status': 'interrupted'},
|
|
plugin_state={},
|
|
)
|
|
|
|
sanitized = _sanitize_stored_session_for_resume(stored)
|
|
|
|
self.assertEqual(
|
|
[message['content'] for message in sanitized.messages],
|
|
['hello', 'continue'],
|
|
)
|
|
self.assertEqual(
|
|
[message['content'] for message in sanitized.display_messages],
|
|
['hello'],
|
|
)
|
|
|
|
def test_runtime_event_stage_names_long_running_work(self) -> None:
|
|
self.assertEqual(
|
|
_runtime_event_stage({'type': 'tool_start', 'tool_name': 'python_exec'}),
|
|
'调用工具 python_exec',
|
|
)
|
|
self.assertEqual(
|
|
_runtime_event_stage({'type': 'final_text_start'}),
|
|
'正在整理回复',
|
|
)
|
|
self.assertEqual(_runtime_event_stage({'type': 'server_heartbeat'}), '')
|
|
|
|
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_in_progress_session_writes_first_message_title(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
_, state = _build_client(root)
|
|
session_id = 'first-title'
|
|
directory = state.account_paths('alice')['sessions']
|
|
agent = state.agent_for('alice', session_id)
|
|
|
|
title = _save_in_progress_session(
|
|
directory=directory,
|
|
agent=agent,
|
|
session_id=session_id,
|
|
prompt='帮我分析一下线上 router session 数据，然后统计 device 分布。',
|
|
)
|
|
|
|
session_file = directory / session_id / 'session.json'
|
|
data = json.loads(session_file.read_text(encoding='utf-8'))
|
|
self.assertTrue(title.startswith('帮我分析一下线上 router'))
|
|
self.assertEqual(data['title'], title)
|
|
self.assertEqual(data['title_source'], 'first_message')
|
|
|
|
def test_session_title_refines_first_message_title_after_run(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
_, state = _build_client(root)
|
|
session_id = 'refine-title'
|
|
directory = state.account_paths('alice')['sessions']
|
|
agent = state.agent_for('alice', session_id)
|
|
fallback_title = _save_in_progress_session(
|
|
directory=directory,
|
|
agent=agent,
|
|
session_id=session_id,
|
|
prompt='帮我生成地图和餐饮服务的边界数据。',
|
|
)
|
|
session_file = directory / session_id / 'session.json'
|
|
data = json.loads(session_file.read_text(encoding='utf-8'))
|
|
data['messages'] = [
|
|
{'role': 'user', 'content': '帮我生成地图和餐饮服务的边界数据。'}
|
|
]
|
|
session_file.write_text(json.dumps(data), encoding='utf-8')
|
|
|
|
with patch(
|
|
'backend.api.server._generate_session_title',
|
|
return_value='地图餐饮边界数据',
|
|
) as generate:
|
|
_ensure_session_title(
|
|
directory,
|
|
session_id,
|
|
model='test-model',
|
|
base_url='http://127.0.0.1:8000/v1',
|
|
api_key='local-token',
|
|
fallback_title=fallback_title,
|
|
)
|
|
|
|
generate.assert_called_once()
|
|
updated = json.loads(session_file.read_text(encoding='utf-8'))
|
|
self.assertEqual(updated['title'], '地图餐饮边界数据')
|
|
self.assertEqual(updated['title_source'], 'llm')
|
|
|
|
def test_session_title_preserves_existing_manual_title(self) -> None:
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root = Path(d)
|
|
_, state = _build_client(root)
|
|
session_id = 'manual-preserved'
|
|
directory = state.account_paths('alice')['sessions']
|
|
session_dir = directory / session_id
|
|
session_dir.mkdir(parents=True)
|
|
session_file = session_dir / 'session.json'
|
|
session_file.write_text(
|
|
json.dumps(
|
|
{
|
|
'session_id': session_id,
|
|
'title': '手动标题',
|
|
'title_source': 'manual',
|
|
'messages': [
|
|
{'role': 'user', 'content': '帮我分析数据。'},
|
|
],
|
|
}
|
|
),
|
|
encoding='utf-8',
|
|
)
|
|
|
|
with patch('backend.api.server._generate_session_title') as generate:
|
|
_ensure_session_title(
|
|
directory,
|
|
session_id,
|
|
model='test-model',
|
|
base_url='http://127.0.0.1:8000/v1',
|
|
api_key='local-token',
|
|
previous_title='手动标题',
|
|
)
|
|
|
|
generate.assert_not_called()
|
|
updated = json.loads(session_file.read_text(encoding='utf-8'))
|
|
self.assertEqual(updated['title'], '手动标题')
|
|
|
|
def test_derive_initial_session_title_strips_session_context(self) -> None:
|
|
title = _derive_initial_session_title(
|
|
'请帮我生成线上挖掘数据。\n\n[当前会话目录]\n/root/session'
|
|
)
|
|
self.assertEqual(title, '请帮我生成线上挖掘数据')
|
|
|
|
def test_generate_session_title_uses_model_compat_client(self) -> None:
|
|
with patch('backend.api.server.OpenAICompatClient') as client_cls:
|
|
client = client_cls.return_value
|
|
client.complete.return_value = AssistantTurn(content='“模型兼容标题”')
|
|
|
|
title = _generate_session_title(
|
|
['帮我分析一下线上 session 数据'],
|
|
model='ppio/pa/claude-opus-4-7',
|
|
base_url='http://model.mify.ai.srv/v1',
|
|
api_key='local-token',
|
|
)
|
|
|
|
self.assertEqual(title, '模型兼容标题')
|
|
config = client_cls.call_args.args[0]
|
|
self.assertEqual(config.model, 'ppio/pa/claude-opus-4-7')
|
|
self.assertEqual(config.base_url, 'http://model.mify.ai.srv/v1')
|
|
client.complete.assert_called_once()
|
|
|
|
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()
|