Running Tests¶
Guide to running and writing tests for Adaptive Sentience.
Test Suite Overview¶
The test suite covers:
- Unit tests: Individual components
- Integration tests: Components working together
- End-to-end tests: Full workflow execution
- Performance tests: Benchmarking critical paths
Running Tests¶
Quick Start¶
# Run all tests
pytest
# Run with coverage
pytest --cov=. --cov-report=html
# View coverage report
open htmlcov/index.html
Run Specific Tests¶
# Run tests in a specific file
pytest tests/test_gateway.py
# Run tests matching a pattern
pytest -k "test_router"
# Run a specific test
pytest tests/test_gateway.py::test_router_selection
# Run tests in a directory
pytest tests/integration/
Verbose Output¶
# Show test names and results
pytest -v
# Show print statements
pytest -s
# Show local variables on failure
pytest -l
# Stop on first failure
pytest -x
Test Organization¶
Directory Structure¶
tests/
├── unit/ # Unit tests
│ ├── test_gateway.py
│ ├── test_router.py
│ └── test_trust.py
├── integration/ # Integration tests
│ ├── test_workflow_execution.py
│ └── test_mesh_discovery.py
├── e2e/ # End-to-end tests
│ └── test_full_workflow.py
├── performance/ # Performance tests
│ └── test_benchmarks.py
└── conftest.py # Shared fixtures
Test Naming¶
- Test files:
test_*.py - Test functions:
test_* - Test classes:
Test*
Writing Tests¶
Unit Test Example¶
# tests/unit/test_router.py
import pytest
from gateway.router import Router
from gateway.models import Node
def test_router_selects_trusted_node():
"""Router should prefer trusted nodes."""
router = Router()
nodes = [
Node(node_id="untrusted", tools=["pii_redact"], trusted=False),
Node(node_id="trusted", tools=["pii_redact"], trusted=True)
]
selected = router.select_node(nodes, "pii_redact")
assert selected.node_id == "trusted"
def test_router_handles_no_nodes():
"""Router should return None when no nodes available."""
router = Router()
selected = router.select_node([], "pii_redact")
assert selected is None
Async Test Example¶
# tests/unit/test_gateway.py
import pytest
from httpx import AsyncClient
from gateway.http_gateway import app
@pytest.mark.asyncio
async def test_health_endpoint():
"""Health endpoint should return 200."""
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
@pytest.mark.asyncio
async def test_tool_call_endpoint():
"""Tool call endpoint should execute tool."""
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post("/v1/tool/call", json={
"target": {"kind": "local"},
"tool_name": "pii_redact",
"tool_args": {"text": "test@example.com"}
})
assert response.status_code == 200
data = response.json()
assert data["ok"]
assert "[REDACTED]" in data["result"]["redacted_text"]
Integration Test Example¶
# tests/integration/test_workflow_execution.py
import pytest
from gateway.http_gateway import app
from edge_node.node import EdgeNode
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_full_workflow_execution(gateway, edge_node):
"""Test complete workflow from gateway to edge node."""
# Start edge node
edge_node.start()
# Execute workflow via gateway
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post("/v1/orchestrate", json={
"text": "Redact PII: contact@example.com"
})
assert response.status_code == 200
data = response.json()
assert data["ok"]
assert data["verified"]
assert "[REDACTED]" in data["result"]["summary"]
# Stop edge node
edge_node.stop()
Test Fixtures¶
Using Fixtures¶
# tests/conftest.py
import pytest
from gateway.router import Router
from gateway.models import Node
from trust.store import TrustStore
@pytest.fixture
def router():
"""Provide a Router instance."""
return Router()
@pytest.fixture
def mock_nodes():
"""Provide mock nodes for testing."""
return [
Node(node_id="node_a", tools=["pii_redact"], trusted=True),
Node(node_id="node_b", tools=["summarize"], trusted=True),
Node(node_id="node_c", tools=["pii_redact", "summarize"], trusted=False)
]
@pytest.fixture
def trust_store(tmp_path):
"""Provide a TrustStore with temporary storage."""
store_path = tmp_path / "trust_store.json"
return TrustStore(store_path=str(store_path))
@pytest.fixture
async def edge_node():
"""Provide a running edge node."""
node = EdgeNode(port=9000)
await node.start()
yield node
await node.stop()
Using Fixtures in Tests¶
def test_with_router(router):
"""Use router fixture."""
assert router is not None
def test_with_nodes(router, mock_nodes):
"""Use multiple fixtures."""
selected = router.select_node(mock_nodes, "pii_redact")
assert selected is not None
Mocking¶
Mock External Services¶
# tests/unit/test_external_api.py
import pytest
from unittest.mock import patch, Mock
from gateway.external import ExternalAPIClient
@pytest.mark.asyncio
@patch("httpx.AsyncClient.post")
async def test_external_api_call(mock_post):
"""Mock external API calls."""
# Configure mock
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"result": "success"}
mock_post.return_value = mock_response
# Test
client = ExternalAPIClient()
result = await client.call_api("test_endpoint")
assert result == {"result": "success"}
mock_post.assert_called_once()
Mock File System¶
import pytest
from unittest.mock import mock_open, patch
def test_read_config():
"""Mock file reading."""
mock_data = '{"key": "value"}'
with patch("builtins.open", mock_open(read_data=mock_data)):
config = read_config("config.json")
assert config["key"] == "value"
Parameterized Tests¶
Test Multiple Inputs¶
import pytest
@pytest.mark.parametrize("input,expected", [
("test@example.com", "[REDACTED]"),
("John Doe", "[REDACTED]"),
("555-1234", "[REDACTED]"),
])
def test_pii_redaction(input, expected):
"""Test PII redaction with multiple inputs."""
result = redact_pii(input)
assert expected in result
@pytest.mark.parametrize("tool_name,expected_node", [
("pii_redact", "node_a"),
("summarize", "node_b"),
("validate_schema", "node_c"),
])
def test_router_selection(router, mock_nodes, tool_name, expected_node):
"""Test router selects correct node for each tool."""
selected = router.select_node(mock_nodes, tool_name)
assert selected.node_id == expected_node
Testing Best Practices¶
1. AAA Pattern¶
Structure tests with Arrange, Act, Assert:
def test_router_selection():
# Arrange
router = Router()
nodes = [Node(...), Node(...)]
# Act
selected = router.select_node(nodes, "pii_redact")
# Assert
assert selected.node_id == "expected_node"
2. Test One Thing¶
Each test should verify one behavior:
# ✅ Good - tests one thing
def test_router_prefers_trusted_nodes():
...
def test_router_handles_no_nodes():
...
# ❌ Bad - tests multiple things
def test_router_everything():
# Tests trusted nodes, no nodes, failover, etc.
...
3. Use Descriptive Names¶
4. Avoid Test Interdependencies¶
# ✅ Good - independent tests
def test_a():
data = create_test_data()
...
def test_b():
data = create_test_data()
...
# ❌ Bad - test_b depends on test_a
def test_a():
global data
data = create_test_data()
...
def test_b():
# Uses global data from test_a
...
Coverage¶
Measure Coverage¶
# Run with coverage
pytest --cov=. --cov-report=html
# View report
open htmlcov/index.html
# Show missing lines
pytest --cov=. --cov-report=term-missing
Coverage Goals¶
- Overall: >80%
- Critical paths: >95%
- New code: 100%
Exclude from Coverage¶
# .coveragerc
[run]
omit =
tests/*
*/migrations/*
*/venv/*
[report]
exclude_lines =
pragma: no cover
def __repr__
raise AssertionError
raise NotImplementedError
if __name__ == .__main__.:
Continuous Integration¶
GitHub Actions¶
# .github/workflows/test.yml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: 3.11
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pytest pytest-cov
- name: Run tests
run: pytest --cov=. --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v2
Performance Testing¶
Benchmark Example¶
# tests/performance/test_benchmarks.py
import pytest
import time
def test_router_performance(benchmark, router, mock_nodes):
"""Benchmark router selection."""
result = benchmark(router.select_node, mock_nodes, "pii_redact")
assert result is not None
def test_workflow_execution_latency(benchmark):
"""Measure workflow execution time."""
def execute_workflow():
# Execute workflow
return workflow_runner.execute(workflow)
result = benchmark(execute_workflow)
assert result["ok"]
Run Benchmarks¶
# Install pytest-benchmark
pip install pytest-benchmark
# Run benchmarks
pytest tests/performance/ --benchmark-only
# Compare with baseline
pytest tests/performance/ --benchmark-compare=0001
Debugging Tests¶
Run with Debugger¶
Print Debug Info¶
def test_with_debug():
nodes = [...]
print(f"Nodes: {nodes}") # Will show with -s flag
selected = router.select_node(nodes, "tool")
print(f"Selected: {selected}")
assert selected is not None
Next Steps¶
- Contributing - Contribution guidelines
- Local Development - Development setup
- Architecture - System architecture