---

name: pytest-adapter-integration-testing description: | Testing adapters and gateways: when to use unit test mocks vs testcontainers, mocking HTTP responses (aioresponses, responses), mocking Kafka messages, mocking ClickHouse responses, testing error paths and edge cases.

Use when: Testing gateways and adapters, testing external service integrations, deciding between mocks and real containers, testing error handling in adapters, integration testing with real services. allowed-tools: Read, Bash, Write

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.

Quick Start

Choose testing strategy based on scope:

Copy# 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 published

Instructions

Step 1: Unit Test with Mocked HTTP Responses

Copyfrom __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 strategy

Step 2: Mock Kafka Messages

Copyfrom __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/handled

Step 3: Mock ClickHouse Client Responses

Copyfrom __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

Copyfrom __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 consumed

Step 5: Integration Tests with Real ClickHouse

Copydef 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

Copyclass 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():
                    pass

Examples

Example 1: Complete Adapter Unit Test

Copyclass 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) == 0

Example 2: Choose Test Strategy Matrix

CopyAdapter           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"