Reactive Test Assertions

MeshWeaver is reactive end-to-end: services, handlers, layout areas, and activities return IObservable<T> and never await (see Asynchronous Calls). Its tests follow the same principle — you assert on the stream directly, and the assertion OWNS the wait instead of your test body.

All test assertions flow through MeshWeaver.Reactive.Assertions, an in-house library (it replaced FluentAssertions — that package is no longer referenced) wired in via a global using and project reference in test/Directory.Build.props. The names and chaining patterns (.And / .Which, trailing because args) are familiar. What is different is that the observable assertions own the wait: each terminal method returns a Task that the test body awaits, so a test method is async Task and contains no .FirstAsync().ToTask() — none at all, anywhere — no .Result, no .Wait() and no Task.Delay.

🚨 .ToTask() is forbidden repo-wide, test/** included (maintainer ruling, 2026-08-30: "totask is forbidden", "no totask ever" — the earlier "tests are the one sanctioned place" carve-out is RETRACTED). Rx's bridge completes its TaskCompletionSource without TaskCreationOptions.RunContinuationsAsynchronously, so it resumes the awaiter inline on the signalling thread, still inside Rx's trampoline. In a test that thread then carries the rest of the test, its mesh teardown, and — under xUnit — the runner starting the next test class, so the green proves the wrong thing. await source.FirstAsync() is not a lighter alternative: Rx's own awaiter is an AsyncSubject<T> that completes its continuation from inside OnCompleted, and measures the same (INLINE=True for both spellings; only the queued completion below measures INLINE=False). Where an assertion genuinely cannot express the wait, the bridge is MeshWeaver.Messaging.ReactiveCompletion.ObserveCompletion(reportLateFault, ct).

🚨 The terminal assertions are Task-returning, not blocking. Emit() / Match() return Task<T>; Be() / Complete() / NotEmit() return Task<ObservableAssertions<T>>. Every one must be awaited. An un-awaited obs.Should().Emit(); statement compiles without a warning, subscribes, and then lets the test race on — the classic green-but-lying test. ObservableAssertions.cs says so in its own doc comment: the wait is "never a thread-blocking ManualResetEventSlim + Wait".

For the surrounding test-writing rules, see Writing Tests.

The assertion subscribes synchronously on the calling thread with the ambient SynchronizationContext suppressed (SubscribeHereOffSyncContext), so the observer is attached in time to see a hot subject's emission while the mesh's continuations still capture null and land on the pool. It then waits through the package's own ReactiveWait.First — a TaskCompletionSource created with TaskCreationOptions.RunContinuationsAsynchronously — not .ToTask(), so the test's resume is queued instead of running on whichever mesh thread signalled. Blocking that thread yourself (.Result / .Wait()) funnels those continuations back onto xUnit's single-threaded MaxConcurrencySyncContext — measured at 8 s versus 3 ms on AddressResolutionTest (2026-06-15). Do not reintroduce SubscribeOn, and do not drop the suppression.

1. The Observable Assertion Surface

obs.Should() returns an ObservableAssertions<T>. Each terminal method subscribes, waits up to the configured timeout (default 10 s) for the emission you describe, asserts the result, and returns a Task carrying the matched value so you can chain further assertions on the awaited result.

[Fact]
public async Task ObserveQuery_EmitsInitialResults()            // ← async Task; every terminal call awaited
{
    var p = P();
    // Cold observables: the assertion's Subscribe IS the write.
    await NodeFactory.CreateNode(MeshNode.FromPath($"{p}/Project1") with { NodeType = "Markdown" }).Should().Emit();
    await NodeFactory.CreateNode(MeshNode.FromPath($"{p}/Project2") with { NodeType = "Markdown" }).Should().Emit();

    // Fold the assertion into the predicate: wait for the snapshot that has both items.
    var changes = await ObserveAccumulated($"path:{p} nodeType:Markdown scope:descendants")
        .Should(WaitTimeout).Match(acc => acc.Count >= 1 && acc[0].Items.Count >= 2);
    changes[0].ChangeType.Should().Be(QueryChangeType.Initial);
}

(That is ObservableQueryTests.ObserveQuery_EmitsInitialResults in test/MeshWeaver.Query.Test — the role model for this shape.)

.Emit() and .Match() return the matched value, so var x = await obs.Should().Match(...) replaces a hand-rolled var x = await obs.FirstAsync().ToTask() one-for-one — and that hand-rolled shape is now a defect wherever it appears, not merely a verbose alternative.

🚨 There is no sanctioned Rx→Task bridge — not even at the test edge (2026-08-30). Nothing in the assertion path calls .ToTask().

The assertion owns the wait. The source is subscribed, filtered, Take(1)'d, collected with ToList() (so an empty completion arrives as data rather than as an exception indistinguishable from a source fault), bounded with Timeout (throwing a private sentinel so the assertion's own timeout stays distinguishable from a TimeoutException raised by the source), and settled through ReactiveWait — a TaskCompletionSource created with RunContinuationsAsynchronously, which is the line that keeps the test's continuation off the signalling thread. The subscription is disposed when the wait settles, so a settled assertion stops consuming a stream its siblings still need. Nothing blocks a thread. A timed-out assertion reports what the stream actually emitted — "emitted nothing at all" versus "last of N emissions was …" — because those two failures have opposite fixes.

2. The Rule That Makes It Safe: Await the Assertion, Never Block

The assertion never blocks a thread — it SubscribeOns the source onto TaskPoolScheduler and hands you a Task. That SubscribeOn is load-bearing, not cosmetic: xUnit runs async Task tests under a single-threaded MaxConcurrencySyncContext (maxParallelThreads: 1 in test/xunit.runner.json). Subscribe a cold mesh observable directly on that thread and every mesh continuation is funnelled back onto the one sync-context thread and serialises.

A test method that uses a reactive assertion is async Task, and every terminal call is awaited.

In practice that means the following substitutions:

• Stream waits → await obs.Should().Match(...)
• Cold observable-returning calls (IMeshService.CreateNode/UpdateNode/DeleteNode, hub.Observe(...), ReadNode) → await ….Should().Emit() — the subscribe is the work
• Request/response polling loops → await Observable.Interval(...).SelectMany(...).Should().Match(...)
• CancellationToken declarations → deleted (the assertion's .Within(t) is the deadline)

hub.Observe(...) and the IMeshService write methods return IObservable<T>, not Task<T>. Awaiting the observable directly looks like a normal Task await but isn't — Rx's awaiter yields the last value, so it silently waits for completion rather than the first matching emission. Go through .Should() instead.

A hand-rolled .Result / .Wait() / .GetAwaiter().GetResult() in a test body is the failure mode from the other direction — it holds the sync-context thread and bypasses the pooled subscribe. Always go through .Should().

2.1 A settled assertion has already unsubscribed

When the await returns, the assertion is no longer a subscriber. You may rely on that with no grace period, no poll and no retry — which matters whenever the very next thing the test measures is "is anything still watching this?":

await Cache.GetStream(path, options).Materialize().Should().Match(n => n.Kind == NotificationKind.OnError);

// Legitimate: the only subscriber was the assertion above, and it is gone.
Cache.ReleaseIfUnwatched(path).Should().BeTrue();

That question is common — a refcounted cache entry that may only be reclaimed unwatched, a released claim, an idle sweep that skips a live path — and until 2026-09-05 the guarantee did not hold. ReactiveWait.First disposed as a continuation on the settled task, which structurally cannot get there first: the task carries RunContinuationsAsynchronously, so TrySetResult queues the awaiting test immediately, while Take(1) disposes its upstream from inside ForwardOnCompleted — after it has pushed the value into the handler. The unsubscribe was always last, by a few instructions, and a loaded runner is enough to lose that race. ChangeFeedResetReleasesUpstreamTest.ControlArm_ReleaseIfUnwatched_… did lose it: the entry it asked the cache to release was pinned by the assertion that had just returned. The wait now disposes inside the handler, ahead of the settle; AssertionUnsubscribesBeforeItSettlesTest pins it.

The converse still holds and is deliberate: a fault arriving after a wait has settled can no longer be carried by the task, so it is traced rather than dropped — never rethrown into an unrelated test, and never left to surface on the finalizer as an UnobservedTaskException (xUnit v3 escalates those to a Catastrophic failure that poisons the next class).

The mirror-image mistake is dropping the await. obs.Should().Emit(); as a bare statement is a discarded Task<T>: it subscribes and returns immediately, so the test asserts nothing and proceeds on a race. In an expression position the compiler usually catches it (Task<MeshNode?> will not bind to MeshNode?), but as a statement it is silent.

3. Genuinely-Async Constructs Alongside the Assertions

Test bodies are async Task throughout, so nothing has to be "converted away" from async any more. What still deserves a deliberate decision is which async construct to reach for:

• Stream mocks. A fake IChatClient (GetStreamingResponseAsync / GetResponseAsync with await Task.Delay / Task.Yield between chunks) or a fake IAsyncEnumerable (await Task.CompletedTask; yield break;). These implement async interfaces — leave them async; they are infrastructure, not the test body.
• Async system-under-test. The test drives a genuinely async API: an await foreach over the SUT's IAsyncEnumerable, a parser's ParseAsync, an ASP.NET middleware's InvokeAsync(context), a controller's ExchangeToken(...). The async is what is being verified.
• Genuine file / network / process I/O. File.ReadAllTextAsync, StreamReader.ReadToEndAsync, Process.WaitForExitAsync, an HTTP handler.
• Concurrency as the SUT. Task.WhenAll / WaitAsync where the in-flight concurrency or a deadlock reproduction is the thing under test.

What is never acceptable in a test body: .Result, .Wait(), .GetAwaiter().GetResult(), and Task.Delay used as a propagation wait (see §5 for the only sanctioned fixed wait).

Bridging a genuine Task<T> / ValueTask<T> SDK boundary into the reactive surface:

await call(...).AsTask().ToObservable().Should().Within(t).Emit()
// requires: using System.Reactive.Threading.Tasks;

Use this only for a real async boundary such as AIFunction.InvokeAsync — not for MeshWeaver's own observable-returning methods, which are already observables.

4. Asserting an Expected Error — .Materialize()

.Should().Emit() and .Match() wrap an OnError inside an ObservableAssertionException — they do not rethrow the original exception type. To assert that a stream errors with a specific type, fold OnError into a value using .Materialize():

var error = await source.Take(1).Materialize()
    .Should().Within(5.Seconds()).Match(n => n.Kind == NotificationKind.OnError);
error.Exception.Should().BeOfType<DeliveryFailureException>();

using System.Reactive; provides NotificationKind. This is the reactive replacement for await act.Should().ThrowAsync<T>().

For a synchronous throwing call, the FluentAssertions-shaped ((Action)(() => …)).Should().Throw<T>().WithMessage("*…*") still applies.

5. Fold the Assertion Into the Predicate

.Match(items => items.Count == 2) waits for the right state, eliminating the classic "wait, then assert" race that passes locally but flakes under CI load.

Never take the first emission unconditionally. A synced or multi-query stream's first emission may carry only one upstream's partial result. If items trickle in (Added events after a short Initial), accumulate first, then assert:

await stream.Scan(ImmutableList<T>.Empty, (acc, change) => acc.AddRange(change.Items))
    .Should().Match(acc => acc.Count == N);

.NotEmit(within) is the only place a fixed wait is intentional — a "nothing should happen" test has no positive signal to wait for. Keep the window short.

6. BeEquivalentTo — Pass the Hub's JsonSerializerOptions

Equivalence serializes both sides with System.Text.Json, so the polymorphic $type discriminators must line up. The options must come from the owning hub — this deliberately routes equivalence through the real serializer:

actual.Should().BeEquivalentTo(expected, hub.JsonSerializerOptions);
actual.Should().BeEquivalentTo(expected, hub.JsonSerializerOptions, o => o.Excluding(x => x.Message));

Use whichever hub the test has in scope: hub.JsonSerializerOptions, Mesh.JsonSerializerOptions, GetClient().JsonSerializerOptions.

For a plain DTO with no hub in scope (a parser-result record, no polymorphism), use JsonSerializerOptions.Default — do not new() a fresh instance.

The builder is FluentAssertions-shaped (Excluding, Including, WithStrictOrdering() — collections compare order-insensitively without it) plus JSON-flavoured extensions: ExcludeTypeDiscriminator(), IncludingTypeDiscriminator(), ExcludeProperty<TDecl,TProp>(...), UsingJson(...). There is no NotBeEquivalentTo in this library — assert the negative some other way (e.g. compare a specific member, or .Should().NotBe(...)).

7. JSON-Element Assertions

var root = serialized.Should().NotBeNull().And.BeValidJson().Which;   // string -> JsonNode
root.Should().HaveElement("message").Which
    .Should().HaveElement("$type").Which
    .Should().HaveValue(typeof(SubscribeRequest).FullName);

The provided helpers are: BeValidJson() (on string), HaveElement(name) / HaveValue(text) (on JsonNode), and .As<T>().

8. The FluentAssertions-Shaped Value Surface

The familiar names, chaining, and because arguments carry over unchanged:

• Object: Be / NotBe / BeNull / NotBeNull / BeSameAs / NotBeSameAs / BeOfType<T>() / BeAssignableTo<T>
• Boolean: BeTrue / BeFalse
• String: Contain / NotContain / StartWith / EndWith / Match / MatchRegex / BeEmpty / NotBeNullOrEmpty / NotBeNullOrWhiteSpace / HaveLength
• Comparable: BeGreaterThan(OrEqualTo) / BeLessThan(OrEqualTo) / BeInRange / BePositive / BeAfter / BeBefore
• Collection: HaveCount(GreaterThan…) / BeEmpty / NotBeEmpty / Contain / ContainSingle / OnlyContain / AllSatisfy / Equal / BeSubsetOf / OnlyHaveUniqueItems / BeInAscendingOrder
• Dictionary: ContainKey / ContainValue
• Enum: HaveFlag / NotHaveFlag
• Action / async: Throw<T>().WithMessage(...) / NotThrow / ThrowAsync<T>() / NotThrowAsync
• Time helpers: 10.Seconds(), 200.Milliseconds(), 1.5.Minutes()

AssertionScope collects failures and throws on dispose. All failures throw AssertionException (stream expectations throw the derived ObservableAssertionException).

9. Test as If You Were Inside an Activity

Production work runs on an activity hub — its own sandbox with its own AccessContext, single-threaded action block, and Status / RequestedStatus lifecycle (see Activity Control Plane). A test that calls an internal method directly on the test thread skips that context and can pass while production fails — the recurring AccessContext-propagation bug follows exactly this pattern.

Drive the work the way production does: set the control property, observe the result reactively.

await workspace.GetMeshNodeStream(activityPath)
    .Update(node => node with { Content = ((ActivityLog)node.Content) with { RequestedStatus = ActivityStatus.Running } })
    .Should().Emit();

await workspace.GetMeshNodeStream(activityPath)
    .Select(n => (ActivityLog)n.Content)
    .Should().Match(a => a.Status == ActivityStatus.Succeeded);

This exercises the real control plane: the owning hub's watcher reacts to RequestedStatus, runs the work under the activity's identity, and writes Status back — exactly the path production takes.

10. Extending the Library

The library lives in src/MeshWeaver.Reactive.Assertions (System.Reactive only) — the observable surface in ObservableAssertions.cs, the value surface in ObjectAssertions.cs / CollectionAssertions.cs / MoreAssertions.cs, and the equivalency + JSON helpers in Equivalency.cs / JsonAssertions.cs.

If a genuinely missing assertion is blocking a test, add it to the library with a unit test in test/MeshWeaver.Reactive.Assertions.Test that exercises both the pass and the fail path — do not work around it with a hand-rolled .FirstAsync().ToTask() in the test body. That workaround is not merely discouraged; it is the forbidden shape, and the library's ReactiveWait.First exists precisely so no call site ever needs it.

See Also

• Writing Tests — surrounding rules: CQRS-correct reads, the init-ping nuance, Orleans clients
• Asynchronous Calls — why nothing in hub-reachable code is async
• Activity Control Plane — operations as content patches on an activity node
• src/MeshWeaver.Reactive.Assertions/ObservableAssertions.cs — the Emit / Match / Be / Complete / NotEmit / Within implementation