Writing Tests¶
This guide covers how to write tests for Carbon Connect, including fixtures, async patterns, and mocking strategies.
Test Structure¶
Tests are organized by type and subject:
tests/
unit/
services/
test_matching_engine.py
test_application_assistant.py
test_claude_client.py
test_climatiq_client.py
scrapers/
test_cordis_scraper.py
test_cohesion_client.py
test_innovate_uk_client.py
api/
test_auth.py
test_companies.py
test_grants.py
test_matches.py
integration/
test_grant_pipeline.py
test_email_service.py
e2e/
test_full_flow.py
Async Testing¶
All database and service tests use pytest-asyncio:
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from backend.app.main import app
@pytest.mark.asyncio
async def test_create_company(db_session):
"""Test creating a company with valid data."""
company = Company(
name="Test Corp",
country="DE",
tenant_id=test_tenant.id,
)
db_session.add(company)
await db_session.flush()
assert company.id is not None
assert company.name == "Test Corp"
Async Fixtures¶
@pytest_asyncio.fixture(scope="function")
async def db_session():
"""Provide a transactional database session."""
async with async_engine.connect() as connection:
async with connection.begin() as transaction:
session = AsyncSession(bind=connection)
yield session
await transaction.rollback()
@pytest_asyncio.fixture
async def test_tenant(db_session):
"""Create a test tenant."""
tenant = Tenant(name="Test Tenant", is_active=True)
db_session.add(tenant)
await db_session.flush()
return tenant
@pytest_asyncio.fixture
async def test_user(db_session, test_tenant):
"""Create a test user with authentication."""
user = User(
email="test@example.com",
hashed_password=hash_password("TestPass123!"),
tenant_id=test_tenant.id,
)
db_session.add(user)
await db_session.flush()
return user
API Testing¶
Use httpx.AsyncClient with ASGITransport for testing API endpoints:
@pytest_asyncio.fixture
async def client(db_session, test_user):
"""Create an authenticated API client."""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Login to get token
response = await client.post("/api/v1/auth/login", data={
"username": test_user.email,
"password": "TestPass123!",
})
token = response.json()["access_token"]
client.headers["Authorization"] = f"Bearer {token}"
yield client
@pytest.mark.asyncio
async def test_list_grants(client):
"""Test listing grants with pagination."""
response = await client.get("/api/v1/grants", params={"page": 1, "per_page": 10})
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
Mocking External APIs¶
Claude Client¶
from unittest.mock import AsyncMock, MagicMock, patch
@pytest.mark.asyncio
async def test_generate_content():
"""Test content generation with Claude."""
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.text = "Generated executive summary..."
mock_response.prompt_feedback = None # Explicitly None
mock_response.total_tokens = 500
mock_response.model = "claude-sonnet-4-20250514"
mock_client.generate = AsyncMock(return_value=mock_response)
with patch(
"backend.app.services.claude_client.get_claude_client",
return_value=mock_client,
):
result = await generate_application_section(
section="executive_summary",
company=test_company,
grant=test_grant,
)
assert "executive summary" in result.lower()
Climatiq Client¶
@pytest.mark.asyncio
async def test_calculate_electricity_emissions():
"""Test emission calculation for electricity."""
mock_client = AsyncMock()
mock_result = MagicMock()
mock_result.co2e = 4.5
mock_result.co2e_unit = "kg"
mock_client.calculate_electricity = AsyncMock(return_value=mock_result)
with patch(
"backend.app.services.climatiq_client.get_climatiq_client",
return_value=mock_client,
):
result = await calculate_scope2_emissions(kwh=10000, region="DE")
assert result.co2e == 4.5
HTTP Clients (Scrapers)¶
@pytest.mark.asyncio
async def test_cordis_fetch_projects():
"""Test CORDIS project fetching with mocked HTTP."""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"results": [
{"rcn": "12345", "title": "Green Energy Project"}
],
"total": 1,
}
mock_response.raise_for_status = MagicMock()
with patch("httpx.AsyncClient.get", new_callable=AsyncMock) as mock_get:
mock_get.return_value = mock_response
config = CordisScraperConfig(requests_per_second=100)
async with CordisScraper(config) as scraper:
projects = await scraper.fetch_projects(framework="HORIZON")
assert len(projects) == 1
assert projects[0].title == "Green Energy Project"
Testing Error Paths¶
Content Filter Errors¶
@pytest.mark.asyncio
async def test_generate_handles_content_filter():
"""Test that content filter errors are properly wrapped."""
mock_client = AsyncMock()
mock_client.generate = AsyncMock(
side_effect=ContentFilterError("Content blocked by safety filter")
)
with patch(
"backend.app.services.claude_client.get_claude_client",
return_value=mock_client,
):
with pytest.raises(ContentFilterError) as exc_info:
await generate_application_section("summary", company, grant)
assert "safety filter" in str(exc_info.value)
Network Errors¶
@pytest.mark.asyncio
async def test_scraper_retries_on_timeout():
"""Test that scrapers retry on timeout errors."""
mock_get = AsyncMock(side_effect=httpx.TimeoutException("Connection timeout"))
with patch("httpx.AsyncClient.get", mock_get):
config = CordisScraperConfig(max_retries=3)
async with CordisScraper(config) as scraper:
with pytest.raises(httpx.TimeoutException):
await scraper.fetch_projects()
# Should have been called max_retries times
assert mock_get.call_count == 3
Testing the Matching Engine¶
The matching engine has specialized test patterns:
@pytest.mark.asyncio
async def test_matching_engine_country_disqualifier(db_session):
"""Country mismatch should return score of 0."""
company = Company(country="DE", nace_codes=["C25.1"])
grant = Grant(countries=["FR"], nace_codes=["C25.1"])
score = calculate_rule_score(company, grant)
assert score == 0.0
@pytest.mark.asyncio
async def test_matching_engine_nace_partial_match(db_session):
"""Same NACE section should give 50% of NACE weight."""
company = Company(country="DE", nace_codes=["C25.1"])
grant = Grant(countries=["DE"], nace_codes=["C28.2"])
score = calculate_rule_score(company, grant)
# Country (0.4) + NACE partial (0.35 * 0.5) + Size open (0.25)
expected = 0.40 + 0.175 + 0.25
assert abs(score - expected) < 0.01
@pytest.mark.asyncio
async def test_carbon_bonus_applied():
"""Carbon-focused grants should get 1.2x bonus."""
# ... test that is_carbon_focused=True results in bonus
Test Data Fixtures¶
Sample Grant Data¶
@pytest.fixture
def sample_grant_data():
"""Standard test grant data."""
return {
"title": "Horizon Europe Clean Energy Innovation",
"description": "Funding for clean energy R&D projects",
"countries": ["DE", "FR", "NL"],
"nace_codes": ["D35.1", "C28.2"],
"company_sizes": ["small", "medium"],
"funding_amount_min": 50000,
"funding_amount_max": 500000,
"deadline": datetime(2025, 6, 30, tzinfo=UTC),
"source": "eu_portal",
"status": "active",
"is_carbon_focused": True,
"carbon_categories": ["renewable_energy", "energy_efficiency"],
}
Sample Company Data¶
@pytest.fixture
def sample_company_data():
"""Standard test company data."""
return {
"name": "GreenTech Solutions GmbH",
"country": "DE",
"nace_codes": ["D35.1"],
"employee_count_max": 45,
"scope1_emissions": 120.5,
"scope2_emissions": 85.3,
"certifications": ["ISO_14001"],
"reduction_target_percent": 30,
"reduction_target_year": 2030,
}
AsyncMock vs MagicMock¶
Common Pitfall
Using MagicMock() for methods that are awaited causes TypeError: object MagicMock can't be used in 'await' expression.
# WRONG
mock_client = MagicMock()
mock_client.request = MagicMock(return_value=response) # Cannot be awaited
# CORRECT
mock_client = MagicMock()
mock_client.request = AsyncMock(return_value=response)
mock_client.aclose = AsyncMock() # Required for async context managers
Rule of thumb: If a method is ever awaited in the source code, it must be an AsyncMock.