Pytest Adapter Integration Testing
Purpose
Adapters connect your application to external systems. This skill covers testing both unit-level (mocked) and integration-level (real services) scenarios.
When to Use This Skill
Use when testing adapters and gateways with "test Shopify gateway", "mock HTTP responses", "test Kafka adapter", or "decide between mocks and containers".
Do NOT use for domain testing (use pytest-domain-model-testing), application layer (use pytest-application-layer-testing), or general mocking patterns (use pytest-mocking-strategy).
Quick Start
Choose testing strategy based on scope:
# UNIT TEST: Mock HTTP responses
import pytest
from aioresponses import aioresponses
async def test_shopify_gateway_unit() -> None:
"""Unit test with mocked HTTP."""
with aioresponses() as mock_http:
mock_http.get(
"https://test.myshopify.com/api/orders",
payload={"orders": [{"id": "123", "total": 100}]},
)
gateway = ShopifyGateway(url="https://test.myshopify.com")
orders = await gateway.fetch_orders()
assert len(orders) == 1
# INTEGRATION TEST: Real container
from testcontainers.kafka import KafkaContainer
import pytest
@pytest.fixture(scope="module")
def kafka_container():
"""Real Kafka for integration testing."""
with KafkaContainer() as kafka:
yield kafka
async def test_kafka_producer_integration(kafka_container):
"""Integration test with real Kafka."""
producer = KafkaProducerAdapter(kafka_container)
await producer.publish("test_topic", {"data": "value"})
# Verify message was actually publishedInstructions
Step 1: Unit Test with Mocked HTTP Responses
from __future__ import annotations
from aioresponses import aioresponses
from unittest.mock import AsyncMock
import pytest
from app.extraction.adapters.shopify import ShopifyGateway
class TestShopifyGatewayUnit:
"""Unit tests with mocked HTTP responses."""
async def test_fetch_orders_success(self) -> None:
"""Test successful order fetching."""
# Mock Shopify API response
shopify_response = {
"orders": [
{
"id": "12345",
"created_at": "2024-01-01T10:00:00Z",
"customer": {"name": "John Doe"},
"line_items": [
{
"product_id": "prod_1",
"title": "Laptop",
"quantity": 1,
"price": "999.99",
}
],
"total_price": "999.99",
}
]
}
with aioresponses() as mock_http:
mock_http.get(
"https://test.myshopify.com/admin/api/2024-01/orders.json",
payload=shopify_response,
)
gateway = ShopifyGateway(
shop_url="https://test.myshopify.com",
access_token="test_token",
)
# Act
orders = []
async for order in await gateway.fetch_orders():
orders.append(order)
# Assert
assert len(orders) == 1
assert orders[0].customer_name == "John Doe"
async def test_fetch_orders_with_pagination(self) -> None:
"""Test pagination handling."""
# First page response
page1 = {
"orders": [{"id": "1", ...}],
"next_page_url": "https://test.myshopify.com/api/orders?page=2",
}
# Second page response
page2 = {
"orders": [{"id": "2", ...}],
}
with aioresponses() as mock_http:
mock_http.get(
"https://test.myshopify.com/api/orders",
payload=page1,
)
mock_http.get(
"https://test.myshopify.com/api/orders?page=2",
payload=page2,
)
gateway = ShopifyGateway(...)
orders = []
async for order in await gateway.fetch_orders():
orders.append(order)
assert len(orders) == 2
async def test_fetch_orders_handles_api_errors(self) -> None:
"""Test error handling for API failures."""
from app.extraction.adapters.shopify import ShopifyApiException
with aioresponses() as mock_http:
# First request fails, second succeeds (retry)
mock_http.get(
"https://test.myshopify.com/api/orders",
status=500,
)
mock_http.get(
"https://test.myshopify.com/api/orders",
payload={"orders": []},
)
gateway = ShopifyGateway(...)
# Should succeed after retry
orders = []
async for order in await gateway.fetch_orders():
orders.append(order)
assert len(orders) == 0 # Empty but no exception
async def test_rate_limiting_handling(self) -> None:
"""Test rate limit handling."""
with aioresponses() as mock_http:
# Rate limited response
mock_http.get(
"https://test.myshopify.com/api/orders",
status=429,
headers={"Retry-After": "1"},
)
gateway = ShopifyGateway(...)
# Gateway should handle 429 gracefully
# Implementation depends on retry strategyStep 2: Mock Kafka Messages
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock
import pytest
@pytest.fixture
def mock_kafka_message():
"""Factory for creating mock Kafka messages."""
def _create(key: str, value: dict) -> MagicMock:
message = MagicMock()
message.key.return_value = key.encode()
message.value.return_value = json.dumps(value).encode()
message.error.return_value = None
return message
return _create
@pytest.fixture
def mock_kafka_consumer(mock_kafka_message):
"""Mock Kafka consumer for unit tests."""
mock = AsyncMock()
# Return fake messages
messages = [
mock_kafka_message("order_1", {"id": "1", "total": 100}),
mock_kafka_message("order_2", {"id": "2", "total": 200}),
]
mock.consume.return_value = messages
mock.close.return_value = None
return mock
class TestKafkaConsumerAdapter:
"""Test Kafka consumer adapter with mocks."""
async def test_consume_orders(self, mock_kafka_consumer) -> None:
"""Test consuming orders from Kafka."""
adapter = KafkaConsumerAdapter(mock_kafka_consumer)
messages = []
async for msg in adapter.consume():
messages.append(msg)
assert len(messages) == 2
assert messages[0]["id"] == "1"
async def test_handle_deserialization_error(self, mock_kafka_consumer) -> None:
"""Test handling malformed messages."""
# Mock message with invalid JSON
bad_message = MagicMock()
bad_message.value.return_value = b"invalid json"
mock_kafka_consumer.consume.return_value = [bad_message]
adapter = KafkaConsumerAdapter(mock_kafka_consumer)
# Should either skip or raise, depending on implementation
messages = []
async for msg in adapter.consume():
messages.append(msg)
# Verify error was logged/handledStep 3: Mock ClickHouse Client Responses
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
@pytest.fixture
def mock_clickhouse_client():
"""Mock ClickHouse client."""
from clickhouse_driver import Client
mock = MagicMock(spec=Client)
# Mock successful query execution
mock.execute.return_value = [
("Laptop", 100),
("Mouse", 50),
("Keyboard", 30),
]
return mock
class TestClickHouseQueryGateway:
"""Test ClickHouse gateway with mocks."""
def test_query_top_products(self, mock_clickhouse_client) -> None:
"""Test query execution."""
gateway = ClickHouseQueryGateway(mock_clickhouse_client)
results = gateway.query_top_products(limit=10)
assert len(results) == 3
assert results[0][0] == "Laptop"
assert results[0][1] == 100
# Verify correct query was executed
mock_clickhouse_client.execute.assert_called_once()
call_args = mock_clickhouse_client.execute.call_args
assert "ORDER BY" in call_args[0][0] # Query contains ORDER BY
def test_query_empty_result(self, mock_clickhouse_client) -> None:
"""Test query with no results."""
mock_clickhouse_client.execute.return_value = []
gateway = ClickHouseQueryGateway(mock_clickhouse_client)
results = gateway.query_top_products(limit=10)
assert len(results) == 0
def test_query_error_handling(self, mock_clickhouse_client) -> None:
"""Test error handling."""
from clickhouse_driver import Error
mock_clickhouse_client.execute.side_effect = Error("Connection failed")
gateway = ClickHouseQueryGateway(mock_clickhouse_client)
with pytest.raises(Error):
gateway.query_top_products(limit=10)Step 4: Integration Tests with Testcontainers
from __future__ import annotations
from testcontainers.kafka import KafkaContainer
from testcontainers.clickhouse import ClickHouseContainer
import pytest
@pytest.fixture(scope="module")
def kafka_container():
"""Real Kafka container for integration tests."""
with KafkaContainer() as kafka:
yield kafka
@pytest.fixture(scope="module")
def clickhouse_container():
"""Real ClickHouse container for integration tests."""
with ClickHouseContainer() as ch:
yield ch
class TestKafkaIntegration:
"""Integration tests with real Kafka."""
async def test_produce_and_consume(self, kafka_container) -> None:
"""Test real message flow."""
from confluent_kafka import Producer, Consumer
# Create real producer
producer = Producer({
"bootstrap.servers": kafka_container.get_bootstrap_server(),
})
# Produce message
producer.produce(
"test_topic",
key="order_1",
value=b'{"id": "1", "total": 100}',
)
producer.flush()
# Create real consumer
consumer = Consumer({
"bootstrap.servers": kafka_container.get_bootstrap_server(),
"group.id": "test_group",
"auto.offset.reset": "earliest",
})
consumer.subscribe(["test_topic"])
# Consume message
messages = []
for _ in range(1):
msg = consumer.poll(timeout=5)
if msg:
messages.append(msg.value())
consumer.close()
assert len(messages) == 1
async def test_multiple_partitions(self, kafka_container) -> None:
"""Test handling multiple partitions."""
# Create topic with multiple partitions
# Produce messages to different partitions
# Verify all messages are consumedStep 5: Integration Tests with Real ClickHouse
def test_schema_initialization(clickhouse_container) -> None:
"""Test creating schema in real ClickHouse."""
from clickhouse_driver import Client
client = Client(
host=clickhouse_container.get_container_host_ip(),
port=clickhouse_container.get_exposed_port(8123),
)
# Create test database
client.execute("CREATE DATABASE IF NOT EXISTS test_db")
# Create table
client.execute("""
CREATE TABLE IF NOT EXISTS test_db.products (
title String,
cnt_bought UInt32,
)
ENGINE = MergeTree()
ORDER BY cnt_bought DESC
""")
# Insert data
client.execute(
"INSERT INTO test_db.products VALUES",
[("Laptop", 100), ("Mouse", 50)],
)
# Query data
results = client.execute("SELECT * FROM test_db.products")
assert len(results) == 2
assert results[0][0] == "Mouse" # Sorted by cnt_bought DESC
# Cleanup
client.execute("DROP TABLE test_db.products")
client.execute("DROP DATABASE test_db")Step 6: Test Error Paths
class TestErrorHandling:
"""Test adapter error handling."""
async def test_timeout_handling(self) -> None:
"""Test timeout errors."""
from aioresponses import aioresponses
import asyncio
with aioresponses() as mock_http:
mock_http.get(
"https://test.myshopify.com/api/orders",
exception=asyncio.TimeoutError("Request timeout"),
)
gateway = ShopifyGateway(...)
with pytest.raises(asyncio.TimeoutError):
await gateway.fetch_orders()
async def test_connection_error(self) -> None:
"""Test connection errors."""
from aioresponses import aioresponses
with aioresponses() as mock_http:
mock_http.get(
"https://test.myshopify.com/api/orders",
exception=ConnectionError("Unable to connect"),
)
gateway = ShopifyGateway(...)
with pytest.raises(ConnectionError):
await gateway.fetch_orders()
async def test_invalid_response_format(self) -> None:
"""Test handling malformed responses."""
from aioresponses import aioresponses
with aioresponses() as mock_http:
mock_http.get(
"https://test.myshopify.com/api/orders",
payload={"invalid_format": "missing_orders_key"},
)
gateway = ShopifyGateway(...)
with pytest.raises(KeyError):
async for order in await gateway.fetch_orders():
passExamples
Example 1: Complete Adapter Unit Test
class TestShopifyGatewayComplete:
"""Comprehensive adapter unit tests."""
@pytest.mark.asyncio
async def test_successful_order_fetch(self) -> None:
"""Happy path test."""
with aioresponses() as mock_http:
mock_http.get("https://api.example.com/orders", payload={
"orders": [{"id": "1", "total": 100}]
})
gateway = ShopifyGateway("https://api.example.com")
orders = [o async for o in await gateway.fetch_orders()]
assert len(orders) == 1
@pytest.mark.asyncio
async def test_error_recovery(self) -> None:
"""Test retry logic."""
with aioresponses() as mock_http:
mock_http.get("...", status=500)
mock_http.get("...", payload={"orders": []})
gateway = ShopifyGateway(...)
orders = [o async for o in await gateway.fetch_orders()]
assert len(orders) == 0Example 2: Choose Test Strategy Matrix
Adapter Unit Test (Mock) Integration (Container)
─────────────────────────────────────────────────────────────────────
Shopify API ✅ aioresponses ❌ (use mock)
Kafka Producer ✅ AsyncMock ✅ KafkaContainer
Kafka Consumer ✅ AsyncMock ✅ KafkaContainer (preferred)
ClickHouse ✅ MagicMock(Client) ✅ ClickHouseContainer (preferred)
HTTP Clients ✅ aioresponses/responses ❌ (use mock)Requirements
- Python 3.11+
- pytest >= 7.0
- pytest-asyncio >= 0.20.0
- aioresponses (for async HTTP mocking)
- testcontainers >= 3.7.0 (for integration tests)
See Also
- pytest-mocking-strategy - Mocking with AsyncMock
- pytest-async-testing - Async testing patterns
- PROJECT_UNIT_TESTING_STRATEGY.md - Section: "Adapter Layer Testing"
- PYTHON_UNIT_TESTING_BEST_PRACTICES.md - Section: "Mocking Strategy"