test(api): verify bot response and service delegation contracts

This commit is contained in:
huanghuoguoguo
2026-10-17 15:53:00 +08:00
parent c05aac538f
commit cbc049400c
+63 -7
View File
@@ -125,7 +125,7 @@ def fake_bot_app():
'webhook_url': 'https://example.com/webhook/test-bot-uuid',
}
)
app.bot_service.create_bot = AsyncMock(return_value={'uuid': 'new-bot-uuid'})
app.bot_service.create_bot = AsyncMock(return_value='new-bot-uuid')
app.bot_service.update_bot = AsyncMock(return_value={})
app.bot_service.delete_bot = AsyncMock()
app.bot_service.list_event_logs = AsyncMock(return_value=([{'uuid': 'log-1', 'message': 'test log'}], 1))
@@ -179,6 +179,11 @@ def fake_bot_app():
return app
@pytest.fixture(autouse=True)
def reset_service_calls(fake_bot_app):
fake_bot_app.bot_service.reset_mock()
@pytest.fixture(scope='module')
async def quart_test_client(fake_bot_app, http_controller_cls):
"""Create Quart test client (module scope to avoid route re-registration)."""
@@ -194,7 +199,7 @@ class TestBotEndpoints:
"""Tests for /api/v1/platform/bots endpoints."""
@pytest.mark.asyncio
async def test_get_bots_success(self, quart_test_client):
async def test_get_bots_success(self, quart_test_client, fake_bot_app):
"""GET /api/v1/platform/bots returns bot list."""
response = await quart_test_client.get('/api/v1/platform/bots', headers={'Authorization': 'Bearer test_token'})
@@ -204,8 +209,17 @@ class TestBotEndpoints:
assert 'data' in data
assert 'bots' in data['data']
called = fake_bot_app.bot_service.get_bots
assert called.await_count == 1
assert called.await_args.args[0].workspace_uuid == 'workspace-test'
assert called.await_args.args[1:] == ()
assert called.await_args.kwargs == {'include_secret': True}
assert data['data']['bots'] == [
{'uuid': 'test-bot-uuid', 'name': 'Test Bot', 'platform': 'telegram', 'pipeline_uuid': 'test-pipeline-uuid'}
]
@pytest.mark.asyncio
async def test_create_bot_success(self, quart_test_client):
async def test_create_bot_success(self, quart_test_client, fake_bot_app):
"""POST /api/v1/platform/bots creates new bot."""
response = await quart_test_client.post(
'/api/v1/platform/bots',
@@ -218,8 +232,17 @@ class TestBotEndpoints:
assert data['code'] == 0
assert 'uuid' in data['data']
called = fake_bot_app.bot_service.create_bot
assert called.await_count == 1
assert called.await_args.args[0].workspace_uuid == 'workspace-test'
assert called.await_args.args[1:] == (
{'name': 'New Bot', 'platform': 'telegram', 'pipeline_uuid': 'test-pipeline'},
)
assert called.await_args.kwargs == {}
assert data['data'] == {'uuid': 'new-bot-uuid'}
@pytest.mark.asyncio
async def test_get_single_bot_success(self, quart_test_client):
async def test_get_single_bot_success(self, quart_test_client, fake_bot_app):
"""GET /api/v1/platform/bots/{uuid} returns bot with runtime info."""
response = await quart_test_client.get(
'/api/v1/platform/bots/test-bot-uuid', headers={'Authorization': 'Bearer test_token'}
@@ -230,8 +253,21 @@ class TestBotEndpoints:
assert data['code'] == 0
assert 'bot' in data['data']
called = fake_bot_app.bot_service.get_runtime_bot_info
assert called.await_count == 1
assert called.await_args.args[0].workspace_uuid == 'workspace-test'
assert called.await_args.args[1:] == ('test-bot-uuid',)
assert called.await_args.kwargs == {'include_secret': True}
assert data['data']['bot'] == {
'uuid': 'test-bot-uuid',
'name': 'Test Bot',
'platform': 'telegram',
'pipeline_uuid': 'test-pipeline-uuid',
'webhook_url': 'https://example.com/webhook/test-bot-uuid',
}
@pytest.mark.asyncio
async def test_update_bot_success(self, quart_test_client):
async def test_update_bot_success(self, quart_test_client, fake_bot_app):
"""PUT /api/v1/platform/bots/{uuid} updates bot."""
response = await quart_test_client.put(
'/api/v1/platform/bots/test-bot-uuid',
@@ -243,8 +279,14 @@ class TestBotEndpoints:
data = await response.get_json()
assert data['code'] == 0
called = fake_bot_app.bot_service.update_bot
assert called.await_count == 1
assert called.await_args.args[0].workspace_uuid == 'workspace-test'
assert called.await_args.args[1:] == ('test-bot-uuid', {'name': 'Updated Bot'})
assert called.await_args.kwargs == {}
@pytest.mark.asyncio
async def test_delete_bot_success(self, quart_test_client):
async def test_delete_bot_success(self, quart_test_client, fake_bot_app):
"""DELETE /api/v1/platform/bots/{uuid} deletes bot."""
response = await quart_test_client.delete(
'/api/v1/platform/bots/test-bot-uuid', headers={'Authorization': 'Bearer test_token'}
@@ -252,13 +294,20 @@ class TestBotEndpoints:
assert response.status_code == 200
called = fake_bot_app.bot_service.delete_bot
assert called.await_count == 1
assert called.await_args.args[0].workspace_uuid == 'workspace-test'
assert called.await_args.args[1:] == ('test-bot-uuid',)
assert called.await_args.kwargs == {}
assert (await response.get_json())['code'] == 0
@pytest.mark.usefixtures('mock_circular_import_chain')
class TestBotLogsEndpoint:
"""Tests for bot logs endpoint."""
@pytest.mark.asyncio
async def test_get_bot_logs_success(self, quart_test_client):
async def test_get_bot_logs_success(self, quart_test_client, fake_bot_app):
"""POST /api/v1/platform/bots/{uuid}/logs returns logs."""
response = await quart_test_client.post(
'/api/v1/platform/bots/test-bot-uuid/logs',
@@ -272,6 +321,13 @@ class TestBotLogsEndpoint:
assert 'logs' in data['data']
assert 'total_count' in data['data']
called = fake_bot_app.bot_service.list_event_logs
assert called.await_count == 1
assert called.await_args.args[0].workspace_uuid == 'workspace-test'
assert called.await_args.args[1:] == ('test-bot-uuid', -1, 10)
assert called.await_args.kwargs == {}
assert data['data'] == {'logs': [{'uuid': 'log-1', 'message': 'test log'}], 'total_count': 1}
@pytest.mark.asyncio
async def test_viewer_can_read_ordinary_bot_logs(self, quart_test_client, fake_bot_app):
access = fake_bot_app.workspace_collaboration_service.resolve_account_workspace.return_value