Testing Guide¶
Comprehensive testing practices for DSPU.
Overview¶
DSPU uses pytest for testing with emphasis on: - High coverage (> 90%) - Fast execution - Clear test organization - Comprehensive edge cases
Test Organization¶
tests/
├── unit/ # Unit tests (fast, isolated)
│ ├── config/
│ ├── io/
│ ├── aio/
│ ├── validation/
│ ├── security/
│ ├── observability/
│ └── ml/
└── integration/ # Integration tests (slower)
├── config/
└── io/
Running Tests¶
All Tests¶
Specific Module¶
With Coverage¶
Fast Tests Only¶
Verbose Output¶
Writing Tests¶
Basic Test¶
def test_scaler_standard_scaling():
"""Test standard scaling transforms data correctly."""
scaler = Scaler(method="standard")
data = [[1.0, 2.0], [3.0, 4.0]]
result = scaler.fit_transform(data)
assert len(result) == 2
assert result[0][0] < result[1][0] # Relative ordering preserved
Async Tests¶
import pytest
@pytest.mark.asyncio
async def test_async_operation():
"""Test async operation."""
result = await async_function()
assert result == expected
Parametrized Tests¶
import pytest
@pytest.mark.parametrize("method,expected", [
("standard", [[0.0, 0.0], [1.0, 1.0]]),
("minmax", [[0.0, 0.0], [1.0, 1.0]]),
])
def test_scaler_methods(method, expected):
"""Test different scaling methods."""
scaler = Scaler(method=method)
data = [[1.0, 2.0], [3.0, 4.0]]
result = scaler.fit_transform(data)
assert result == pytest.approx(expected, abs=0.1)
Test Fixtures¶
import pytest
@pytest.fixture
def sample_data():
"""Provide sample data for tests."""
return [[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]]
@pytest.fixture
def scaler():
"""Provide configured scaler."""
return Scaler(method="standard")
def test_with_fixtures(scaler, sample_data):
"""Test using fixtures."""
result = scaler.fit_transform(sample_data)
assert len(result) == len(sample_data)
Test Patterns¶
Arrange-Act-Assert¶
def test_processor_transforms_data():
# Arrange
processor = DataProcessor()
data = [[1, 2], [3, 4]]
# Act
result = processor.transform(data)
# Assert
assert len(result) == 2
assert result[0] == expected_value
Testing Exceptions¶
import pytest
from dspu.core import ValidationError
def test_validator_raises_on_invalid_input():
"""Test validator raises appropriate exception."""
validator = MyValidator()
with pytest.raises(ValidationError) as exc_info:
validator.validate(invalid_data)
assert "field" in str(exc_info.value)
assert exc_info.value.field == "expected_field"
Testing Edge Cases¶
def test_scaler_handles_empty_data():
"""Test scaler handles empty input gracefully."""
scaler = Scaler(method="standard")
with pytest.raises(ValueError, match="empty"):
scaler.fit_transform([])
def test_scaler_handles_single_sample():
"""Test scaler with single sample."""
scaler = Scaler(method="standard")
data = [[1.0, 2.0]]
result = scaler.fit_transform(data)
assert len(result) == 1
def test_scaler_handles_nan_values():
"""Test scaler handles NaN values."""
scaler = Scaler(method="standard")
data = [[1.0, float('nan')], [3.0, 4.0]]
# Should either handle or raise clear error
...
Mocking¶
Mock External Dependencies¶
from unittest.mock import Mock, patch
def test_with_mock():
"""Test with mocked dependency."""
mock_client = Mock()
mock_client.get.return_value = {"status": "success"}
result = process_with_client(mock_client)
assert result["status"] == "success"
mock_client.get.assert_called_once()
@patch('dspu.io.httpx.AsyncClient')
async def test_with_patch(mock_client):
"""Test with patched HTTP client."""
mock_client.return_value.get.return_value = {"data": "test"}
result = await fetch_data()
assert result["data"] == "test"
Mock Files¶
from unittest.mock import mock_open, patch
def test_read_config():
"""Test reading config file."""
mock_data = "key: value\n"
with patch("builtins.open", mock_open(read_data=mock_data)):
config = load_config("config.yaml")
assert config["key"] == "value"
Test Markers¶
Mark Slow Tests¶
import pytest
@pytest.mark.slow
def test_large_dataset_processing():
"""Test with large dataset (slow)."""
...
# Skip slow tests
# pytest -m "not slow"
Mark Integration Tests¶
@pytest.mark.integration
async def test_full_pipeline():
"""Test complete pipeline integration."""
...
# Run only integration tests
# pytest -m integration
Skip Tests¶
import sys
@pytest.mark.skipif(sys.platform == "win32", reason="Unix only")
def test_unix_specific():
...
@pytest.mark.skip(reason="Not implemented yet")
def test_future_feature():
...
Coverage¶
Minimum Coverage¶
Maintain > 90% coverage:
View Coverage Report¶
Coverage Configuration¶
# pyproject.toml
[tool.coverage.run]
source = ["src"]
omit = ["*/tests/*", "*/test_*.py"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise AssertionError",
"raise NotImplementedError",
"if __name__ == .__main__.:",
]
Testing Best Practices¶
Do's¶
✅ DO: - Write tests before fixing bugs - Test edge cases (empty, None, invalid) - Use descriptive test names - Keep tests fast and isolated - Mock external dependencies - Test error conditions - Aim for > 90% coverage
Don'ts¶
❌ DON'T: - Don't test implementation details - Don't write flaky tests - Don't skip edge cases - Don't test third-party libraries - Don't use sleep() (use timeouts) - Don't share state between tests
Example: Complete Test Suite¶
import pytest
from dspu.ml import Scaler
class TestScaler:
"""Test suite for Scaler class."""
@pytest.fixture
def scaler(self):
"""Provide scaler instance."""
return Scaler(method="standard")
@pytest.fixture
def sample_data(self):
"""Provide sample data."""
return [[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]]
def test_fit_transform_standard(self, scaler, sample_data):
"""Test fit_transform with standard scaling."""
result = scaler.fit_transform(sample_data)
assert len(result) == len(sample_data)
# Mean should be ~0
assert abs(sum(r[0] for r in result) / len(result)) < 0.1
def test_transform_without_fit_raises(self, scaler):
"""Test transform without fit raises error."""
with pytest.raises(ValueError, match="not fitted"):
scaler.transform([[1.0, 2.0]])
def test_empty_data_raises(self, scaler):
"""Test empty data raises appropriate error."""
with pytest.raises(ValueError, match="empty"):
scaler.fit_transform([])
def test_persistence(self, scaler, sample_data, tmp_path):
"""Test saving and loading scaler."""
scaler.fit_transform(sample_data)
# Save
path = tmp_path / "scaler.json"
scaler.save_to_file(str(path))
# Load
loaded = Scaler.load_from_file(str(path))
# Should produce same results
result1 = scaler.transform(sample_data)
result2 = loaded.transform(sample_data)
assert result1 == pytest.approx(result2)
@pytest.mark.parametrize("method", ["standard", "minmax", "robust"])
def test_all_methods(self, method, sample_data):
"""Test all scaling methods work."""
scaler = Scaler(method=method)
result = scaler.fit_transform(sample_data)
assert len(result) == len(sample_data)
assert len(result[0]) == len(sample_data[0])
Continuous Integration¶
GitHub Actions¶
# .github/workflows/test.yml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install uv
uv sync --all-extras
- name: Run tests
run: uv run pytest --cov=src/dspu
- name: Upload coverage
uses: codecov/codecov-action@v3
Debugging Tests¶
Run Single Test¶
Print Debug Output¶
Drop into Debugger¶
def test_something():
result = complex_operation()
import pdb; pdb.set_trace() # Debug here
assert result == expected