Testing Plugins

fapilog provides comprehensive testing utilities to help you develop and validate custom plugins.

Installation

Basic utilities (mocks, validators, factories, benchmarks) are included in the main package:

from fapilog.testing import (
    # Mock plugins
    MockSink,
    MockEnricher,
    MockRedactor,
    MockProcessor,
    # Validators
    validate_sink,
    validate_enricher,
    validate_redactor,
    validate_processor,
    validate_plugin_lifecycle,
    # Factories
    create_log_event,
    create_batch_events,
    create_sensitive_event,
)

For pytest fixtures, install the testing extra:

pip install fapilog[testing]

This enables additional fixtures like mock_sink, mock_enricher, and assertion helpers.

Mock Plugins

MockSink

A configurable mock sink that captures written events for inspection:

from fapilog.testing import MockSink, MockSinkConfig

# Basic usage
sink = MockSink()
await sink.start()
await sink.write({"level": "INFO", "message": "test"})

assert len(sink.events) == 1
assert sink.events[0]["message"] == "test"

# With failure simulation
config = MockSinkConfig(
    fail_after=5,  # Fail after 5 writes
    fail_with=IOError("connection lost"),
    latency_seconds=0.1,  # Simulate network latency
)
sink = MockSink(config)

# Assertion helpers
sink.assert_event_count(1)
sink.assert_event_contains(0, level="INFO", message="test")

# Reset for reuse
sink.reset()

MockEnricher

A mock enricher that adds configurable fields:

from fapilog.testing import MockEnricher, MockEnricherConfig

config = MockEnricherConfig(
    fields_to_add={"service": "test", "version": "1.0.0"},
    latency_seconds=0.05,
)
enricher = MockEnricher(config)

result = await enricher.enrich({"message": "hello"})
assert result == {"service": "test", "version": "1.0.0"}

MockRedactor

A mock redactor that masks specified fields:

from fapilog.testing import MockRedactor, MockRedactorConfig

config = MockRedactorConfig(
    fields_to_mask=["user.password", "auth.token"],
    mask_string="***REDACTED***",
)
redactor = MockRedactor(config)

event = {"user": {"name": "alice", "password": "secret"}}
result = await redactor.redact(event)
assert result["user"]["password"] == "***REDACTED***"

MockProcessor

A mock processor for testing memoryview operations:

from fapilog.testing import MockProcessor

processor = MockProcessor()
view = memoryview(b'{"message": "test"}')
result = await processor.process(view)

assert processor.call_count == 1

Protocol Validators

Use validators to verify your plugins implement protocols correctly:

from fapilog.testing import validate_sink, validate_enricher

class MyCustomSink:
    name = "my-sink"
    
    async def start(self) -> None:
        pass
    
    async def stop(self) -> None:
        pass
    
    async def write(self, entry: dict) -> None:
        print(entry)


def test_my_sink_protocol():
    sink = MyCustomSink()
    result = validate_sink(sink)
    
    assert result.valid, f"Validation failed: {result.errors}"
    assert not result.warnings

What Validators Check

Each validator checks:

name attribute - Must be present and be a string
Required methods - Must exist and be async
Method signatures - Must accept correct parameters

validate_plugin_lifecycle

Test that lifecycle methods work correctly:

import pytest
from fapilog.testing import validate_plugin_lifecycle

@pytest.mark.asyncio
async def test_my_sink_lifecycle():
    sink = MyCustomSink()
    result = await validate_plugin_lifecycle(sink)
    
    assert result.valid
    # Checks:
    # - start() doesn't raise
    # - stop() doesn't raise
    # - stop() is idempotent (can be called twice)

Test Data Factories

Generate realistic test data for your tests:

from fapilog.testing import (
    create_log_event,
    create_batch_events,
    create_sensitive_event,
)

# Single event with defaults
event = create_log_event()
# {"level": "INFO", "message": "Test message 1234", ...}

# With custom values
event = create_log_event(
    level="ERROR",
    message="Something went wrong",
    user_id="user-123",
)

