Testing

Unit test services, mock RPC calls and microservices, verify webhooks, and run integration tests using Singularity's built-in testing harness.

Singularity includes a purpose-built testing harness in singularity/testing/ that lets you unit-test services in isolation without starting the server, connecting to a database, or running Redis. The harness provides a mock Acquire container, a ServiceTestClient that mirrors how Manager wires up services at runtime, mock proxies for RPC and Microservice descriptors, and a fake Starlette Request builder for webhook testing.

Test Configuration

pytest + pytest-asyncio Setup

The project is configured for async-first testing in pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["src", "tests"]
filterwarnings = [
  "ignore::pytest.PytestCollectionWarning",
]

Key settings:

Setting	Value	Purpose
`asyncio_mode`	`"auto"`	All `async def test_*` functions are treated as async tests without needing `@pytest.mark.asyncio`
`testpaths`	`["tests"]`	pytest discovers tests only in the `tests/` directory
`pythonpath`	`["src", "tests"]`	Both `src` (production code) and `tests` (testing utilities) are importable

Running Tests

# Run all tests
uv run pytest

# Run with verbose output
uv run pytest -v

# Run a specific test file
uv run pytest tests/test_example.py

# Run a specific test function
uv run pytest tests/test_example.py::test_simple_get

# Run with coverage
uv run pytest --cov=src

The Testing Harness

All testing utilities are importable from testing:

from testing import (
    ServiceTestClient,    # Main test client for services
    TestAcquire,          # Mock Acquire DI container
    MockRPCService,       # Mock for RPC descriptors
    MockMicroservice,     # Mock for Microservice descriptors
    build_webhook_request,  # Fake Starlette Request builder
)

TestAcquire -- Mock Dependency Injection

TestAcquire is a lightweight mock of the Acquire container. It provides the same interface as the real one but backed by MagicMock and AsyncMock instances instead of real connections.

from testing import TestAcquire

# Default: all fields are mocks
acquire = TestAcquire()

# With custom settings
acquire = TestAcquire(settings={
    "environment": "test",
    "jwt_secret": "test-secret-key",
    "stripe_api_key": "sk_test_xxx",
})

# Access mocked resources
acquire.db_session        # AsyncMock
acquire.settings          # SimpleNamespace with your settings
acquire.tasks             # MagicMock (task runner)
acquire.ws_manager        # MagicMock (WebSocket manager)
acquire.cache             # MagicMock
acquire.logger            # MagicMock
acquire.schemas           # dict
acquire.utils             # MagicMock

Custom Overrides

Pass any keyword argument to replace a specific attribute:

from unittest.mock import AsyncMock

custom_db = AsyncMock()
custom_db.execute.return_value.scalars.return_value.all.return_value = [
    {"id": 1, "email": "user@example.com"},
]

acquire = TestAcquire(db_session=custom_db)

conftest.py Fixture

The project's conftest.py provides a fresh TestAcquire for each test:

# tests/conftest.py
import pytest
from testing import TestAcquire

@pytest.fixture
def acquire():
    """Fresh TestAcquire for each test."""
    return TestAcquire()

ServiceTestClient -- Testing Services in Isolation

ServiceTestClient is the primary tool for testing services. It replicates the same setup that Manager performs at runtime:

Creates a TestAcquire (or uses the one you provide)
Instantiates the service class with acquire (if the constructor accepts it)
Injects _service_name for RPC caller identity
Shadows RPC descriptors with mock proxies
Shadows Microservice descriptors with mock proxies

Basic Usage

from testing import ServiceTestClient

class UsersService:
    def __init__(self, acquire):
        self.acquire = acquire

    async def get(self):
        return {"users": [{"id": 1, "name": "Alice"}]}

    async def post(self, data: dict):
        return {"status": "created", "data": data}


async def test_users_get():
    client = ServiceTestClient(UsersService)
    result = await client.get()
    assert "users" in result
    assert result["users"][0]["name"] == "Alice"


async def test_users_post():
    client = ServiceTestClient(UsersService)
    result = await client.post({"name": "Bob", "email": "bob@example.com"})
    assert result["status"] == "created"
    assert result["data"]["name"] == "Bob"

Call Helpers

The client provides async helper methods that delegate to the service instance:

