Add tests

Author: febus982

tests/common/test_asyncapi.py

@@ -0,0 +1,133 @@
+import pytest
+from pydantic import BaseModel
+
+from common.asyncapi import (
+    get_schema,
+    init_asyncapi_info,
+    register_channel,
+    register_channel_operation,
+    register_server,
+)
+
+
+# Test fixtures
+@pytest.fixture
+def reset_asyncapi_state():
+    """Reset all global state between tests"""
+    from common.asyncapi import _channels, _components_schemas, _operations, _servers
+
+    _servers.clear()
+    _channels.clear()
+    _operations.clear()
+    _components_schemas.clear()
+    yield
+    _servers.clear()
+    _channels.clear()
+    _operations.clear()
+    _components_schemas.clear()
+
+
+# Test message models
+class TestMessage(BaseModel):
+    content: str
+    timestamp: int
+
+
+class AnotherTestMessage(BaseModel):
+    status: bool
+    code: int
+
+
+# Test cases
+def test_init_asyncapi_info():
+    """Test initialization of AsyncAPI info"""
+    title = "Test API"
+    version = "2.0.0"
+
+    init_asyncapi_info(title=title, version=version)
+    schema = get_schema()
+
+    assert schema.info.title == title
+    assert schema.info.version == version
+
+
+def test_register_server(reset_asyncapi_state):
+    """Test server registration"""
+    server_id = "test-server"
+    host = "localhost"
+    protocol = "ws"
+    pathname = "/ws"
+
+    register_server(id=server_id, host=host, protocol=protocol, pathname=pathname)
+
+    schema = get_schema()
+    assert server_id in schema.servers
+    assert schema.servers[server_id].host == host
+    assert schema.servers[server_id].protocol == protocol
+    assert schema.servers[server_id].pathname == pathname
+
+
+def test_register_channel(reset_asyncapi_state):
+    """Test channel registration"""
+    channel_id = "test-channel"
+    address = "test/topic"
+    description = "Test channel"
+    title = "Test Channel"
+
+    register_channel(address=address, id=channel_id, description=description, title=title)
+
+    schema = get_schema()
+    assert channel_id in schema.channels
+    assert schema.channels[channel_id].address == address
+    assert schema.channels[channel_id].description == description
+    assert schema.channels[channel_id].title == title
+
+
+def test_register_channel_with_server(reset_asyncapi_state):
+    """Test channel registration with server reference"""
+    server_id = "test-server"
+    channel_id = "test-channel"
+
+    register_server(id=server_id, host="localhost", protocol="ws")
+    register_channel(address="test/topic", id=channel_id, server_id=server_id)
+
+    schema = get_schema()
+    assert len(schema.channels[channel_id].servers) == 1
+    assert schema.channels[channel_id].servers[0].ref == f"#/servers/{server_id}"
+
+
+def test_register_channel_operation(reset_asyncapi_state):
+    """Test channel operation registration"""
+    channel_id = "test-channel"
+    operation_type = "receive"
+
+    register_channel(address="test/topic", id=channel_id)
+    register_channel_operation(
+        channel_id=channel_id, operation_type=operation_type, messages=[TestMessage], operation_name="test-operation"
+    )
+
+    schema = get_schema()
+    assert "test-operation" in schema.operations
+    assert schema.operations["test-operation"].action == operation_type
+    assert schema.operations["test-operation"].channel.ref == f"#/channels/{channel_id}"
+    assert TestMessage.__name__ in schema.components.schemas
+
+
+def test_register_channel_operation_invalid_channel(reset_asyncapi_state):
+    """Test channel operation registration with invalid channel"""
+    with pytest.raises(ValueError, match="Channel non-existent does not exist"):
+        register_channel_operation(channel_id="non-existent", operation_type="receive", messages=[TestMessage])
+
+
+def test_multiple_messages_registration(reset_asyncapi_state):
+    """Test registration of multiple messages for an operation"""
+    channel_id = "test-channel"
+
+    register_channel(address="test/topic", id=channel_id)
+    register_channel_operation(channel_id=channel_id, operation_type="send", messages=[TestMessage, AnotherTestMessage])
+
+    schema = get_schema()
+    assert TestMessage.__name__ in schema.components.schemas
+    assert AnotherTestMessage.__name__ in schema.components.schemas
+    assert TestMessage.__name__ in schema.channels[channel_id].messages
+    assert AnotherTestMessage.__name__ in schema.channels[channel_id].messages

tests/conftest.py

@@ -12,7 +12,7 @@ def anyio_backend():
     return "asyncio"
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="session")
 def test_config() -> AppConfig:
     return AppConfig(
         SQLALCHEMY_CONFIG={},

tests/http_app/conftest.py

@@ -7,7 +7,7 @@
 from http_app import create_app
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="session")
 def testapp(test_config) -> Iterator[FastAPI]:
     # We don't need the storage to test the HTTP app
     with patch("common.bootstrap.init_storage", return_value=None):

tests/http_app/routes/test_docs_ws.py

