Testing Guide¶

This guide covers testing practices and guidelines for ZephCast.

Test Structure¶

tests/
├── unit/
│   ├── aio/
│   │   ├── test_kafka.py
│   │   ├── test_rabbitmq.py
│   │   └── test_redis.py
│   ├── sync/
│   │   ├── test_kafka.py
│   │   ├── test_rabbitmq.py
│   │   └── test_redis.py
│   └── core/
│       ├── test_consumers.py
│       ├── test_retry.py
│       └── test_utils.py
└── integration/
    ├── conftest.py
    ├── aio/
    │   ├── test_kafka.py
    │   ├── test_rabbitmq.py
    │   └── test_redis.py
    └── sync/
        ├── test_kafka.py
        ├── test_rabbitmq.py
        └── test_redis.py

Running Tests¶

All Tests¶

make test

Unit Tests Only¶

make unit-test

Integration Tests Only¶

make integration-test

Test Configuration¶

pytest Configuration¶

The pyproject.toml file contains pytest configuration:

[tool.pytest.ini_options]
asyncio_mode = "strict"
testpaths = ["tests"]
timeout = 30

Environment Variables¶

Set these environment variables for integration tests:

# Kafka
KAFKA_BOOTSTRAP_SERVERS=localhost:9092

# RabbitMQ
RABBITMQ_URL=amqp://guest:guest@localhost:5672/

# Redis
REDIS_URL=redis://localhost:6379

Writing Tests¶

Unit Tests¶

import pytest
from zephcast.aio.kafka.client import KafkaClient
from zephcast.aio.kafka.config import KafkaConfig

@pytest.mark.asyncio
async def test_kafka_client_creation():
    client = KafkaClient(
        stream_name="test-topic",
        config=KafkaConfig(
            bootstrap_servers="localhost:9092"
        )
    )
    assert client.stream_name == "test-topic"
    assert client.config.bootstrap_servers == "localhost:9092"

Integration Tests¶

import pytest
from zephcast.aio.rabbit.client import RabbitClient
from zephcast.aio.rabbit.config import RabbitConfig

@pytest.mark.asyncio
async def test_rabbitmq_send_receive(rabbitmq_client):
    # Send message
    await rabbitmq_client.send("test message")

    # Receive message
    async for message in rabbitmq_client:
        assert message == "test message"
        break

Test Fixtures¶

import pytest
from zephcast.aio.redis.client import RedisClient
from zephcast.aio.redis.config import RedisConfig

@pytest.fixture
async def redis_client():
    client = RedisClient(
        stream_name="test-stream",
        config=RedisConfig(
            redis_url="redis://localhost:6379"
        )
    )
    await client.connect()
    yield client
    await client.close()

Test Categories¶

Functional Tests¶

Test basic functionality:

@pytest.mark.asyncio
async def test_basic_functionality(kafka_client):
    # Test sending and receiving
    message = "test message"
    await kafka_client.send(message)

    async for received in kafka_client:
        assert received == message
        break

Error Tests¶

Test error conditions:

@pytest.mark.asyncio
async def test_connection_error():
    client = KafkaClient(
        stream_name="test-topic",
        config=KafkaConfig(
            bootstrap_servers="invalid:9092"
        )
    )

    with pytest.raises(ConnectionError):
        await client.connect()

Performance Tests¶

Test performance characteristics:

import asyncio
import time

@pytest.mark.asyncio
async def test_batch_performance(kafka_client):
    start_time = time.time()
    message_count = 1000

    # Send batch of messages
    for i in range(message_count):
        await kafka_client.send(f"message-{i}")

    # Calculate throughput
    elapsed = time.time() - start_time
    rate = message_count / elapsed
    assert rate > 100  # Messages per second

Test Best Practices¶

1. Test Isolation¶

Use unique resource names per test
Clean up resources after tests
Don't rely on test execution order

2. Async Testing¶

Use pytest.mark.asyncio
Handle async cleanup properly
Set appropriate timeouts

3. Error Handling¶

Test both success and failure paths
Verify error messages
Test timeout scenarios

4. Resource Management¶

Use fixtures for setup/teardown
Clean up resources in finally blocks
Handle connection cleanup

5. Test Organization¶

Group related tests in classes
Use descriptive test names
Keep tests focused and small

Continuous Integration¶

GitHub Actions¶

name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      kafka:
        image: confluentinc/cp-kafka
      rabbitmq:
        image: rabbitmq:3-management
      redis:
        image: redis

    steps:
      - uses: actions/checkout@v2

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.10'

      - name: Install dependencies
        run: |
          pip install poetry
          poetry install

      - name: Run tests
        run: poetry run pytest

Test Coverage¶

Running Coverage¶

poetry run pytest --cov=zephcast

Coverage Report¶

poetry run pytest --cov=zephcast --cov-report=html

Debugging Tests¶

Using pdb¶

@pytest.mark.asyncio
async def test_with_debugger():
    import pdb; pdb.set_trace()
    # Test code here

Verbose Output¶

poetry run pytest -v --tb=long

Common Issues¶

Hanging Tests
Use timeouts
Check for unclosed resources
Verify async cleanup
Flaky Tests
Add retries for network operations
Use unique resource names
Handle race conditions
Resource Leaks
Use context managers
Clean up in finally blocks
Monitor resource usage