testing

Testing

Learn how to write and run tests for your RouteMQ application.

Topics

• Running Tests - Execute the test suite
• Writing Unit Tests - Test controllers and middleware
• Integration Tests - Test complete workflows
• Mocking - Mock external dependencies
• Test Coverage - Measure test coverage

Quick Overview

Run the default test suite:

uv run python run_tests.py

Run Docker-backed integration tests explicitly:

RUN_INTEGRATION_TESTS=1 uv run python -m unittest tests.integration.test_queue_backends tests.integration.test_mqtt_end_to_end

Writing Tests

Create your tests in the tests/ directory:

import unittest
from unittest.mock import Mock, AsyncMock, patch
from app.controllers.sensor_controller import SensorController
from app.middleware.rate_limit import RateLimitMiddleware

class TestSensorController(unittest.IsolatedAsyncioTestCase):
    async def test_handle_temperature(self):
        # Mock client and Redis
        client = Mock()
        client.publish = Mock()
        
        # Test data
        sensor_id = "temp_001"
        payload = {"value": 25.5, "unit": "celsius"}
        
        # Call controller
        result = await SensorController.handle_temperature(sensor_id, payload, client)
        
        # Assertions
        self.assertEqual(result["status"], "processed")
        self.assertEqual(result["sensor_id"], sensor_id)

class TestRateLimitMiddleware(unittest.IsolatedAsyncioTestCase):
    async def test_rate_limiting(self):
        # Create rate limiter
        rate_limiter = RateLimitMiddleware(max_requests=2, window_seconds=60)
        
        # Mock context and handler
        context = {"topic": "test/topic", "payload": {}}
        next_handler = AsyncMock(return_value={"success": True})
        
        # First two requests should pass
        result1 = await rate_limiter.handle(context, next_handler)
        self.assertEqual(result1["success"], True)
        
        result2 = await rate_limiter.handle(context, next_handler)
        self.assertEqual(result2["success"], True)
        
        # Third request should be rate limited
        result3 = await rate_limiter.handle(context, next_handler)
        self.assertIn("rate_limit_exceeded", result3.get("error", ""))

Test Organization

tests/
├── __init__.py
├── unit/
│   └── core/
│       ├── test_controller.py
│       ├── test_middleware.py
│       └── test_router.py
├── integration/
│   ├── helpers.py
│   ├── test_mqtt_end_to_end.py
│   └── test_queue_backends.py

Best Practices

• Test Each Component: Controllers, middleware, models separately
• Mock External Dependencies: Redis, database, MQTT client
• Gate Docker Tests: Use RUN_INTEGRATION_TESTS=1 for real service tests
• Use Descriptive Names: Clear test method names
• Test Edge Cases: Error conditions and boundary values
• Keep Tests Fast: Use mocks to avoid slow operations

Next Steps

• Running Tests - Execute tests
• Unit Tests - Test individual components
• Integration Tests - Test complete flows