Skip to content

Test Query Shape

DDD CQRS ES

Functional tests confirm a query returns the right rows, but they say nothing about how expensive it was. A query can return correct results while issuing an extra round trip per row, wrapping a count in a subquery, or fetching three times more rows than it needs. On small test datasets these pathologies are invisible; in production they are the difference between a healthy queue and a stalled one.

Protean ships three pytest context managers that assert query shape and round-trip count, so a cost regression fails at PR time instead of in production. A fourth, capture_queries, yields the raw statements for custom assertions the three do not cover (such as INSERT ordering):

# fragment
import re

import pytest
from ticketing import domain

from protean.integrations.pytest import (
    assert_no_overfetch,
    assert_no_subquery_wrap,
    assert_query_count,
    capture_queries,
)

When to use these

Reach for query-shape assertions on hot read paths where cost matters: poll loops, claim-and-process paths, dashboards, and any repository method that runs under load. They complement functional tests rather than replace them: assert what a query returns with ordinary assertions, and how it runs with these.

They are SQLAlchemy-specific. When the active provider is the in-memory adapter there is no engine to observe, so the context managers are no-ops and assert nothing. Mark tests that use them with @pytest.mark.database (or a specific backend marker) so they run against a real SQL backend where the assertion is meaningful. Protean's pytest plugin does not register database, so register it in your conftest.py, as the fixtures below do, or pytest warns about an unknown mark.

The examples on this page run against SQLite, with this domain and these fixtures:

The domain and fixtures under test 
# fragment
from protean import Domain
from protean.fields import Float, HasMany, Integer, String

domain = Domain(name="Shop")


@domain.entity(part_of="Order")
class OrderItem:
    product_id: String(max_length=50, required=True)
    quantity: Integer(min_value=1, required=True)


@domain.aggregate
class Ticket:
    title: String(max_length=100, required=True)
    status: String(max_length=20, default="PENDING")


@domain.repository(part_of=Ticket)
class TicketRepository:
    def find_pending(self, limit):
        query = self._dao.query.filter(status="PENDING").limit(limit)
        return query.all(with_total=False).items

    def count_pending(self):
        return self._dao.query.filter(status="PENDING").count()


@domain.aggregate
class Order:
    customer_name: String(max_length=100, required=True)
    status: String(max_length=20, default="PENDING")
    total_amount: Float(default=0.0)
    items = HasMany(OrderItem)

# fragment
import pytest
from ticketing import Order, OrderItem, Ticket, domain

from protean.integrations.pytest import DomainFixture


def pytest_configure(config):
    config.addinivalue_line("markers", "database: runs against a real SQL backend")


@pytest.fixture(scope="session")
def app_fixture(tmp_path_factory):
    # Query-shape assertions need a SQLAlchemy provider. SQLite needs no server.
    database = tmp_path_factory.getbasetemp() / "shop.db"
    domain.config["databases"]["default"] = {
        "provider": "sqlite",
        "database_uri": f"sqlite:///{database}",
    }

    fixture = DomainFixture(domain)
    fixture.setup()
    yield fixture
    fixture.teardown()


@pytest.fixture(autouse=True)
def _ctx(app_fixture):
    with app_fixture.domain_context():
        yield


@pytest.fixture
def ticket_repo():
    return domain.repository_for(Ticket)


@pytest.fixture
def seeded_tickets(ticket_repo):
    for number in range(20):
        ticket_repo.add(Ticket(title=f"Ticket {number}"))


@pytest.fixture
def order_repo():
    return domain.repository_for(Order)


@pytest.fixture
def order():
    return Order(
        customer_name="Alice",
        items=[OrderItem(product_id="dune", quantity=2)],
    )

assert_query_count

Asserts an exact number of queries (data round trips) are issued in the block. Catches N+1 patterns and accidental extra round trips:

# fragment
@pytest.mark.database
def test_poll_issues_a_single_query(ticket_repo, seeded_tickets):
    with assert_query_count(1):
        ticket_repo.find_pending(limit=10)

A "query" is a SELECT, INSERT, UPDATE, DELETE, or WITH statement. Connection setup (such as a PRAGMA or SET) and transaction control (BEGIN/COMMIT) are not counted, so adapters that issue per-connection pragmas do not inflate the number. The context manager yields the captured statements, so a failing assertion prints exactly what ran.

assert_no_subquery_wrap

Fails if any query wraps a count around a subquery, the SELECT count(*) FROM (SELECT ... ) AS anon_1 shape, where a flat SELECT count(*) would do. The first test guards a repository count against that regression. The second issues a wrapped count on purpose and shows the assertion catches it:

# fragment
@pytest.mark.database
def test_count_uses_a_flat_count(ticket_repo):
    with assert_no_subquery_wrap():
        ticket_repo.count_pending()


@pytest.mark.database
def test_a_wrapped_count_fails(ticket_repo):
    engine = domain.providers["default"]._engine
    wrapped = "SELECT count(*) FROM (SELECT id FROM ticket LIMIT 5) AS anon_1"

    with pytest.raises(AssertionError):
        with assert_no_subquery_wrap():
            with engine.connect() as connection:
                connection.exec_driver_sql(wrapped)

assert_no_overfetch

Fails if any LIMIT exceeds the rows the caller actually needs. Pass the expected row count; the assertion allows headroom via ratio (default 1.5):

# fragment
@pytest.mark.database
def test_poll_does_not_overfetch(ticket_repo, seeded_tickets):
    with assert_no_overfetch(expected_returned=10):
        ticket_repo.find_pending(limit=10)

A LIMIT 10 passes (10 <= 10 * 1.5); a min(limit * 3, 1000) over-fetch that emits LIMIT 30 fails. Widen ratio when a path legitimately reads a small multiple of what it returns.

capture_queries

When the three assertions above do not fit, capture_queries yields the raw statements a block issues so you can write a custom assertion. It powers, for example, an insert-ordering check that a parent row is written before its children:

# fragment
@pytest.mark.database
def test_parent_inserted_before_children(order_repo, order):
    with capture_queries() as captured:
        order_repo.add(order)

    tables = [
        re.search(r'INSERT INTO "?(\w+)', statement, re.IGNORECASE).group(1).lower()
        for statement, _params in captured
        if statement.lstrip().upper().startswith("INSERT")
    ]
    assert tables, "no INSERTs captured; the hook did not fire"
    assert tables.index("order") < tables.index("order_item")

capture_queries yields a list of (statement, parameters) tuples. Like the assertions above it is a no-op on the in-memory adapter, where there is no engine to observe, so the list is empty. Guard on that: on a real backend an empty capture means the hook never fired, so assert the list is non-empty before trusting an ordering check.

Engine resolution

By default the engine is resolved from the active domain's default provider. For tests that manage their own domain or target a non-default provider, pass the engine explicitly:

# fragment
@pytest.mark.database
def test_poll_on_an_explicit_engine(ticket_repo):
    my_engine = domain.providers["default"]._engine

    with assert_query_count(1, engine=my_engine):
        ticket_repo.find_pending(limit=10)

Avoiding false positives

  • Scope the block tightly: Only wrap the call under test. Seeding, fixture setup, and assertions that themselves query should sit outside the block.
  • Count what the operation issues, not the framework: A .all() with the default with_total=True issues a second COUNT query; use with_total=False when you mean to assert a single round trip.
  • These are deterministic, not statistical: They catch shape regressions, not latency. For timing-based regression detection, use a benchmark instead.