# Batch of events
events = create_batch_events(count=100, level="DEBUG")

# Event with sensitive data (for redaction testing)
sensitive = create_sensitive_event()
# Contains: password, ssn, card_number, api_key, etc.

Testing Patterns

Testing a Custom Sink

import pytest
from fapilog.testing import validate_sink, create_log_event


class MyDatabaseSink:
    name = "my-database"
    
    def __init__(self, connection_string: str):
        self._conn_str = connection_string
        self._client = None
    
    async def start(self) -> None:
        self._client = await connect(self._conn_str)
    
    async def stop(self) -> None:
        if self._client:
            await self._client.close()
    
    async def write(self, entry: dict) -> None:
        await self._client.insert(entry)
    
    async def health_check(self) -> bool:
        return self._client is not None and self._client.is_connected()


def test_protocol_compliance():
    sink = MyDatabaseSink("sqlite:///:memory:")
    result = validate_sink(sink)
    result.raise_if_invalid()


@pytest.mark.asyncio
async def test_write_events():
    sink = MyDatabaseSink("sqlite:///:memory:")
    await sink.start()
    
    event = create_log_event(level="INFO", message="test")
    await sink.write(event)
    
    # Verify event was written
    # ...
    
    await sink.stop()

Testing a Custom Enricher

import time
import pytest
from fapilog.testing import validate_enricher, create_log_event


class MyEnricher:
    name = "my-enricher"
    
    async def start(self) -> None:
        pass
    
    async def stop(self) -> None:
        pass
    
    async def enrich(self, event: dict) -> dict:
        return {"enriched_at": time.time()}


def test_protocol_compliance():
    enricher = MyEnricher()
    result = validate_enricher(enricher)
    assert result.valid


@pytest.mark.asyncio
async def test_enrichment():
    enricher = MyEnricher()
    event = create_log_event()
    
    additions = await enricher.enrich(event)
    
    assert "enriched_at" in additions
    assert isinstance(additions["enriched_at"], float)

Testing Redaction

import pytest
from fapilog.testing import create_sensitive_event


@pytest.mark.asyncio
async def test_redacts_sensitive_fields():
    redactor = MyRedactor(fields=["password", "ssn"])
    event = create_sensitive_event()
    
    result = await redactor.redact(event)
    
    assert result["user"]["password"] != "supersecret123"
    assert result["user"]["ssn"] != "123-45-6789"

Test Isolation with Logger Caching

Since fapilog caches logger instances by name, tests need isolation to avoid shared state. Two approaches:

Option 1: Use `reuse=False` (Recommended)

Create independent logger instances per test:

@pytest.mark.asyncio
async def test_my_feature():
    # reuse=False creates a fresh instance not added to cache
    logger = await get_async_logger("test", reuse=False)

    await logger.info("test message")

    # Clean up when done
    await logger.drain()

Option 2: Clear Cache in Fixtures

Clear the cache before/after tests:

import pytest
from fapilog import clear_logger_cache

@pytest.fixture(autouse=True)
async def isolate_logger_cache():
    await clear_logger_cache()
    yield
    await clear_logger_cache()

The fapilog test suite uses an autouse fixture in conftest.py that clears the cache between tests.

Best Practices

Always validate protocol compliance before testing behavior
Use mock plugins to test integration without external dependencies
Use factories for consistent, realistic test data
Test lifecycle methods to ensure proper resource management
Test error handling by configuring mocks to fail
Test idempotency - call stop() twice to verify it doesn’t break
Use reuse=False when creating loggers in tests that need isolation

Migration Notes

As of fapilog 0.4.0, all plugin protocols require a name attribute:

# Before (may fail validation)
class MySink:
    async def write(self, entry: dict) -> None:
        ...

# After (correct)
class MySink:
    name = "my-sink"  # Required!
    
    async def write(self, entry: dict) -> None:
        ...

If you have existing plugins without name, add it as a class attribute with a unique identifier for your plugin.