Testing Your App

Testing applications that use Rdakt AI requires mocking or replacing the HTTP transport layer. Here are patterns for unit and integration testing.

Unit Testing with a Fake Transport

Create a fake transport that returns a predefined response:

import httpx

class FakeTransport(httpx.AsyncBaseTransport):
    """Return a fixed response for testing."""

    def __init__(self, response_body: bytes, status_code: int = 200):
        self.response_body = response_body
        self.status_code = status_code
        self.last_request: httpx.Request | None = None

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        self.last_request = request
        return httpx.Response(
            status_code=self.status_code,
            content=self.response_body,
            headers={"content-type": "application/json"},
        )

Testing Anonymization

Verify that the middleware anonymizes the request before sending:

import json
from rdakt_ai import RdaktMiddleware

async def test_request_is_anonymized():
    fake = FakeTransport(response_body=b'{"choices": []}')
    middleware = RdaktMiddleware(inner=fake)

    async with httpx.AsyncClient(transport=middleware) as client:
        await client.post(
            "https://api.openai.com/v1/chat/completions",
            json={
                "model": "gpt-4o",
                "messages": [{"role": "user", "content": "Email me at john@example.com"}],
            },
        )

    # Inspect what was actually sent
    sent_body = json.loads(fake.last_request.content)
    sent_text = sent_body["messages"][0]["content"]

    assert "john@example.com" not in sent_text
    assert "<EMAIL_1>" in sent_text

Testing Streaming Deanonymization

For streaming tests, return raw SSE bytes:

async def test_streaming_deanonymization():
    sse_data = (
        b'data: {"choices":[{"delta":{"content":"Hello <PERSON_1>"}}]}\n\n'
        b'data: [DONE]\n\n'
    )
    fake = FakeTransport(
        response_body=sse_data,
        status_code=200,
    )
    # Override content-type for SSE
    fake.headers = {"content-type": "text/event-stream"}

    middleware = RdaktMiddleware(inner=fake, session_key="test")
    # Pre-populate the session with a mapping
    # ... (details depend on your session setup)

Testing Detection Independently

Test the detection pipeline without the middleware:

from rdakt_ai import RdaktPipeline, DetectorStage, RegexDetector

def test_regex_detection():
    pipeline = RdaktPipeline(stages=[
        DetectorStage(RegexDetector()),
    ])

    entities = pipeline.detect_sync("Contact john@example.com or call (555) 123-4567")

    assert len(entities) == 2
    types = {e.type for e in entities}
    assert types == {"EMAIL", "PHONE"}

Testing Anonymization Independently

from rdakt_ai import Anonymizer, AnonymizationStrategy, Entity, RdaktSession

def test_anonymization():
    anonymizer = Anonymizer(default_strategy=AnonymizationStrategy.TOKEN)
    session = RdaktSession()

    entities = [Entity(value="john@example.com", type="EMAIL", start=0, end=16)]
    text, mapping = anonymizer.anonymize("john@example.com", entities, session=session)

    assert text == "<EMAIL_1>"
    assert mapping["<EMAIL_1>"] == "john@example.com"

Integration Testing

For full integration tests that hit the actual LLM API, use audit mode to verify detection without sending anonymized data:

async def test_full_integration_audit():
    detected: list = []

    middleware = RdaktMiddleware(
        mode="audit",
        on_entities_detected=lambda entities, text: detected.extend(entities),
    )

    client = AsyncOpenAI(
        http_client=httpx.AsyncClient(transport=middleware),
    )

    await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "My SSN is 123-45-6789"}],
    )

    assert any(e.type == "SSN" for e in detected)