mirror of
https://github.com/Significant-Gravitas/AutoGPT.git
synced 2026-02-09 22:35:54 -05:00
a3d082a5fa
This pull request introduces a comprehensive backend testing guide and adds new tests for analytics logging and various API endpoints, focusing on snapshot testing. It also includes corresponding snapshot files for these tests. Below are the most significant changes: ### Documentation Updates: * Added a detailed `TESTING.md` file to the backend, providing a guide for running tests, snapshot testing, writing API route tests, and best practices. It includes examples for mocking, fixtures, and CI/CD integration. ### Analytics Logging Tests: * Implemented tests for logging raw metrics and analytics in `analytics_test.py`, covering success scenarios, various input values, invalid requests, and complex nested data. These tests utilize snapshot testing for response validation. * Added snapshot files for analytics logging tests, including responses for success cases, various metric values, and complex analytics data. [[1]](diffhunk://#diff-654bc5aa1951008ec5c110a702279ef58709ee455ba049b9fa825fa60f7e3869R1-R3) [[2]](diffhunk://#diff-e0a434b107abc71aeffb7d7989dbfd8f466b5e53f8dea25a87937ec1b885b122R1-R3) [[3]](diffhunk://#diff-dd0bc0b72264de1a0c0d3bd0c54ad656061317f425e4de461018ca51a19171a0R1-R3) [[4]](diffhunk://#diff-63af007073db553d04988544af46930458a768544cabd08412265e0818320d11R1-R30) ### Snapshot Files for API Endpoints: * Added snapshot files for various API endpoint tests, such as: - Graph-related operations (`graphs_get_single_response`, `graphs_get_all_response`, `blocks_get_all_response`). [[1]](diffhunk://#diff-b25dba271606530cfa428c00073d7e016184a7bb22166148ab1726b3e113dda8R1-R29) [[2]](diffhunk://#diff-1054e58ec3094715660f55bfba1676d65b6833a81a91a08e90ad57922444d056R1-R31) [[3]](diffhunk://#diff-cfd403ab6f3efc89188acaf993d85e6f792108d1740c7e7149eb05efb73d918dR1-R14) - User-related operations (`auth_get_or_create_user_response`, `auth_update_email_response`). [[1]](diffhunk://#diff-49e65ab1eb6af4d0163a6c54ed10be621ce7336b2ab5d47d47679bfaefdb7059R1-R5) [[2]](diffhunk://#diff-ac1216f96878bd4356454c317473654d5d5c7c180125663b80b0b45aa5ab52cbR1-R3) - Credit-related operations (`credits_get_balance_response`, `credits_get_auto_top_up_response`, `credits_top_up_request_response`). [[1]](diffhunk://#diff-189488f8da5be74d80ac3fb7f84f1039a408573184293e9ba2e321d535c57cddR1-R3) [[2]](diffhunk://#diff-ba3c4a6853793cbed24030cdccedf966d71913451ef8eb4b2c4f426ef18ed87aR1-R4) [[3]](diffhunk://#diff-43d7daa0c82070a9b6aee88a774add8e87533e630bbccbac5a838b7a7ae56a75R1-R3) - Graph execution and deletion (`blocks_execute_response`, `graphs_delete_response`). [[1]](diffhunk://#diff-a2ade7d646ad85a2801e7ff39799a925a612548a1cdd0ed99b44dd870d1465b5R1-R12) [[2]](diffhunk://#diff-c0d1cd0a8499ee175ce3007c3a87ba5f3235ce02d38ce837560b36a44fdc4a22R1-R3)## Summary - add pytest-snapshot to backend dev requirements - snapshot server route response JSONs - mention how to update stored snapshots ## Testing - `poetry run format` - `poetry run test` ### Checklist 📋 #### For code changes: - [x] I have clearly listed my changes in the PR description - [x] I have made a test plan - [x] I have tested my changes according to the test plan: <!-- Put your test plan here: --> - [x] run poetry run test --------- Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Nicholas Tindle <nicholas.tindle@agpt.co>
225 lines
7.3 KiB
Python
225 lines
7.3 KiB
Python
import json
|
|
from typing import cast
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import WebSocket, WebSocketDisconnect
|
|
from pytest_snapshot.plugin import Snapshot
|
|
|
|
from backend.data.user import DEFAULT_USER_ID
|
|
from backend.server.conn_manager import ConnectionManager
|
|
from backend.server.ws_api import (
|
|
WSMessage,
|
|
WSMethod,
|
|
handle_subscribe,
|
|
handle_unsubscribe,
|
|
websocket_router,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_websocket() -> AsyncMock:
|
|
return AsyncMock(spec=WebSocket)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_manager() -> AsyncMock:
|
|
return AsyncMock(spec=ConnectionManager)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_router_subscribe(
|
|
mock_websocket: AsyncMock, mock_manager: AsyncMock, snapshot: Snapshot
|
|
) -> None:
|
|
mock_websocket.receive_text.side_effect = [
|
|
WSMessage(
|
|
method=WSMethod.SUBSCRIBE_GRAPH_EXEC,
|
|
data={"graph_exec_id": "test-graph-exec-1"},
|
|
).model_dump_json(),
|
|
WebSocketDisconnect(),
|
|
]
|
|
mock_manager.subscribe_graph_exec.return_value = (
|
|
f"{DEFAULT_USER_ID}|graph_exec#test-graph-exec-1"
|
|
)
|
|
|
|
await websocket_router(
|
|
cast(WebSocket, mock_websocket), cast(ConnectionManager, mock_manager)
|
|
)
|
|
|
|
mock_manager.connect_socket.assert_called_once_with(mock_websocket)
|
|
mock_manager.subscribe_graph_exec.assert_called_once_with(
|
|
user_id=DEFAULT_USER_ID,
|
|
graph_exec_id="test-graph-exec-1",
|
|
websocket=mock_websocket,
|
|
)
|
|
mock_websocket.send_text.assert_called_once()
|
|
assert (
|
|
'"method":"subscribe_graph_execution"'
|
|
in mock_websocket.send_text.call_args[0][0]
|
|
)
|
|
assert '"success":true' in mock_websocket.send_text.call_args[0][0]
|
|
|
|
# Capture and snapshot the WebSocket response message
|
|
sent_message = mock_websocket.send_text.call_args[0][0]
|
|
parsed_message = json.loads(sent_message)
|
|
snapshot.snapshot_dir = "snapshots"
|
|
snapshot.assert_match(json.dumps(parsed_message, indent=2, sort_keys=True), "sub")
|
|
|
|
mock_manager.disconnect_socket.assert_called_once_with(mock_websocket)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_router_unsubscribe(
|
|
mock_websocket: AsyncMock, mock_manager: AsyncMock, snapshot: Snapshot
|
|
) -> None:
|
|
mock_websocket.receive_text.side_effect = [
|
|
WSMessage(
|
|
method=WSMethod.UNSUBSCRIBE,
|
|
data={"graph_exec_id": "test-graph-exec-1"},
|
|
).model_dump_json(),
|
|
WebSocketDisconnect(),
|
|
]
|
|
mock_manager.unsubscribe_graph_exec.return_value = (
|
|
f"{DEFAULT_USER_ID}|graph_exec#test-graph-exec-1"
|
|
)
|
|
|
|
await websocket_router(
|
|
cast(WebSocket, mock_websocket), cast(ConnectionManager, mock_manager)
|
|
)
|
|
|
|
mock_manager.connect_socket.assert_called_once_with(mock_websocket)
|
|
mock_manager.unsubscribe_graph_exec.assert_called_once_with(
|
|
user_id=DEFAULT_USER_ID,
|
|
graph_exec_id="test-graph-exec-1",
|
|
websocket=mock_websocket,
|
|
)
|
|
mock_websocket.send_text.assert_called_once()
|
|
assert '"method":"unsubscribe"' in mock_websocket.send_text.call_args[0][0]
|
|
assert '"success":true' in mock_websocket.send_text.call_args[0][0]
|
|
|
|
# Capture and snapshot the WebSocket response message
|
|
sent_message = mock_websocket.send_text.call_args[0][0]
|
|
parsed_message = json.loads(sent_message)
|
|
snapshot.snapshot_dir = "snapshots"
|
|
snapshot.assert_match(json.dumps(parsed_message, indent=2, sort_keys=True), "unsub")
|
|
|
|
mock_manager.disconnect_socket.assert_called_once_with(mock_websocket)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_router_invalid_method(
|
|
mock_websocket: AsyncMock, mock_manager: AsyncMock
|
|
) -> None:
|
|
mock_websocket.receive_text.side_effect = [
|
|
WSMessage(method=WSMethod.GRAPH_EXECUTION_EVENT).model_dump_json(),
|
|
WebSocketDisconnect(),
|
|
]
|
|
|
|
await websocket_router(
|
|
cast(WebSocket, mock_websocket), cast(ConnectionManager, mock_manager)
|
|
)
|
|
|
|
mock_manager.connect_socket.assert_called_once_with(mock_websocket)
|
|
mock_websocket.send_text.assert_called_once()
|
|
assert '"method":"error"' in mock_websocket.send_text.call_args[0][0]
|
|
assert '"success":false' in mock_websocket.send_text.call_args[0][0]
|
|
mock_manager.disconnect_socket.assert_called_once_with(mock_websocket)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_subscribe_success(
|
|
mock_websocket: AsyncMock, mock_manager: AsyncMock
|
|
) -> None:
|
|
message = WSMessage(
|
|
method=WSMethod.SUBSCRIBE_GRAPH_EXEC,
|
|
data={"graph_exec_id": "test-graph-exec-id"},
|
|
)
|
|
mock_manager.subscribe_graph_exec.return_value = (
|
|
"user-1|graph_exec#test-graph-exec-id"
|
|
)
|
|
|
|
await handle_subscribe(
|
|
connection_manager=cast(ConnectionManager, mock_manager),
|
|
websocket=cast(WebSocket, mock_websocket),
|
|
user_id="user-1",
|
|
message=message,
|
|
)
|
|
|
|
mock_manager.subscribe_graph_exec.assert_called_once_with(
|
|
user_id="user-1",
|
|
graph_exec_id="test-graph-exec-id",
|
|
websocket=mock_websocket,
|
|
)
|
|
mock_websocket.send_text.assert_called_once()
|
|
assert (
|
|
'"method":"subscribe_graph_execution"'
|
|
in mock_websocket.send_text.call_args[0][0]
|
|
)
|
|
assert '"success":true' in mock_websocket.send_text.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_subscribe_missing_data(
|
|
mock_websocket: AsyncMock, mock_manager: AsyncMock
|
|
) -> None:
|
|
message = WSMessage(method=WSMethod.SUBSCRIBE_GRAPH_EXEC)
|
|
|
|
await handle_subscribe(
|
|
connection_manager=cast(ConnectionManager, mock_manager),
|
|
websocket=cast(WebSocket, mock_websocket),
|
|
user_id="user-1",
|
|
message=message,
|
|
)
|
|
|
|
mock_manager.subscribe_graph_exec.assert_not_called()
|
|
mock_websocket.send_text.assert_called_once()
|
|
assert '"method":"error"' in mock_websocket.send_text.call_args[0][0]
|
|
assert '"success":false' in mock_websocket.send_text.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_unsubscribe_success(
|
|
mock_websocket: AsyncMock, mock_manager: AsyncMock
|
|
) -> None:
|
|
message = WSMessage(
|
|
method=WSMethod.UNSUBSCRIBE, data={"graph_exec_id": "test-graph-exec-id"}
|
|
)
|
|
mock_manager.unsubscribe_graph_exec.return_value = (
|
|
"user-1|graph_exec#test-graph-exec-id"
|
|
)
|
|
|
|
await handle_unsubscribe(
|
|
connection_manager=cast(ConnectionManager, mock_manager),
|
|
websocket=cast(WebSocket, mock_websocket),
|
|
user_id="user-1",
|
|
message=message,
|
|
)
|
|
|
|
mock_manager.unsubscribe_graph_exec.assert_called_once_with(
|
|
user_id="user-1",
|
|
graph_exec_id="test-graph-exec-id",
|
|
websocket=mock_websocket,
|
|
)
|
|
mock_websocket.send_text.assert_called_once()
|
|
assert '"method":"unsubscribe"' in mock_websocket.send_text.call_args[0][0]
|
|
assert '"success":true' in mock_websocket.send_text.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_unsubscribe_missing_data(
|
|
mock_websocket: AsyncMock, mock_manager: AsyncMock
|
|
) -> None:
|
|
message = WSMessage(method=WSMethod.UNSUBSCRIBE)
|
|
|
|
await handle_unsubscribe(
|
|
connection_manager=cast(ConnectionManager, mock_manager),
|
|
websocket=cast(WebSocket, mock_websocket),
|
|
user_id="user-1",
|
|
message=message,
|
|
)
|
|
|
|
mock_manager._unsubscribe.assert_not_called()
|
|
mock_websocket.send_text.assert_called_once()
|
|
assert '"method":"error"' in mock_websocket.send_text.call_args[0][0]
|
|
assert '"success":false' in mock_websocket.send_text.call_args[0][0]
|