Method	Calls
`client.get(...)`	`service.get(...)`
`client.post(...)`	`service.post(...)`
`client.put(...)`	`service.put(...)`
`client.delete(...)`	`service.delete(...)`
`client.patch(...)`	`service.patch(...)`
`client.call("method_name", ...)`	`service.method_name(...)`
`client.webhook(payload, headers)`	`service.post(fake_request)`

Custom Settings

async def test_with_custom_settings():
    client = ServiceTestClient(UsersService, settings={
        "environment": "test",
        "max_page_size": 50,
    })
    assert client.acquire.settings.environment == "test"
    assert client.acquire.settings.max_page_size == 50

Pre-Built Acquire

async def test_with_prebuilt_acquire():
    acquire = TestAcquire(settings={"debug": True})
    # Configure the mock DB to return specific data
    acquire.db_session.execute.return_value.scalar.return_value = 42

    client = ServiceTestClient(UsersService, acquire=acquire)
    assert client.acquire is acquire

Mocking RPC Calls

When a service calls another service via RPC, the ServiceTestClient intercepts these calls with MockRPCService. You define the expected responses upfront and can assert on the recorded calls afterward.

Setting Up RPC Mocks

from testing import ServiceTestClient


class OrdersService:
    def __init__(self, acquire):
        self.acquire = acquire

    async def post(self, data: dict):
        # This would normally call PaymentsService via RPC
        charge = await self.billing.charge(
            user_id=data["user_id"],
            amount=data["amount"],
        )
        return {"order": "confirmed", "charge": charge}


async def test_order_with_rpc():
    client = ServiceTestClient(OrdersService, rpc={
        "billing": {
            "charge": {"charged": True, "transaction_id": "txn_123"},
        },
    })

    result = await client.post({"user_id": "usr_1", "amount": 29.99})

    # Assert on the service result
    assert result["order"] == "confirmed"
    assert result["charge"]["transaction_id"] == "txn_123"

    # Assert on the RPC calls that were made
    calls = client.rpc_mocks["billing"].calls
    assert len(calls) == 1
    assert calls[0]["method"] == "charge"
    assert calls[0]["kwargs"]["user_id"] == "usr_1"
    assert calls[0]["kwargs"]["amount"] == 29.99

Dynamic RPC Responses

Use a callable for dynamic responses based on input:

async def test_rpc_dynamic_response():
    def mock_get_balance(user_id: str):
        balances = {"usr_1": 100.0, "usr_2": 0.0}
        return {"balance": balances.get(user_id, -1)}

    client = ServiceTestClient(OrdersService, rpc={
        "billing": {"get_balance": mock_get_balance},
    })

Mocking Microservice Calls

For services that use remote Microservice descriptors, the MockMicroservice supports both single-service (direct method calls) and multi-service (chained access) patterns.

Single-Service Mode

class InvoiceService:
    def __init__(self, acquire):
        self.acquire = acquire

    async def post(self, order_id: str):
        invoice = await self.billing.create_invoice(order_id=order_id)
        return {"invoice": invoice}


async def test_microservice_single():
    client = ServiceTestClient(InvoiceService, microservices={
        "billing": {"create_invoice": {"id": "inv_123", "amount": 99.99}},
    })

    result = await client.post("ord_456")

    assert result["invoice"]["id"] == "inv_123"

    calls = client.ms_mocks["billing"].calls
    assert calls[0]["method"] == "create_invoice"
    assert calls[0]["kwargs"]["order_id"] == "ord_456"

Multi-Service Mode (Chained Access)

When the mock value for a service name is a dict, it creates a sub-service with method-level responses:

class FinanceService:
    def __init__(self, acquire):
        self.acquire = acquire

    async def post(self, order_id: str):
        invoice = await self.billing_ms.invoice.create(order_id=order_id)
        customer = await self.billing_ms.customer.get(order_id=order_id)
        return {"invoice": invoice, "customer": customer}


async def test_microservice_chained():
    client = ServiceTestClient(FinanceService, microservices={
        "billing_ms": {
            "invoice": {"create": {"id": "inv_1"}},
            "customer": {"get": {"name": "Acme Corp"}},
        },
    })

    result = await client.post("ord_789")

    assert result["invoice"]["id"] == "inv_1"
    assert result["customer"]["name"] == "Acme Corp"

    # All calls are recorded with service and method names
    calls = client.ms_mocks["billing_ms"].calls
    assert calls[0]["service"] == "invoice"
    assert calls[0]["method"] == "create"

