Skip to content
Prague
Improve this page

Test without a Broker

Goal: fast, deterministic tests for everything that reads a cache, with Kafka nowhere in the picture.

Prerequisite: none beyond the generated caches. The code is in TestWithoutABroker.cs; every block runs as a test. All test projects in the repository use NUnit; the patterns are framework-neutral.

The code under test

A read model that takes two caches and knows nothing about where their data comes from. That is the seam: in production AddKafkaCaches fills them, in a test you do.

/// <summary>The code under test: a read model over two caches. It neither knows nor cares where the data came from.</summary>
public sealed class CustomerStatement(OrderCache orders, CustomerCache customers) {
    public string? Describe(int customerId) {
        if (!customers.Cache.TryGet(customerId, out var customer))
            return null;
        using var open = orders.Query()
            .WithCustomerId(customerId)
            .Or(b => b.WithStatus(OrderStatus.Pending), b => b.WithStatus(OrderStatus.Paid))
            .ExecutePooled();
        return $"{customer.Name}: {open.Count} open order(s), {open.Sum(o => o.Total):0.00} outstanding";
    }
}

1. A stand-alone cache

new OrderCache() is a complete cache. AddOrUpdate is the same method the consumer calls per message; pass a timestamp when the last-update index matters.

var orders = new OrderCache();                          // no registry, no Kafka, no DI
orders.AddOrUpdate(NewOrder("ORD-001", 42, OrderStatus.Pending, 120m));
orders.AddOrUpdate(NewOrder("ORD-002", 42, OrderStatus.Cancelled, 80m), timestamp: 1_700_000_000_000);   // explicit timestamp

using var open = orders.Query().WithCustomerId(42).WithStatus(OrderStatus.Pending).ExecutePooled();

Assert.That(open.Count, Is.EqualTo(1));
Assert.That(open[0].OrderId, Is.EqualTo("ORD-001"));
Assert.That(open.Select(o => o.Total), Is.EqualTo(new[] { 120m }));   // QueryResults<T> is an IReadOnlyList<T>

2. The registry, for joins and global indices

Joins and global last-update indices are wired when caches are built together. DataCacheRegistryBuilder.Register<OrderCache>() pulls in every cache the foreign keys reach.

var registry = new DataCacheRegistryBuilder().Register<OrderCache>().Build();   // wires FK joins + CustomerActivityIndex
var orders = registry.GetCache<OrderCache>();
var customers = registry.GetCache<CustomerCache>();
customers.AddOrUpdate(new Customer { Id = 42, Email = "ada@example.com", Country = "GB", Tier = CustomerTier.Gold, Name = "Ada" });
orders.AddOrUpdate(NewOrder("ORD-001", 42, OrderStatus.Paid, 120m));

var statement = new CustomerStatement(orders, customers);

Assert.That(statement.Describe(42), Is.EqualTo("Ada: 1 open order(s), 120.00 outstanding"));
Assert.That(statement.Describe(7), Is.Null);

3. After-handlers

An ICacheAfterHandler has one method; call it with the UpdateType sequence a topic would produce (see React to changes for the arguments per case).

var projector = new OrderProjector();                   // the ICacheAfterHandler<string, Order> from "React to changes"
var order = NewOrder("ORD-001", 42, OrderStatus.Pending, 100m);

await projector.Handle(UpdateType.Add, order.OrderId, order, null);
await projector.Handle(UpdateType.Delete, order.OrderId, null, order);

Assert.That(projector.OpenTotalByCustomer[42], Is.Zero);

4. Asserting on pooled results

QueryResults<T> is an IReadOnlyList<T>, so NUnit's collection constraints work on it directly. Assert the set, not the order, unless the query sorts.

using var rows = orders.Query().WithCustomerId(42).ExecutePooled();   // `using`: the buffer goes back when the test ends

Assert.That(rows.Count, Is.EqualTo(2));
Assert.That(rows.Select(o => o.OrderId), Is.EquivalentTo(new[] { "ORD-001", "ORD-002" }));   // order is not asserted
Assert.That(rows.TotalCount, Is.EqualTo(2));

var ids = rows.Map(o => o.OrderId);                     // consumes `rows`; the later `using` Dispose is a no-op
using (ids)
    Assert.That(ids.Count, Is.EqualTo(2));

5. Frozen against eager

A frozen query is a plan; the eager query is its specification. Compare them over several argument values, as sets when unsorted.

var frozen = orders.Prepare<int>().WithCustomerId(static c => c).WithStatus(OrderStatus.Paid).BuildFrozen();

foreach (var customerId in new[] { 0, 1, 2, 99 }) {
    using var expected = orders.Query().WithCustomerId(customerId).WithStatus(OrderStatus.Paid).ExecutePooled();
    using var actual = frozen.ExecutePooled(customerId);
    // Same rows, same counts; the unsorted order is not part of the frozen contract, so compare as sets.
    Assert.That(actual.Select(o => o.OrderId), Is.EquivalentTo(expected.Select(o => o.OrderId)));
    Assert.That(actual.TotalCount, Is.EqualTo(expected.TotalCount));
}

6. The real container, not started

AddKafkaCaches contacts no broker until the hosted service starts. Build the provider, resolve the cache singletons and seed them: your services then run against the same instances the app would inject.

var services = new ServiceCollection();
services.AddSingleton<IConfiguration>(new ConfigurationBuilder().Build());
services.AddLogging();
services.AddKafkaCaches(o => o.BootstrapServers = "unreachable:9092",   // never contacted: no hosted service starts
    b => b.AddCache<OrderCache, string, Order>().WithAfterHandler<OrderProjector>());
// CustomerCache rides in through the FK: it is in the registry but not a DI service until you expose it.
services.AddSingleton(sp => sp.GetRequiredService<IDataCacheRegistry>().GetCache<CustomerCache>());
services.AddSingleton<CustomerStatement>();

using var provider = services.BuildServiceProvider();   // do NOT resolve/start IHostedService in a unit test
var orders = provider.GetRequiredService<OrderCache>();
orders.AddOrUpdate(NewOrder("ORD-001", 42, OrderStatus.Paid, 10m));   // seed the real singleton the app would use

Assert.That(provider.GetRequiredService<ICacheAfterHandler<string, Order>>(), Is.TypeOf<OrderProjector>());
Assert.That(provider.GetRequiredService<CustomerStatement>().Describe(42), Is.Null);   // no customer seeded

When you do want a broker

The repository's Prague.Kafka.IntegrationTests start Kafka with Testcontainers and exercise the consumer end to end: initial load, live updates, tombstones, filters, after-handlers, health. Without Docker they fail on the socket within seconds, which is the only reason they fail. Use that project as the template when a test has to cover deserialization or topic semantics; everything above it belongs in a unit test.

Pitfalls

  • Asserting order on an unsorted frozen result. It carries no row-order guarantee. Use Is.EquivalentTo, or add a total sort.
  • Forgetting using on pooled results in tests. The test passes and the buffer leaks until the pool refills; under a long test run that shows up as allocation noise. The using is free.
  • Resolving IHostedService in a unit test. It starts the consumer. Build the provider and stop there.
  • Caches pulled in by foreign keys are not DI services. Register a factory from the registry if the code under test injects them (step 6 does).
Prague documentation