StreamFlow uses pytest for backend unit and integration tests. The frontend currently has no automated tests (Jest/React Testing Library not configured).
api_service/tests/
├── conftest.py # Fixtures and test configuration
├── test_api_integration.py # API endpoint tests
├── test_validators.py # Input validation tests
└── test_decorator.py # Retry decorator tests
cd api_service
uv pip install -e ".[dev]" # Install dev dependencies including pytest
pytest # Run all tests
pytest -v # Verbose output
pytest -k "test_validators" # Run specific test file
pytest --cov=src # With coverage reportIntegration tests use SQLite in-memory database and mock Kafka:
pytest tests/test_api_integration.py -vTest individual functions/classes in isolation:
def test_validate_symbol():
from schemas.validators import validate_symbol
assert validate_symbol("VNM") == "VNM"
with pytest.raises(ValueError):
validate_symbol("invalid")Common fixtures in conftest.py:
test_engine: SQLite in-memory engine with all tables createdtest_db_session: Fresh database session per test (rolls back after)client: TestClient with database dependency overrides
Use pytest-mock for mocking:
def test_something(mocker):
mocker.patch('some.module.function', return_value='mocked')Current coverage areas:
- Input validation (validators.py)
- Retry decorator
- Health endpoint
- Stocks endpoint (structure)
Areas needing coverage:
- StockService methods (with mocked DB)
- Auth endpoints
- WebSocket manager
- Kafka bridge
# Health check
curl http://localhost:8000/health
# Get stocks (requires auth)
curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/v1/stocks
# Get specific quote
curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/v1/stocks/VNMVisit http://localhost:8000/docs for interactive API testing.
Recommended tool: locust
pip install locust
locust -f tests/load_test.pyAdd to your CI pipeline:
test:
script:
- cd api_service
- uv pip install -e ".[dev]"
- pytest --cov=src --cov-report=xml