@@ -0,0 +1,92 @@
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import FastAPI, status
+from fastapi.testclient import TestClient
+from pydantic_asyncapi.v3 import AsyncAPI, Info
+
+fake_schema = AsyncAPI(
+    asyncapi="3.0.0",
+    info=Info(
+        title="Some fake schema",
+        version="1.2.3",
+    ),
+)
+
+
+@patch("http_app.routes.docs_ws.get_schema", return_value=fake_schema)
+async def test_asyncapi_json_is_whatever_returned_by_schema(
+    mock_schema: MagicMock,
+    testapp: FastAPI,
+):
+    ac = TestClient(app=testapp, base_url="http://test")
+    response = ac.get(
+        "/docs/ws/asyncapi.json",
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.text == fake_schema.model_dump_json(exclude_unset=True)
+
+
+@pytest.mark.parametrize("sidebar", (True, False))
+@pytest.mark.parametrize("info", (True, False))
+@pytest.mark.parametrize("servers", (True, False))
+@pytest.mark.parametrize("operations", (True, False))
+@pytest.mark.parametrize("messages", (True, False))
+@pytest.mark.parametrize("schema", (True, False))
+@pytest.mark.parametrize("errors", (True, False))
+@pytest.mark.parametrize("expand_message_examples", (True, False))
+async def test_ws_docs_renders_config_based_on_params(
+    sidebar: bool,
+    info: bool,
+    servers: bool,
+    operations: bool,
+    messages: bool,
+    schema: bool,
+    errors: bool,
+    expand_message_examples: bool,
+    testapp: FastAPI,
+):
+    config = json.dumps(
+        {
+            "schema": {
+                "url": "/docs/ws/asyncapi.json",
+            },
+            "config": {
+                "show": {
+                    "sidebar": sidebar,
+                    "info": info,
+                    "servers": servers,
+                    "operations": operations,
+                    "messages": messages,
+                    "schemas": schema,
+                    "errors": errors,
+                },
+                "expand": {
+                    "messageExamples": expand_message_examples,
+                },
+                "sidebar": {
+                    "showServers": "byDefault",
+                    "showOperations": "byDefault",
+                },
+            },
+        }
+    )
+
+    ac = TestClient(app=testapp, base_url="http://test")
+    response = ac.get(
+        "/docs/ws",
+        params={
+            "sidebar": sidebar,
+            "info": info,
+            "servers": servers,
+            "operations": operations,
+            "messages": messages,
+            "schemas": schema,
+            "errors": errors,
+            "expand_message_examples": expand_message_examples,
+        },
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert config in response.text

tests/http_app/routes/ws/__init__.py

@@ -0,0 +1,85 @@
+import pytest
+from fastapi.testclient import TestClient
+
+from http_app.routes.ws.chat import ConnectionManager
+
+
+@pytest.fixture
+def test_client(testapp):
+    return TestClient(testapp)
+
+
+@pytest.fixture
+def connection_manager():
+    return ConnectionManager()
+
+
+def test_websocket_connection(test_client):
+    with test_client.websocket_connect("/ws/chat/1") as websocket:
+        websocket.send_text("Hello!")
+        data = websocket.receive_text()
+
+        assert data == "You wrote: Hello!"
+        broadcast = websocket.receive_text()
+        assert broadcast == "Client #1 says: Hello!"
+
+
+def test_multiple_clients(test_client):
+    with test_client.websocket_connect("/ws/chat/1") as websocket1:
+        with test_client.websocket_connect("/ws/chat/2") as websocket2:
+            # Client 1 sends message
+            websocket1.send_text("Hello from client 1")
+
+            # Client 1 receives personal message
+            data1 = websocket1.receive_text()
+            assert data1 == "You wrote: Hello from client 1"
+
+            # Both clients receive broadcast
+            broadcast1 = websocket1.receive_text()
+            broadcast2 = websocket2.receive_text()
+            assert broadcast1 == "Client #1 says: Hello from client 1"
+            assert broadcast2 == "Client #1 says: Hello from client 1"
+
+
+def test_client_disconnect(test_client):
+    with test_client.websocket_connect("/ws/chat/1") as websocket1:
+        with test_client.websocket_connect("/ws/chat/2") as websocket2:
+            # Close first client
+            websocket1.close()
+
+            # Second client should receive disconnect message
+            disconnect_message = websocket2.receive_text()
+            assert disconnect_message == "Client #1 left the chat"
+
+
+@pytest.mark.asyncio
+async def test_connection_manager():
+    manager = ConnectionManager()
+
+    # Mock WebSocket for testing
+    class MockWebSocket:
+        def __init__(self):
+            self.received_messages = []
+
+        async def accept(self):
+            pass
+
+        async def send_text(self, message: str):
+            self.received_messages.append(message)
+
+    # Test connect
+    websocket = MockWebSocket()
+    await manager.connect(websocket)
+    assert len(manager.active_connections) == 1
+
+    # Test personal message
+    await manager.send_personal_message("test message", websocket)
+    assert websocket.received_messages[-1] == "test message"
+
+    # Test broadcast
+    await manager.broadcast("broadcast message")
+    assert websocket.received_messages[-1] == "broadcast message"
+
+    # Test disconnect
+    manager.disconnect(websocket)
+    assert len(manager.active_connections) == 0

tests/http_app/routes/ws/test_chat.py

@@ -1,9 +1,9 @@
 from common import AppConfig
 from http_app import context
-from http_app.dependencies import app_config
+from http_app.dependencies import get_app_config
 
 
 def test_app_config_return_context_variable():
     config = AppConfig(APP_NAME="SomeOtherAppName")
     context.app_config.set(config)
-    assert app_config() is config
+    assert get_app_config() is config