Testing Webhook Services

The harness includes build_webhook_request for creating fake Starlette Request objects and a client.webhook() helper for one-line webhook testing.

Testing Event Dispatch

from singularity.core.webhook import BaseWebhook
from testing import ServiceTestClient


class StripeWebhookService(BaseWebhook):
    events = {
        "invoice.paid": "handle_invoice_paid",
        "customer.created": "handle_customer_created",
    }

    async def handle_invoice_paid(self, payload: dict, headers: dict):
        invoice_id = payload["data"]["object"]["id"]
        return {"status": "processed", "invoice": invoice_id}

    async def handle_customer_created(self, payload: dict, headers: dict):
        return {"status": "processed"}


async def test_invoice_paid_event():
    client = ServiceTestClient(StripeWebhookService)
    response = await client.webhook(
        payload={
            "type": "invoice.paid",
            "data": {"object": {"id": "inv_123"}},
        },
    )
    assert response.status_code == 200

Testing Signature Verification

Override verify() to test that invalid signatures are rejected:

import hmac
import hashlib


class SecureWebhook(BaseWebhook):
    events = {"test.event": "handle_test"}

    def verify(self, headers: dict, raw_body: bytes) -> bool:
        secret = self.acquire.settings.webhook_secret
        signature = headers.get("x-signature", "")
        expected = hmac.new(
            secret.encode(), raw_body, hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(signature, expected)

    async def handle_test(self, payload: dict, headers: dict):
        return {"ok": True}


async def test_webhook_rejects_bad_signature():
    """Verify that an invalid signature returns 401."""
    client = ServiceTestClient(SecureWebhook, settings={
        "webhook_secret": "whsec_test123",
    })
    response = await client.webhook(
        payload={"type": "test.event"},
        headers={"x-signature": "invalid_signature"},
    )
    assert response.status_code == 401


async def test_webhook_accepts_valid_signature():
    """Verify that a correct signature passes verification."""
    import json

    secret = "whsec_test123"
    payload = {"type": "test.event", "data": {"id": "evt_1"}}
    body = json.dumps(payload).encode("utf-8")
    valid_sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()

    client = ServiceTestClient(SecureWebhook, settings={
        "webhook_secret": secret,
    })
    response = await client.webhook(
        payload=payload,
        headers={"x-signature": valid_sig},
    )
    assert response.status_code == 200

Testing Unhandled Events

class MinimalWebhook(BaseWebhook):
    events = {}

async def test_unhandled_event_returns_200():
    """Unhandled events should return 200 OK (acknowledged but ignored)."""
    client = ServiceTestClient(MinimalWebhook)
    response = await client.webhook(payload={"type": "unknown.event"})
    assert response.status_code == 200

Testing Background Tasks

Tasks are plain synchronous functions, so they can be tested directly without Celery.

Unit Testing a Task Function

from tasks.executable.send_email import send_email


def test_send_email_returns_status():
    """Test the task function directly (no Celery involved)."""
    result = send_email(
        to="user@example.com",
        subject="Welcome",
        body="Hello and welcome!",
    )
    assert result["status"] == "sent"
    assert result["to"] == "user@example.com"

Testing Task Submission from a Service

When testing a service that submits tasks, the acquire.tasks mock records the calls:

from testing import ServiceTestClient


class NotificationService:
    def __init__(self, acquire):
        self.acquire = acquire
        self.tasks = acquire.tasks

    async def post(self, user_id: str, message: str):
        self.tasks.submit(
            "send_push_notification",
            user_id=user_id,
            message=message,
        )
        return {"status": "queued"}


async def test_notification_submits_task():
    client = ServiceTestClient(NotificationService)
    result = await client.post("usr_1", "Hello!")

    assert result["status"] == "queued"

    # Verify the task was submitted with correct arguments
    client.acquire.tasks.submit.assert_called_once_with(
        "send_push_notification",
        user_id="usr_1",
        message="Hello!",
    )

Testing on_complete Callbacks

from singularity.tasks import task


@task(name="test_task", notify=["callback"])
def test_task_fn(value: int) -> dict:
    return {"doubled": value * 2}

callback_results = []

@test_task_fn.on_complete
def capture_result(status: str, result: dict, meta: dict):
    callback_results.append({"status": status, "result": result})


def test_on_complete_callback():
    # Directly call the function
    result = test_task_fn(5)
    assert result["doubled"] == 10

    # Simulate what Celery would do: invoke the callback
    test_task_fn.on_complete_fn(
        status="success",
        result=result,
        meta={},
    )
    assert callback_results[-1]["status"] == "success"
    assert callback_results[-1]["result"]["doubled"] == 10

Integration Testing with TestClient

For end-to-end tests that exercise the full HTTP stack (routes, middleware, serialization), use FastAPI's TestClient:

from fastapi.testclient import TestClient
from app import app


def test_health_endpoint():
    """Integration test: verify the health endpoint responds."""
    client = TestClient(app)
    response = client.get("/api/v1/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"


def test_users_endpoint():
    """Integration test: verify the users list endpoint."""
    client = TestClient(app)
    response = client.get("/api/v1/users")
    assert response.status_code == 200
    assert "users" in response.json()

Integration tests with TestClient require a running database and may trigger startup scripts. Use them sparingly and prefer unit tests with ServiceTestClient for most service logic.

Asserting on Mock Calls

Both MockRPCService and MockMicroservice maintain a calls list that records every invocation:

# RPC mock call record structure
{
    "method": "charge",
    "args": (),
    "kwargs": {"user_id": "usr_1", "amount": 29.99},
}

# Microservice mock call record (multi-service mode)
{
    "service": "invoice",
    "method": "create",
    "args": (),
    "kwargs": {"order_id": "ord_123"},
}

Common Assertion Patterns

# Assert a specific method was called
assert any(c["method"] == "charge" for c in client.rpc_mocks["billing"].calls)

# Assert call count
assert len(client.rpc_mocks["billing"].calls) == 2

# Assert no calls were made
assert len(client.rpc_mocks["billing"].calls) == 0

# Assert specific kwargs
call = client.rpc_mocks["billing"].calls[0]
assert call["kwargs"]["amount"] == 29.99

Best Practices

Use ServiceTestClient for unit tests

It mirrors the same setup as Manager -- acquire injection, _service_name, descriptor shadowing -- so your tests exercise the same code paths as production without external dependencies.

Test webhooks with signature verification

Always test both the happy path (valid signature) and the rejection path (invalid signature). Use build_webhook_request to construct fake requests with precise control over headers and body.

Assert on RPC call recordings

Do not just check the service result -- verify that the correct RPC methods were called with the expected arguments. The calls list on mock proxies makes this straightforward.

Keep integration tests separate

Put TestClient integration tests in a dedicated directory or mark them with @pytest.mark.integration. They are slower, require infrastructure, and should not block the fast unit test suite.

Harness API Reference

`ServiceTestClient`

Parameter	Type	Default	Description
`service_class`	`type`	required	The service class to instantiate
`settings`	`dict`	`None`	Settings passed to `TestAcquire`
`rpc`	`dict[str, dict]`	`None`	RPC mock responses keyed by descriptor name
`microservices`	`dict[str, dict]`	`None`	Microservice mock responses keyed by descriptor name
`acquire`	`TestAcquire`	`None`	Pre-built acquire (skips creating a new one)
`service_name`	`str`	`"test"`	Value injected as `_service_name`

`TestAcquire`

Attribute	Type	Description
`settings`	`SimpleNamespace`	Settings object built from provided dict
`db_session`	`AsyncMock`	Mock database session
`tasks`	`MagicMock`	Mock task runner
`ws_manager`	`MagicMock`	Mock WebSocket manager
`cache`	`MagicMock`	Mock cache
`logger`	`MagicMock`	Mock logger
`schemas`	`dict`	Empty dict for schema registry
`utils`	`MagicMock`	Mock utility functions

`build_webhook_request`

def build_webhook_request(
    payload: dict,
    headers: dict[str, str] | None = None,
) -> Request

Creates a fake Starlette Request with the given JSON payload serialized as the body and optional HTTP headers. The content-type header defaults to application/json.

Use ServiceTestClient for unit tests

Test webhooks with signature verification

Assert on RPC call recordings

Keep integration tests separate

On this page