From 2db9668869ac3063d0f826b26846706758cdab14 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 16:36:03 +0000 Subject: [PATCH 1/3] Add EF Core integration tests for scope per dispatch New project tests/Switchboard.IntegrationTests, added to Switchboard.slnx so CI's plain dotnet restore/build/test runs it. It is kept apart from Switchboard.Tests so EF Core never lifts that project's pinned Microsoft.Extensions.DependencyInjection floor. EF Core Sqlite 8.0.32 / 9.0.21 / 10.0.13 per framework, against a shared-cache in-memory database. A Blazor circuit is simulated as one long-lived scope with a scoped AddDbContext and the ISender resolved from it. A DbCommandInterceptor gate holds one command just before it executes, so a dispatch is provably inside EF's critical section when the next one starts, with no timing involved. Covered: - Without scope per dispatch, overlapping sends share the circuit's DbContext and EF throws "A second operation was started on this context instance". With it, each gets its own context and both rows are saved. - Nested sends and published events share the outer DbContext and transaction; rolling back the outer transaction discards the nested write. - Streaming: a write sent inside await foreach gets its own context and the stream's scope ends with the enumeration. A save still in flight when the shared-context stream reads its next row makes EF throw; with scope per dispatch it works, and abandoning the stream disposes its scope. - The UseScopePerDispatch callback (async and sync) carries the circuit's user into the dispatch, including across overlapping dispatches from two circuits; without it the dispatch scope has no user. Finding: a write awaited inside await foreach over a shared-context AsAsyncEnumerable() does not fail on SQLite. EF holds its concurrency check only during each MoveNextAsync, and Microsoft.Data.Sqlite allows a command while a reader is open. A test documents that behavior as observed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qh4oHys2D2TLycvt3yKVtX --- Switchboard.slnx | 1 + .../AppDbContext.cs | 48 +++++ .../CommandGate.cs | 82 ++++++++ .../ConcurrentDispatchTests.cs | 84 ++++++++ .../CurrentUserTests.cs | 92 ++++++++ .../Switchboard.IntegrationTests/Handlers.cs | 197 ++++++++++++++++++ .../NestedUnitOfWorkTests.cs | 60 ++++++ .../StreamingTests.cs | 156 ++++++++++++++ .../Switchboard.IntegrationTests.csproj | 38 ++++ .../Switchboard.IntegrationTests/TestHost.cs | 159 ++++++++++++++ 10 files changed, 917 insertions(+) create mode 100644 tests/Switchboard.IntegrationTests/AppDbContext.cs create mode 100644 tests/Switchboard.IntegrationTests/CommandGate.cs create mode 100644 tests/Switchboard.IntegrationTests/ConcurrentDispatchTests.cs create mode 100644 tests/Switchboard.IntegrationTests/CurrentUserTests.cs create mode 100644 tests/Switchboard.IntegrationTests/Handlers.cs create mode 100644 tests/Switchboard.IntegrationTests/NestedUnitOfWorkTests.cs create mode 100644 tests/Switchboard.IntegrationTests/StreamingTests.cs create mode 100644 tests/Switchboard.IntegrationTests/Switchboard.IntegrationTests.csproj create mode 100644 tests/Switchboard.IntegrationTests/TestHost.cs diff --git a/Switchboard.slnx b/Switchboard.slnx index 63b89cb..f64dd76 100644 --- a/Switchboard.slnx +++ b/Switchboard.slnx @@ -3,6 +3,7 @@ + diff --git a/tests/Switchboard.IntegrationTests/AppDbContext.cs b/tests/Switchboard.IntegrationTests/AppDbContext.cs new file mode 100644 index 0000000..20d07f2 --- /dev/null +++ b/tests/Switchboard.IntegrationTests/AppDbContext.cs @@ -0,0 +1,48 @@ +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; + +namespace Switchboard.IntegrationTests; + +public sealed class Note +{ + public int Id { get; set; } + + public string Text { get; set; } = ""; + + public string? CreatedBy { get; set; } +} + +public sealed class AuditEntry +{ + public int Id { get; set; } + + public string Action { get; set; } = ""; +} + +/// The application's DbContext, registered scoped with AddDbContext as a Blazor Server app would. +public sealed class AppDbContext : DbContext +{ + public AppDbContext(DbContextOptions options) + : base(options) + { + } + + public DbSet Notes => Set(); + + public DbSet AuditEntries => Set(); + + /// Set once the owning DI scope has disposed this instance. + public bool IsDisposed { get; private set; } + + public override void Dispose() + { + base.Dispose(); + IsDisposed = true; + } + + public override async ValueTask DisposeAsync() + { + await base.DisposeAsync(); + IsDisposed = true; + } +} diff --git a/tests/Switchboard.IntegrationTests/CommandGate.cs b/tests/Switchboard.IntegrationTests/CommandGate.cs new file mode 100644 index 0000000..9216e3a --- /dev/null +++ b/tests/Switchboard.IntegrationTests/CommandGate.cs @@ -0,0 +1,82 @@ +using System; +using System.Data.Common; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore.Diagnostics; + +namespace Switchboard.IntegrationTests; + +/// +/// Holds one database command just before it executes, so a dispatch is provably mid-operation — inside EF Core's +/// critical section for that DbContext — while the test starts another one. Microsoft.Data.Sqlite completes its +/// async calls synchronously, so without this two "concurrent" dispatches would never actually overlap. +/// +/// +/// it, start the first dispatch, await , start the second, then . +/// Only the first matching command after arming is held; every other command passes straight through. +/// +public sealed class CommandGate +{ + // Guards against a hang if a test is wrong; never part of what a test asserts. + private static readonly TimeSpan HangGuard = TimeSpan.FromSeconds(30); + + private readonly TaskCompletionSource _held = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _released = new(TaskCreationOptions.RunContinuationsAsynchronously); + private string? _commandPrefix; + private int _armed; + + /// Completes, with the command's text, once a command is being held. + public Task Held => _held.Task.WaitAsync(HangGuard); + + /// Holds the next command whose SQL starts with (e.g. INSERT, SELECT). + public void Arm(string commandPrefix) + { + _commandPrefix = commandPrefix; + Volatile.Write(ref _armed, 1); + } + + /// Lets the held command execute. + public void Release() => _released.TrySetResult(); + + internal async ValueTask WaitIfArmedAsync(DbCommand command, CancellationToken cancellationToken) + { + if (Volatile.Read(ref _armed) == 0 || + !command.CommandText.TrimStart().StartsWith(_commandPrefix!, StringComparison.OrdinalIgnoreCase) || + Interlocked.CompareExchange(ref _armed, 0, 1) != 1) + { + return; + } + + _held.TrySetResult(command.CommandText); + await _released.Task.WaitAsync(HangGuard, cancellationToken); + } +} + +/// Routes every async command through the . +public sealed class GatingInterceptor : DbCommandInterceptor +{ + private readonly CommandGate _gate; + + public GatingInterceptor(CommandGate gate) => _gate = gate; + + public override async ValueTask> ReaderExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + await _gate.WaitIfArmedAsync(command, cancellationToken); + return result; + } + + public override async ValueTask> NonQueryExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + await _gate.WaitIfArmedAsync(command, cancellationToken); + return result; + } + + public override async ValueTask> ScalarExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + await _gate.WaitIfArmedAsync(command, cancellationToken); + return result; + } +} diff --git a/tests/Switchboard.IntegrationTests/ConcurrentDispatchTests.cs b/tests/Switchboard.IntegrationTests/ConcurrentDispatchTests.cs new file mode 100644 index 0000000..956e0a0 --- /dev/null +++ b/tests/Switchboard.IntegrationTests/ConcurrentDispatchTests.cs @@ -0,0 +1,84 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Xunit; + +namespace Switchboard.IntegrationTests; + +/// +/// The README's headline case: two dispatches from one Blazor Server circuit overlap — a click while a save is still +/// in flight. The circuit's scope is the whole connection, so without scope-per-dispatch both share its DbContext. +/// +public sealed class ConcurrentDispatchTests +{ + /// The first sentence of EF Core's ConcurrentMethodInvocation error. + internal const string SecondOperation = + "A second operation was started on this context instance before a previous operation completed."; + + [Fact] + public async Task Without_scope_per_dispatch_overlapping_sends_share_the_circuits_DbContext_and_EF_throws() + { + await using var host = await TestHost.StartAsync(); + await using var circuit = host.OpenCircuit(); + + host.Gate.Arm("INSERT"); + var first = circuit.Sender.Send(new CreateNote("first", "saved first")); + var heldCommand = await host.Gate.Held; + + // The first dispatch is now inside SaveChangesAsync, its INSERT held just before it executes. + Assert.False(first.IsCompleted); + Assert.StartsWith("INSERT INTO \"Notes\"", heldCommand); + + var second = circuit.Sender.Send(new CreateNote("second", "saved second")); + + var error = await Assert.ThrowsAsync(() => second.Guarded()); + Assert.StartsWith(SecondOperation, error.Message); + + // Both handlers were handed the circuit's one DbContext. + Assert.Same(circuit.DbContext, host.Contexts["first"]); + Assert.Same(circuit.DbContext, host.Contexts["second"]); + + host.Gate.Release(); + await first.Guarded(); + + var saved = await host.WithFreshContextAsync(db => db.Notes.Select(n => n.Text).ToListAsync()); + Assert.Equal(new[] { "saved first" }, saved); + } + + [Fact] + public async Task With_scope_per_dispatch_overlapping_sends_each_get_their_own_DbContext_and_both_save() + { + await using var host = await TestHost.StartAsync(cfg => cfg.UseScopePerDispatch()); + await using var circuit = host.OpenCircuit(); + + host.Gate.Arm("INSERT"); + var first = circuit.Sender.Send(new CreateNote("first", "saved first")); + await host.Gate.Held; + + // The same overlap as above: the first dispatch is mid-SaveChangesAsync while the second starts and finishes. + Assert.False(first.IsCompleted); + + var secondId = await circuit.Sender.Send(new CreateNote("second", "saved second")).Guarded(); + Assert.False(first.IsCompleted); + + host.Gate.Release(); + var firstId = await first.Guarded(); + + var firstDb = host.Contexts["first"]; + var secondDb = host.Contexts["second"]; + Assert.NotSame(firstDb, secondDb); + Assert.NotSame(circuit.DbContext, firstDb); + Assert.NotSame(circuit.DbContext, secondDb); + + // Each dispatch scope, and its DbContext, ended with its dispatch. + Assert.True(firstDb.IsDisposed); + Assert.True(secondDb.IsDisposed); + Assert.False(circuit.DbContext.IsDisposed); + + var saved = await host.WithFreshContextAsync(db => db.Notes.OrderBy(n => n.Id).Select(n => new { n.Id, n.Text }).ToListAsync()); + Assert.Equal( + new[] { new { Id = secondId, Text = "saved second" }, new { Id = firstId, Text = "saved first" } }, + saved); + } +} diff --git a/tests/Switchboard.IntegrationTests/CurrentUserTests.cs b/tests/Switchboard.IntegrationTests/CurrentUserTests.cs new file mode 100644 index 0000000..b79d18a --- /dev/null +++ b/tests/Switchboard.IntegrationTests/CurrentUserTests.cs @@ -0,0 +1,92 @@ +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace Switchboard.IntegrationTests; + +/// +/// A dispatch scope is new, so scoped state from the circuit — who is signed in — has to be carried into it by the +/// UseScopePerDispatch callback. Mirrors the README's AuthenticationStateProvider example. +/// +public sealed class CurrentUserTests +{ + private static SwitchboardConfiguration CarryCurrentUser(SwitchboardConfiguration cfg) + => cfg.UseScopePerDispatch(async (scope, cancellationToken) => + { + var auth = await scope.Parent.GetRequiredService().GetAuthenticationStateAsync(); + scope.ServiceProvider.GetRequiredService().Set(auth); + }); + + private static Task CreatedBy(TestHost host, int noteId) + => host.WithFreshContextAsync(db => db.Notes.Where(n => n.Id == noteId).Select(n => n.CreatedBy).SingleAsync()); + + [Fact] + public async Task The_callback_carries_the_circuits_user_into_the_dispatch_and_the_handler_stamps_it() + { + await using var host = await TestHost.StartAsync(cfg => CarryCurrentUser(cfg)); + await using var circuit = host.OpenCircuit("alice"); + + var noteId = await circuit.Sender.Send(new CreateNote("alice's", "stamped")); + + Assert.Equal("alice", await CreatedBy(host, noteId)); + } + + [Fact] + public async Task The_synchronous_callback_carries_the_user_too() + { + await using var host = await TestHost.StartAsync(cfg => cfg.UseScopePerDispatch(scope => + scope.ServiceProvider.GetRequiredService().Set( + scope.Parent.GetRequiredService().User))); + await using var circuit = host.OpenCircuit("alice"); + + var noteId = await circuit.Sender.Send(new CreateNote("alice's", "stamped")); + + Assert.Equal("alice", await CreatedBy(host, noteId)); + } + + [Fact] + public async Task Overlapping_dispatches_from_two_circuits_each_stamp_their_own_user() + { + await using var host = await TestHost.StartAsync(cfg => CarryCurrentUser(cfg)); + await using var alice = host.OpenCircuit("alice"); + await using var bob = host.OpenCircuit("bob"); + + host.Gate.Arm("INSERT"); + var alicesSave = alice.Sender.Send(new CreateNote("alice's", "by alice")); + await host.Gate.Held; + + var bobsNote = await bob.Sender.Send(new CreateNote("bob's", "by bob")).Guarded(); + + host.Gate.Release(); + var alicesNote = await alicesSave.Guarded(); + + Assert.Equal("alice", await CreatedBy(host, alicesNote)); + Assert.Equal("bob", await CreatedBy(host, bobsNote)); + } + + [Fact] + public async Task Without_the_callback_the_dispatch_scope_has_no_user() + { + await using var host = await TestHost.StartAsync(cfg => cfg.UseScopePerDispatch()); + await using var circuit = host.OpenCircuit("alice"); + + var noteId = await circuit.Sender.Send(new CreateNote("alice's", "not stamped")); + + // The circuit's CurrentUser knows alice, but the handler got the dispatch scope's fresh, empty one. + Assert.Equal("alice", circuit.CurrentUser.Name); + Assert.Null(await CreatedBy(host, noteId)); + } + + [Fact] + public async Task Without_scope_per_dispatch_the_handler_sees_the_circuits_user_directly() + { + await using var host = await TestHost.StartAsync(); + await using var circuit = host.OpenCircuit("alice"); + + var noteId = await circuit.Sender.Send(new CreateNote("alice's", "stamped")); + + Assert.Equal("alice", await CreatedBy(host, noteId)); + } +} diff --git a/tests/Switchboard.IntegrationTests/Handlers.cs b/tests/Switchboard.IntegrationTests/Handlers.cs new file mode 100644 index 0000000..9c31906 --- /dev/null +++ b/tests/Switchboard.IntegrationTests/Handlers.cs @@ -0,0 +1,197 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; + +namespace Switchboard.IntegrationTests; + +/// Saves a note stamped with the current user. names the dispatch in the . +public sealed record CreateNote(string Label, string Text) : IRequest; + +public sealed class CreateNoteHandler : IRequestHandler +{ + private readonly AppDbContext _db; + private readonly CurrentUser _currentUser; + private readonly ContextLog _contexts; + + public CreateNoteHandler(AppDbContext db, CurrentUser currentUser, ContextLog contexts) + { + _db = db; + _currentUser = currentUser; + _contexts = contexts; + } + + public async Task Handle(CreateNote request, CancellationToken cancellationToken) + { + _contexts.Record(request.Label, _db); + + var note = new Note { Text = request.Text, CreatedBy = _currentUser.Name }; + _db.Notes.Add(note); + await _db.SaveChangesAsync(cancellationToken); + + return note.Id; + } +} + +/// Writes an audit entry and answers with the id of the transaction it was written in, if any. +public sealed record WriteAudit(string Label, string Action) : IRequest; + +public sealed class WriteAuditHandler : IRequestHandler +{ + private readonly AppDbContext _db; + private readonly ContextLog _contexts; + + public WriteAuditHandler(AppDbContext db, ContextLog contexts) + { + _db = db; + _contexts = contexts; + } + + public async Task Handle(WriteAudit request, CancellationToken cancellationToken) + { + _contexts.Record(request.Label, _db); + + _db.AuditEntries.Add(new AuditEntry { Action = request.Action }); + await _db.SaveChangesAsync(cancellationToken); + + return _db.Database.CurrentTransaction?.TransactionId; + } +} + +/// Saves a note, then sends a nested through its injected sender. +public sealed record CreateNoteWithAudit(string Text) : IRequest; + +public sealed class CreateNoteWithAuditHandler : IRequestHandler +{ + private readonly AppDbContext _db; + private readonly ISender _sender; + private readonly ContextLog _contexts; + + public CreateNoteWithAuditHandler(AppDbContext db, ISender sender, ContextLog contexts) + { + _db = db; + _sender = sender; + _contexts = contexts; + } + + public async Task Handle(CreateNoteWithAudit request, CancellationToken cancellationToken) + { + _contexts.Record("outer", _db); + + var note = new Note { Text = request.Text }; + _db.Notes.Add(note); + await _db.SaveChangesAsync(cancellationToken); + + await _sender.Send(new WriteAudit("inner", $"created note {note.Id}"), cancellationToken); + + return note.Id; + } +} + +/// +/// Opens a transaction, saves a note, sends a nested , then rolls everything back. +/// Answers with the transaction it opened and the one the nested handler saw. +/// +public sealed record CreateNoteThenRollBack(string Text) : IRequest<(Guid Outer, Guid? Inner)>; + +public sealed class CreateNoteThenRollBackHandler : IRequestHandler +{ + private readonly AppDbContext _db; + private readonly ISender _sender; + + public CreateNoteThenRollBackHandler(AppDbContext db, ISender sender) + { + _db = db; + _sender = sender; + } + + public async Task<(Guid Outer, Guid? Inner)> Handle(CreateNoteThenRollBack request, CancellationToken cancellationToken) + { + await using var transaction = await _db.Database.BeginTransactionAsync(cancellationToken); + + _db.Notes.Add(new Note { Text = request.Text }); + await _db.SaveChangesAsync(cancellationToken); + + var inner = await _sender.Send(new WriteAudit("inner", "written inside the outer transaction"), cancellationToken); + + await transaction.RollbackAsync(cancellationToken); + + return (transaction.TransactionId, inner); + } +} + +/// A domain event, published by after it saves. +public sealed record NoteCreated(int NoteId) : INotification; + +public sealed class NoteCreatedAuditHandler : INotificationHandler +{ + private readonly AppDbContext _db; + private readonly ContextLog _contexts; + + public NoteCreatedAuditHandler(AppDbContext db, ContextLog contexts) + { + _db = db; + _contexts = contexts; + } + + public async Task Handle(NoteCreated notification, CancellationToken cancellationToken) + { + _contexts.Record("event", _db); + + _db.AuditEntries.Add(new AuditEntry { Action = $"note {notification.NoteId} created" }); + await _db.SaveChangesAsync(cancellationToken); + } +} + +public sealed record CreateNoteAndPublish(string Text) : IRequest; + +public sealed class CreateNoteAndPublishHandler : IRequestHandler +{ + private readonly AppDbContext _db; + private readonly IPublisher _publisher; + private readonly ContextLog _contexts; + + public CreateNoteAndPublishHandler(AppDbContext db, IPublisher publisher, ContextLog contexts) + { + _db = db; + _publisher = publisher; + _contexts = contexts; + } + + public async Task Handle(CreateNoteAndPublish request, CancellationToken cancellationToken) + { + _contexts.Record("outer", _db); + + var note = new Note { Text = request.Text }; + _db.Notes.Add(note); + await _db.SaveChangesAsync(cancellationToken); + + await _publisher.Publish(new NoteCreated(note.Id), cancellationToken); + + return note.Id; + } +} + +/// Streams every note straight off an EF Core query, as the README's GetOrderLines example does. +public sealed record StreamNotes(string Label) : IStreamRequest; + +public sealed class StreamNotesHandler : IStreamRequestHandler +{ + private readonly AppDbContext _db; + private readonly ContextLog _contexts; + + public StreamNotesHandler(AppDbContext db, ContextLog contexts) + { + _db = db; + _contexts = contexts; + } + + public IAsyncEnumerable Handle(StreamNotes request, CancellationToken cancellationToken) + { + _contexts.Record(request.Label, _db); + + return _db.Notes.OrderBy(n => n.Id).AsAsyncEnumerable(); + } +} diff --git a/tests/Switchboard.IntegrationTests/NestedUnitOfWorkTests.cs b/tests/Switchboard.IntegrationTests/NestedUnitOfWorkTests.cs new file mode 100644 index 0000000..21738b1 --- /dev/null +++ b/tests/Switchboard.IntegrationTests/NestedUnitOfWorkTests.cs @@ -0,0 +1,60 @@ +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Xunit; + +namespace Switchboard.IntegrationTests; + +/// +/// With scope-per-dispatch on, a handler that sends or publishes again through its injected sender or publisher +/// stays in the dispatch in flight: one DbContext, one unit of work, one transaction. +/// +public sealed class NestedUnitOfWorkTests +{ + [Fact] + public async Task A_nested_send_shares_the_outer_handlers_DbContext() + { + await using var host = await TestHost.StartAsync(cfg => cfg.UseScopePerDispatch()); + await using var circuit = host.OpenCircuit(); + + var noteId = await circuit.Sender.Send(new CreateNoteWithAudit("with audit")); + + Assert.Same(host.Contexts["outer"], host.Contexts["inner"]); + Assert.NotSame(circuit.DbContext, host.Contexts["outer"]); + Assert.True(host.Contexts["outer"].IsDisposed); + + Assert.Equal(new[] { "with audit" }, await host.WithFreshContextAsync(db => db.Notes.Select(n => n.Text).ToListAsync())); + Assert.Equal( + new[] { $"created note {noteId}" }, + await host.WithFreshContextAsync(db => db.AuditEntries.Select(a => a.Action).ToListAsync())); + } + + [Fact] + public async Task A_nested_send_writes_in_the_outer_handlers_transaction_and_rolls_back_with_it() + { + await using var host = await TestHost.StartAsync(cfg => cfg.UseScopePerDispatch()); + await using var circuit = host.OpenCircuit(); + + var (outer, inner) = await circuit.Sender.Send(new CreateNoteThenRollBack("rolled back")); + + // The nested handler's SaveChangesAsync ran inside the transaction the outer handler opened. + Assert.Equal(outer, inner); + + Assert.Empty(await host.WithFreshContextAsync(db => db.Notes.ToListAsync())); + Assert.Empty(await host.WithFreshContextAsync(db => db.AuditEntries.ToListAsync())); + } + + [Fact] + public async Task A_domain_event_published_from_a_handler_shares_its_DbContext() + { + await using var host = await TestHost.StartAsync(cfg => cfg.UseScopePerDispatch()); + await using var circuit = host.OpenCircuit(); + + var noteId = await circuit.Sender.Send(new CreateNoteAndPublish("with event")); + + Assert.Same(host.Contexts["outer"], host.Contexts["event"]); + Assert.Equal( + new[] { $"note {noteId} created" }, + await host.WithFreshContextAsync(db => db.AuditEntries.Select(a => a.Action).ToListAsync())); + } +} diff --git a/tests/Switchboard.IntegrationTests/StreamingTests.cs b/tests/Switchboard.IntegrationTests/StreamingTests.cs new file mode 100644 index 0000000..d0b0c9c --- /dev/null +++ b/tests/Switchboard.IntegrationTests/StreamingTests.cs @@ -0,0 +1,156 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Xunit; + +namespace Switchboard.IntegrationTests; + +/// +/// The classic Blazor streaming case: a component enumerates an EF Core query through CreateStream and, while +/// the stream is open, sends a command that writes. +/// +public sealed class StreamingTests +{ + private static async Task StartWithNotesAsync(Action? configure = null) + { + var host = await TestHost.StartAsync(configure); + + await host.WithFreshContextAsync(db => + { + db.Notes.AddRange(new Note { Text = "a" }, new Note { Text = "b" }, new Note { Text = "c" }); + return db.SaveChangesAsync(); + }); + + return host; + } + + private static Task> AuditActions(TestHost host) + => host.WithFreshContextAsync(db => db.AuditEntries.OrderBy(a => a.Id).Select(a => a.Action).ToListAsync()); + + [Fact] + public async Task With_scope_per_dispatch_a_write_sent_inside_await_foreach_gets_its_own_DbContext_and_the_streams_scope_ends_with_the_enumeration() + { + await using var host = await StartWithNotesAsync(cfg => cfg.UseScopePerDispatch()); + await using var circuit = host.OpenCircuit(); + + var streamed = new List(); + + await foreach (var note in circuit.Sender.CreateStream(new StreamNotes("stream"))) + { + streamed.Add(note.Text); + await circuit.Sender.Send(new WriteAudit($"write {note.Text}", $"seen {note.Text}")); + + // The stream's scope, and the reader on its DbContext, stay open across the writes. + Assert.False(host.Contexts["stream"].IsDisposed); + Assert.NotSame(host.Contexts["stream"], host.Contexts[$"write {note.Text}"]); + Assert.True(host.Contexts[$"write {note.Text}"].IsDisposed); + } + + Assert.Equal(new[] { "a", "b", "c" }, streamed); + Assert.NotSame(circuit.DbContext, host.Contexts["stream"]); + Assert.True(host.Contexts["stream"].IsDisposed); + + Assert.Equal(new[] { "seen a", "seen b", "seen c" }, await AuditActions(host)); + } + + /// + /// Documents what actually happens in the sequential case without scope-per-dispatch. The write shares the stream's + /// DbContext while its data reader is open, but nothing fails on SQLite: EF Core only holds its concurrency check + /// for the duration of each MoveNextAsync, not between rows, and Microsoft.Data.Sqlite lets a connection + /// run a command while a reader on it is open. Whether this case throws depends on the database provider; the + /// overlap that EF Core itself rejects is the one below. + /// + [Fact] + public async Task Without_scope_per_dispatch_a_write_awaited_inside_await_foreach_shares_the_streams_DbContext() + { + await using var host = await StartWithNotesAsync(); + await using var circuit = host.OpenCircuit(); + + var streamed = new List(); + + await foreach (var note in circuit.Sender.CreateStream(new StreamNotes("stream"))) + { + streamed.Add(note.Text); + await circuit.Sender.Send(new WriteAudit($"write {note.Text}", $"seen {note.Text}")); + + Assert.Same(host.Contexts["stream"], host.Contexts[$"write {note.Text}"]); + } + + Assert.Same(circuit.DbContext, host.Contexts["stream"]); + Assert.Equal(new[] { "a", "b", "c" }, streamed); + Assert.Equal(new[] { "seen a", "seen b", "seen c" }, await AuditActions(host)); + } + + /// + /// A component is part-way through a stream when a click sends a save; the save is still in flight when the + /// component asks for the next row. On the circuit's shared DbContext that is two operations at once. + /// + [Fact] + public async Task Without_scope_per_dispatch_reading_the_next_row_while_a_save_is_in_flight_makes_EF_throw() + { + await using var host = await StartWithNotesAsync(); + await using var circuit = host.OpenCircuit(); + + var stream = circuit.Sender.CreateStream(new StreamNotes("stream")).GetAsyncEnumerator(); + try + { + Assert.True(await stream.MoveNextAsync()); + Assert.Equal("a", stream.Current.Text); + + host.Gate.Arm("INSERT"); + var click = circuit.Sender.Send(new WriteAudit("click", "clicked")); + await host.Gate.Held; + + var error = await Assert.ThrowsAsync(() => stream.MoveNextAsync().AsTask().Guarded()); + Assert.StartsWith(ConcurrentDispatchTests.SecondOperation, error.Message); + Assert.Same(host.Contexts["stream"], host.Contexts["click"]); + + host.Gate.Release(); + await click.Guarded(); + } + finally + { + await stream.DisposeAsync(); + } + + Assert.Equal(new[] { "clicked" }, await AuditActions(host)); + } + + [Fact] + public async Task With_scope_per_dispatch_reading_the_next_row_while_a_save_is_in_flight_works_and_abandoning_the_stream_disposes_its_scope() + { + await using var host = await StartWithNotesAsync(cfg => cfg.UseScopePerDispatch()); + await using var circuit = host.OpenCircuit(); + + var stream = circuit.Sender.CreateStream(new StreamNotes("stream")).GetAsyncEnumerator(); + try + { + Assert.True(await stream.MoveNextAsync()); + Assert.Equal("a", stream.Current.Text); + + host.Gate.Arm("INSERT"); + var click = circuit.Sender.Send(new WriteAudit("click", "clicked")); + await host.Gate.Held; + + Assert.True(await stream.MoveNextAsync().AsTask().Guarded()); + Assert.Equal("b", stream.Current.Text); + Assert.False(click.IsCompleted); + + host.Gate.Release(); + await click.Guarded(); + + Assert.NotSame(host.Contexts["stream"], host.Contexts["click"]); + Assert.False(host.Contexts["stream"].IsDisposed); + } + finally + { + // Stop after two of the three rows, as breaking out of await foreach would. + await stream.DisposeAsync(); + } + + Assert.True(host.Contexts["stream"].IsDisposed); + Assert.Equal(new[] { "clicked" }, await AuditActions(host)); + } +} diff --git a/tests/Switchboard.IntegrationTests/Switchboard.IntegrationTests.csproj b/tests/Switchboard.IntegrationTests/Switchboard.IntegrationTests.csproj new file mode 100644 index 0000000..e92dbc7 --- /dev/null +++ b/tests/Switchboard.IntegrationTests/Switchboard.IntegrationTests.csproj @@ -0,0 +1,38 @@ + + + + + + net8.0;net9.0;net10.0 + latest + false + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/Switchboard.IntegrationTests/TestHost.cs b/tests/Switchboard.IntegrationTests/TestHost.cs new file mode 100644 index 0000000..21accd0 --- /dev/null +++ b/tests/Switchboard.IntegrationTests/TestHost.cs @@ -0,0 +1,159 @@ +using System; +using System.Collections.Concurrent; +using System.Security.Claims; +using System.Threading.Tasks; +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; + +namespace Switchboard.IntegrationTests; + +/// +/// Stands in for Blazor's AuthenticationStateProvider: scoped, so it lives in the circuit's scope and knows +/// who is signed in on that connection. A fresh dispatch scope gets a fresh, anonymous instance. +/// +public sealed class CircuitAuthenticationState +{ + public ClaimsPrincipal User { get; set; } = new(new ClaimsIdentity()); + + public Task GetAuthenticationStateAsync() => Task.FromResult(User); +} + +/// The application's scoped "current user", which handlers read. Empty until something sets it. +public sealed class CurrentUser +{ + public string? Name { get; private set; } + + public void Set(ClaimsPrincipal user) => Name = user.Identity?.Name; +} + +/// Records which DbContext instance each handler ran with, by a label the test chooses. +public sealed class ContextLog +{ + private readonly ConcurrentDictionary _seen = new(); + + public void Record(string label, AppDbContext db) => _seen[label] = db; + + public AppDbContext this[string label] => _seen[label]; +} + +/// +/// One application: a root service provider over a named, shared-cache in-memory SQLite database, kept alive by a +/// connection held open for the host's lifetime. Every DbContext opens its own connection to it, as it would to a +/// real server. +/// +public sealed class TestHost : IAsyncDisposable +{ + private readonly SqliteConnection _keepAlive; + + private TestHost(SqliteConnection keepAlive, ServiceProvider services) + { + _keepAlive = keepAlive; + Services = services; + } + + public ServiceProvider Services { get; } + + public CommandGate Gate => Services.GetRequiredService(); + + public ContextLog Contexts => Services.GetRequiredService(); + + /// Builds the host; adds the scope-per-dispatch setting under test, if any. + public static async Task StartAsync(Action? configure = null) + { + var connectionString = new SqliteConnectionStringBuilder + { + DataSource = $"switchboard-{Guid.NewGuid():N}", + Mode = SqliteOpenMode.Memory, + Cache = SqliteCacheMode.Shared + }.ToString(); + + // An in-memory database lives as long as at least one connection to it is open. + var keepAlive = new SqliteConnection(connectionString); + await keepAlive.OpenAsync(); + + var services = new ServiceCollection() + .AddSingleton() + .AddSingleton() + .AddSingleton() + .AddScoped() + .AddScoped() + .AddDbContext((sp, options) => options + .UseSqlite(connectionString) + .AddInterceptors(sp.GetRequiredService())) + .AddSwitchboard(cfg => + { + cfg.RegisterServicesFromAssemblyContaining(); + configure?.Invoke(cfg); + }) + .BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true, ValidateOnBuild = true }); + + var host = new TestHost(keepAlive, services); + + await host.WithFreshContextAsync(db => db.Database.EnsureCreatedAsync()); + + return host; + } + + /// + /// Opens a Blazor Server circuit: one DI scope that lives as long as the user's connection. Components inject + /// from it, so that is where the circuit's sender is resolved. + /// + public Circuit OpenCircuit(string? userName = null) + { + var scope = Services.CreateAsyncScope(); + + if (userName is not null) + { + // Signed in on this circuit, and known to the circuit scope's own CurrentUser. + var user = new ClaimsPrincipal(new ClaimsIdentity(new[] { new Claim(ClaimTypes.Name, userName) }, authenticationType: "Test")); + scope.ServiceProvider.GetRequiredService().User = user; + scope.ServiceProvider.GetRequiredService().Set(user); + } + + return new Circuit(scope); + } + + /// Runs against a DbContext of its own, outside any circuit: what is really in the database. + public async Task WithFreshContextAsync(Func> work) + { + await using var scope = Services.CreateAsyncScope(); + return await work(scope.ServiceProvider.GetRequiredService()); + } + + public async ValueTask DisposeAsync() + { + // Lets a command still held by a failed test finish, so disposal never waits on it. + Gate.Release(); + await Services.DisposeAsync(); + await _keepAlive.DisposeAsync(); + } +} + +/// A Blazor Server circuit: the long-lived scope, and the a component would inject. +public sealed class Circuit : IAsyncDisposable +{ + private readonly AsyncServiceScope _scope; + + internal Circuit(AsyncServiceScope scope) + { + _scope = scope; + Sender = scope.ServiceProvider.GetRequiredService(); + } + + public ISender Sender { get; } + + /// The circuit scope's own DbContext: the one every dispatch shares when scope-per-dispatch is off. + public AppDbContext DbContext => _scope.ServiceProvider.GetRequiredService(); + + /// The circuit scope's own . + public CurrentUser CurrentUser => _scope.ServiceProvider.GetRequiredService(); + + public ValueTask DisposeAsync() => _scope.DisposeAsync(); +} + +internal static class Timeouts +{ + /// Bounds an await that would otherwise hang a broken test; never what a test asserts. + public static Task Guarded(this Task task) => task.WaitAsync(TimeSpan.FromSeconds(30)); +} From a24641acb2a28dc95b18f658508f0d772d02323a Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 16:44:23 +0000 Subject: [PATCH 2/3] Add Stryker.NET as a local tool for mutation testing `dotnet tool restore && dotnet stryker` mutates src/Switchboard against the test project on net10.0 (Release) and writes progress, cleartext, HTML and JSON reports to StrykerOutput/, which is now git-ignored. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qh4oHys2D2TLycvt3yKVtX --- .config/dotnet-tools.json | 13 +++++++++++++ .gitignore | 3 +++ stryker-config.json | 10 ++++++++++ 3 files changed, 26 insertions(+) create mode 100644 .config/dotnet-tools.json create mode 100644 stryker-config.json diff --git a/.config/dotnet-tools.json b/.config/dotnet-tools.json new file mode 100644 index 0000000..bd86614 --- /dev/null +++ b/.config/dotnet-tools.json @@ -0,0 +1,13 @@ +{ + "version": 1, + "isRoot": true, + "tools": { + "dotnet-stryker": { + "version": "5.0.0", + "commands": [ + "dotnet-stryker" + ], + "rollForward": false + } + } +} \ No newline at end of file diff --git a/.gitignore b/.gitignore index 7b68e30..7a6ce13 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,6 @@ TestResults/ .DS_Store src/notes.txt + +# Stryker.NET mutation-testing output +StrykerOutput/ diff --git a/stryker-config.json b/stryker-config.json new file mode 100644 index 0000000..a7e60bb --- /dev/null +++ b/stryker-config.json @@ -0,0 +1,10 @@ +{ + "stryker-config": { + "project": "Switchboard.csproj", + "test-projects": [ "tests/Switchboard.Tests/Switchboard.Tests.csproj" ], + "target-framework": "net10.0", + "configuration": "Release", + "mutate": [ "**/*.cs" ], + "reporters": [ "progress", "cleartext", "html", "json" ] + } +} From 23e393a524ab3c3ab48d06d19c5733eac25e8352 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 17 Oct 2026 16:47:09 +0000 Subject: [PATCH 3/3] Add tests for gaps found by mutation testing Stryker.NET left 68 mutants alive (47 survived, 21 without coverage); these tests kill the ones that stand for behavior that could regress unnoticed. Validation (ValidationGapTests): - a void request without a handler is reported; - abstract, interface and open generic request types are not reported; - the void-request skip check: five names then "and N more", an exact message for a single skipped request, class/struct/new() constraints on TRequest and TResponse, a constraint on the behavior's own type parameter, and open generic services that are not pipeline behaviors; - ValidateSwitchboard names its null argument. Dispatch (DispatchGapTests): - Send of a void request awaits an async handler and surfaces its failure; - a failing notification handler fails Publish and stops later handlers; - Publish of a non-notification names its parameter; - a stream request that implements another interface still streams; - a request nested in a generic type dispatches and gets a readable name; - UseScopePerDispatch and AddSwitchboard reject null arguments up front. Telemetry (TelemetryGapTests): - histogram units are seconds; - a failing notification handler's span carries its tags, error status and the OpenTelemetry exception event; - cancelled requests and streams are tagged but not marked as errors; - a traced stream still passes the WithCancellation token to the handler (found by hand-mutating ObserveStream, which Stryker's safe mode skips). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qh4oHys2D2TLycvt3yKVtX --- tests/Switchboard.Tests/DispatchGapTests.cs | 203 ++++++++++++++ tests/Switchboard.Tests/TelemetryGapTests.cs | 194 ++++++++++++++ tests/Switchboard.Tests/ValidationGapTests.cs | 248 ++++++++++++++++++ 3 files changed, 645 insertions(+) create mode 100644 tests/Switchboard.Tests/DispatchGapTests.cs create mode 100644 tests/Switchboard.Tests/TelemetryGapTests.cs create mode 100644 tests/Switchboard.Tests/ValidationGapTests.cs diff --git a/tests/Switchboard.Tests/DispatchGapTests.cs b/tests/Switchboard.Tests/DispatchGapTests.cs new file mode 100644 index 0000000..95e4bb2 --- /dev/null +++ b/tests/Switchboard.Tests/DispatchGapTests.cs @@ -0,0 +1,203 @@ +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace Switchboard.Tests; + +/// A void request whose handler only finishes, or fails, once completes. +public sealed record DeferredVoid(Task Gate, bool Fail = false) : IRequest; + +public sealed class DeferredVoidHandler : IRequestHandler +{ + private readonly Recorder _recorder; + + public DeferredVoidHandler(Recorder recorder) => _recorder = recorder; + + public async Task Handle(DeferredVoid request, CancellationToken cancellationToken) + { + await request.Gate; + + if (request.Fail) + { + throw new InvalidOperationException("deferred failure"); + } + + _recorder.Add("deferred:done"); + } +} + +public sealed record FailingNote : INotification; + +public sealed class FailingNoteHandler : INotificationHandler +{ + public async Task Handle(FailingNote notification, CancellationToken cancellationToken) + { + await Task.Yield(); + throw new InvalidOperationException("note handler failed"); + } +} + +public sealed class AfterFailingNoteHandler : INotificationHandler +{ + private readonly Recorder _recorder; + + public AfterFailingNoteHandler(Recorder recorder) => _recorder = recorder; + + public Task Handle(FailingNote notification, CancellationToken cancellationToken) + { + _recorder.Add("after-failing-note"); + return Task.CompletedTask; + } +} + +/// A stream request that implements another, non-generic interface besides . +public interface IStreamMarker; + +public sealed class MarkedCountdown : IStreamMarker, IStreamRequest +{ + public MarkedCountdown(int from) => From = from; + + public int From { get; } +} + +public sealed class MarkedCountdownHandler : IStreamRequestHandler +{ + public async IAsyncEnumerable Handle(MarkedCountdown request, [EnumeratorCancellation] CancellationToken cancellationToken) + { + for (var i = request.From; i > 0; i--) + { + await Task.Yield(); + yield return i; + } + } +} + +public sealed record Widget(int Id); + +/// Requests nested in a generic type: generic types whose own name carries no backtick. +public static class Repository +{ + public sealed record GetById(int Id) : IRequest; +} + +public sealed class GetWidgetByIdHandler : IRequestHandler.GetById, Widget> +{ + public Task Handle(Repository.GetById request, CancellationToken cancellationToken) + => Task.FromResult(new Widget(request.Id)); +} + +public sealed class DispatchGapTests +{ + private static ServiceProvider BuildProvider() => + new ServiceCollection() + .AddSingleton() + .AddSwitchboard(cfg => cfg.RegisterServicesFromAssemblyContaining()) + .BuildServiceProvider(); + + [Fact] + public async Task Send_of_a_void_request_waits_for_a_handler_that_completes_later() + { + await using var provider = BuildProvider(); + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var send = provider.GetRequiredService().Send(new DeferredVoid(gate.Task)); + + Assert.False(send.IsCompleted); + gate.SetResult(); + await send; + Assert.Equal("deferred:done", provider.GetRequiredService().Joined); + } + + [Fact] + public async Task Send_of_a_void_request_surfaces_a_failure_from_an_async_handler() + { + await using var provider = BuildProvider(); + var sender = provider.GetRequiredService(); + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + // Faulted before Send returns, and faulted later. + await Assert.ThrowsAsync(() => sender.Send(new DeferredVoid(Task.CompletedTask, Fail: true))); + var later = sender.Send(new DeferredVoid(gate.Task, Fail: true)); + gate.SetResult(); + await Assert.ThrowsAsync(() => later); + await Assert.ThrowsAsync(() => sender.Send((object)new DeferredVoid(Task.CompletedTask, Fail: true))); + } + + [Fact] + public async Task A_failing_notification_handler_fails_Publish_and_the_handlers_after_it_do_not_run() + { + await using var provider = new ServiceCollection() + .AddSingleton() + .AddSwitchboard(_ => { }) + .AddTransient, FailingNoteHandler>() + .AddTransient, AfterFailingNoteHandler>() + .BuildServiceProvider(); + + var exception = await Assert.ThrowsAsync( + () => provider.GetRequiredService().Publish(new FailingNote())); + + Assert.Equal("note handler failed", exception.Message); + Assert.Empty(provider.GetRequiredService().Entries); + } + + [Fact] + public async Task Publish_of_an_object_that_is_not_a_notification_names_the_parameter() + { + await using var provider = BuildProvider(); + + var exception = await Assert.ThrowsAsync( + () => provider.GetRequiredService().Publish(new Ping("not a notification"))); + + Assert.Equal("notification", exception.ParamName); + } + + [Fact] + public async Task A_stream_request_that_implements_other_interfaces_too_streams_its_items() + { + await using var provider = BuildProvider(); + var sender = provider.GetRequiredService(); + + Assert.Equal(new[] { 3, 2, 1 }, await sender.CreateStream(new MarkedCountdown(3)).Collect()); + Assert.Equal(new object?[] { 2, 1 }, await sender.CreateStream((object)new MarkedCountdown(2)).Collect()); + } + + [Fact] + public async Task A_request_nested_in_a_generic_type_is_dispatched_and_named_with_its_type_arguments() + { + await using var provider = BuildProvider(); + + var widget = await provider.GetRequiredService().Send(new Repository.GetById(7)); + + Assert.Equal(new Widget(7), widget); + Assert.Equal("GetById", SwitchboardTelemetry.DisplayName(typeof(Repository.GetById))); + } + + [Fact] + public void UseScopePerDispatch_rejects_a_null_callback() + { + var configuration = new SwitchboardConfiguration(); + + var sync = Assert.Throws(() => configuration.UseScopePerDispatch((Action)null!)); + var async = Assert.Throws( + () => configuration.UseScopePerDispatch((Func)null!)); + + Assert.Equal("onScopeCreated", sync.ParamName); + Assert.Equal("onScopeCreated", async.ParamName); + } + + [Fact] + public void AddSwitchboard_rejects_null_services_before_running_the_configuration() + { + var configured = false; + + var exception = Assert.Throws( + () => SwitchboardServiceCollectionExtensions.AddSwitchboard(null!, _ => configured = true)); + + Assert.Equal("services", exception.ParamName); + Assert.False(configured); + } +} diff --git a/tests/Switchboard.Tests/TelemetryGapTests.cs b/tests/Switchboard.Tests/TelemetryGapTests.cs new file mode 100644 index 0000000..120a8fb --- /dev/null +++ b/tests/Switchboard.Tests/TelemetryGapTests.cs @@ -0,0 +1,194 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Diagnostics.Metrics; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace Switchboard.Tests; + +public sealed record FailingTelemetryNote : INotification; + +public sealed class FailingTelemetryNoteHandler : INotificationHandler +{ + public async Task Handle(FailingTelemetryNote notification, CancellationToken cancellationToken) + { + await Task.Yield(); + throw new InvalidOperationException("telemetry note failed"); + } +} + +public sealed record CancellableTelemetryProbe : IRequest; + +public sealed class CancellableTelemetryProbeHandler : IRequestHandler +{ + public async Task Handle(CancellableTelemetryProbe request, CancellationToken cancellationToken) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + return 0; + } +} + +public sealed record CancellableTelemetryStreamProbe : IStreamRequest; + +public sealed class CancellableTelemetryStreamProbeHandler : IStreamRequestHandler +{ + public async IAsyncEnumerable Handle( + CancellableTelemetryStreamProbe request, [EnumeratorCancellation] CancellationToken cancellationToken) + { + for (var i = 0; ; i++) + { + await Task.Delay(1, cancellationToken); + yield return i; + } + } +} + +/// See : listeners are process-wide, so every assertion filters on the probes above. +[Collection(TelemetryTests.CollectionName)] +public sealed class TelemetryGapTests +{ + private static ServiceProvider BuildProvider() => + new ServiceCollection() + .AddSwitchboard(cfg => cfg.RegisterServicesFromAssemblyContaining()) + .BuildServiceProvider(); + + private static ActivityListener ListenToActivities(ConcurrentQueue stopped) + { + var listener = new ActivityListener + { + ShouldListenTo = source => source.Name == SwitchboardTelemetry.ActivitySourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = stopped.Enqueue, + }; + ActivitySource.AddActivityListener(listener); + return listener; + } + + [Fact] + public void Every_duration_histogram_is_measured_in_seconds() + { + RuntimeHelpers.RunClassConstructor(typeof(SwitchboardTelemetry).TypeHandle); + var units = new ConcurrentDictionary(); + using var meterListener = new MeterListener + { + // Only looks: no instrument is enabled, so no dispatch elsewhere starts being observed. + InstrumentPublished = (instrument, _) => + { + if (instrument.Meter.Name == SwitchboardTelemetry.MeterName) + { + units[instrument.Name] = instrument.Unit; + } + }, + // See TelemetryTests: on .NET 8, Dispose() only releases the instruments when this callback is set. + MeasurementsCompleted = (_, _) => { }, + }; + meterListener.Start(); + + Assert.Equal("s", units[SwitchboardTelemetry.RequestDurationInstrument]); + Assert.Equal("s", units[SwitchboardTelemetry.StreamDurationInstrument]); + Assert.Equal("s", units[SwitchboardTelemetry.NotificationDurationInstrument]); + } + + [Fact] + public async Task A_failing_notification_handler_marks_its_own_span_and_the_publish_span_as_errors() + { + var stopped = new ConcurrentQueue(); + using var listener = ListenToActivities(stopped); + await using var provider = BuildProvider(); + + await Assert.ThrowsAsync( + () => provider.GetRequiredService().Publish(new FailingTelemetryNote())); + + var publish = Assert.Single(stopped, a => a.DisplayName == "Publish FailingTelemetryNote"); + var handler = Assert.Single(stopped, a => a.DisplayName == "Handle FailingTelemetryNote"); + Assert.Equal(publish.SpanId, handler.ParentSpanId); + Assert.Equal("FailingTelemetryNote", handler.GetTagItem("switchboard.notification")); + Assert.Equal("FailingTelemetryNoteHandler", handler.GetTagItem("switchboard.handler")); + Assert.Equal(ActivityStatusCode.Error, handler.Status); + Assert.Equal("telemetry note failed", handler.StatusDescription); + Assert.Equal(typeof(InvalidOperationException).FullName, handler.GetTagItem("error.type")); + Assert.Equal(ActivityStatusCode.Error, publish.Status); + + // The OpenTelemetry semantic conventions for a recorded exception. + var exceptionEvent = Assert.Single(handler.Events, e => e.Name == "exception"); + var tags = exceptionEvent.Tags.ToDictionary(t => t.Key, t => t.Value); + Assert.Equal(typeof(InvalidOperationException).FullName, tags["exception.type"]); + Assert.Equal("telemetry note failed", tags["exception.message"]); + Assert.Contains("telemetry note failed", Assert.IsType(tags["exception.stacktrace"])); + Assert.Contains(nameof(FailingTelemetryNoteHandler), (string)tags["exception.stacktrace"]!); + } + + [Fact] + public async Task A_cancelled_request_tags_its_span_with_the_cancellation_but_does_not_mark_it_as_an_error() + { + var stopped = new ConcurrentQueue(); + using var listener = ListenToActivities(stopped); + await using var provider = BuildProvider(); + + await Assert.ThrowsAnyAsync( + () => provider.GetRequiredService().Send(new CancellableTelemetryProbe(), new CancellationToken(canceled: true))); + + var span = Assert.Single(stopped, a => a.DisplayName == "Send CancellableTelemetryProbe"); + Assert.Equal(typeof(TaskCanceledException).FullName, span.GetTagItem("error.type")); + Assert.Equal(ActivityStatusCode.Unset, span.Status); + Assert.DoesNotContain(span.Events, e => e.Name == "exception"); + } + + [Fact] + public async Task A_cancelled_stream_tags_its_span_with_the_cancellation_but_does_not_mark_it_as_an_error() + { + var stopped = new ConcurrentQueue(); + using var listener = ListenToActivities(stopped); + await using var provider = BuildProvider(); + using var cancellation = new CancellationTokenSource(); + + await Assert.ThrowsAnyAsync(async () => + { + await foreach (var _ in provider.GetRequiredService() + .CreateStream(new CancellableTelemetryStreamProbe(), cancellation.Token)) + { + cancellation.Cancel(); + } + }); + + var span = Assert.Single(stopped, a => a.DisplayName == "CreateStream CancellableTelemetryStreamProbe"); + Assert.Equal(typeof(TaskCanceledException).FullName, span.GetTagItem("error.type")); + Assert.Equal(1, span.GetTagItem("switchboard.stream.items")); + Assert.Equal(ActivityStatusCode.Unset, span.Status); + Assert.DoesNotContain(span.Events, e => e.Name == "exception"); + } + + [Fact] + public async Task A_traced_stream_still_reaches_the_handler_with_the_WithCancellation_token() + { + var stopped = new ConcurrentQueue(); + using var listener = ListenToActivities(stopped); + await using var provider = BuildProvider(); + using var cancellation = new CancellationTokenSource(); + var received = 0; + + // The handler never ends on its own; only the consumer's token, passed through the traced stream, stops it. + await Assert.ThrowsAnyAsync(async () => + { + await foreach (var _ in provider.GetRequiredService() + .CreateStream(new CancellableTelemetryStreamProbe()) + .WithCancellation(cancellation.Token)) + { + cancellation.Cancel(); + if (++received > 50) + { + break; + } + } + }); + + Assert.Equal(1, received); + Assert.Single(stopped, a => a.DisplayName == "CreateStream CancellableTelemetryStreamProbe"); + } +} diff --git a/tests/Switchboard.Tests/ValidationGapTests.cs b/tests/Switchboard.Tests/ValidationGapTests.cs new file mode 100644 index 0000000..6041cec --- /dev/null +++ b/tests/Switchboard.Tests/ValidationGapTests.cs @@ -0,0 +1,248 @@ +using System; +using System.Linq; +using System.Text.RegularExpressions; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace Switchboard.Tests; + +// --- Request shapes validation must not mistake for requests that need a handler --- + +/// A base request: only the concrete requests derived from it are ever sent. +public abstract record PagedQuery : IRequest; + +public sealed record CountWidgets : PagedQuery; + +public sealed class CountWidgetsHandler : IRequestHandler +{ + public Task Handle(CountWidgets request, CancellationToken cancellationToken) => Task.FromResult(0); +} + +/// A marker interface for a family of commands; never sent itself. +public interface IVoidCommand : IRequest; + +/// An open generic request: only its closed forms can ever be sent. +public sealed record GenericLookup(int Id) : IRequest; + +/// A void request without a parameterless constructor. +public sealed record VoidWithArgument(int Value) : IRequest; + +public sealed class VoidWithArgumentHandler : IRequestHandler +{ + public Task Handle(VoidWithArgument request, CancellationToken cancellationToken) => Task.CompletedTask; +} + +// --- Behaviors that constrain TRequest to IRequest plus something else --- + +/// The trap, narrowed to requests: it skips exactly the audited void requests. +public sealed class AuditedTypedOnlyBehavior : IPipelineBehavior + where TRequest : IAudited, IRequest +{ + public Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken) + => next(cancellationToken); +} + +/// Only for reference-type responses (a cache, say), so it never applies to anyway. +public sealed class ReferenceResponseBehavior : IPipelineBehavior + where TRequest : IRequest + where TResponse : class +{ + public Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken) + => next(cancellationToken); +} + +/// Only for value-type responses, which is: void requests are skipped by the trap alone. +public sealed class ValueResponseBehavior : IPipelineBehavior + where TRequest : IRequest + where TResponse : struct +{ + public Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken) + => next(cancellationToken); +} + +/// Only for value-type requests; every void request in this assembly is a class. +public sealed class ValueRequestBehavior : IPipelineBehavior + where TRequest : struct, IRequest +{ + public Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken) + => next(cancellationToken); +} + +/// Only for requests with a parameterless constructor. +public sealed class NewableRequestBehavior : IPipelineBehavior + where TRequest : IRequest, new() +{ + public Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken) + => next(cancellationToken); +} + +/// A constraint that refers to the behavior's own type parameter, as a de-duplicating behavior might use. +public sealed class EquatableRequestBehavior : IPipelineBehavior + where TRequest : IRequest, IEquatable +{ + public Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken) + => next(cancellationToken); +} + +// --- An open generic service that is not a pipeline behavior --- + +public interface IRequestAuditor; + +public sealed class TypedRequestAuditor : IRequestAuditor + where TRequest : IRequest +{ +} + +public sealed class ValidationGapTests +{ + private static IServiceCollection Scanned(Action? configure = null) + => new ServiceCollection() + .AddSingleton() + .AddSwitchboard(cfg => + { + cfg.RegisterServicesFromAssemblyContaining(); + configure?.Invoke(cfg); + }); + + private static SwitchboardValidationException Invalid(IServiceCollection services, Action? configure = null) + => Assert.Throws(() => services.ValidateSwitchboard(configure)); + + private static string[] SkipErrors(Type openBehavior) + { + var exception = Invalid(Scanned(cfg => cfg.AddOpenBehavior(openBehavior)), o => o.RequireHandlerForEveryRequest = false); + return exception.Errors.ToArray(); + } + + /// Every concrete void request this assembly declares, the set validation scans. + private static Type[] VoidRequests() + => typeof(ValidationGapTests).Assembly.GetTypes() + .Where(t => !t.IsAbstract && !t.IsInterface && !t.IsGenericTypeDefinition && typeof(IRequest).IsAssignableFrom(t)) + .ToArray(); + + private static (int Count, string[] Listed, int? More) ParseSkipped(string error) + { + var match = Regex.Match( + error, @"never runs for (?\d+) void request\(s\) \((?[^()]*?)(?: and (?\d+) more)?\) because it constrains "); + Assert.True(match.Success, error); + + return ( + int.Parse(match.Groups["count"].Value), + match.Groups["listed"].Value.Split(", "), + match.Groups["more"].Success ? int.Parse(match.Groups["more"].Value) : null); + } + + [Fact] + public void Reports_a_void_request_without_a_handler() + { + var exception = Invalid(Scanned()); + + Assert.Contains("VoidOrphan has no handler. Add a class implementing IRequestHandler.", exception.Errors); + Assert.DoesNotContain(exception.Errors, e => e.StartsWith("FireAndForget ")); + } + + [Fact] + public void Abstract_interface_and_open_generic_request_types_are_not_reported_as_missing_a_handler() + { + var exception = Invalid(Scanned()); + + Assert.Contains(exception.Errors, e => e.StartsWith("Orphan ")); + Assert.DoesNotContain(exception.Errors, e => e.StartsWith("PagedQuery ")); + Assert.DoesNotContain(exception.Errors, e => e.StartsWith("CountWidgets ")); + Assert.DoesNotContain(exception.Errors, e => e.StartsWith("IVoidCommand ")); + Assert.DoesNotContain(exception.Errors, e => e.StartsWith("GenericLookup<")); + } + + [Fact] + public void A_typed_only_behavior_names_five_void_requests_and_counts_the_rest() + { + // The count covers every void request, VoidWithArgument (no parameterless constructor) included. + var voidRequests = VoidRequests().Select(SwitchboardTelemetry.DisplayName).ToArray(); + Assert.True(voidRequests.Length > 5, "this test needs more than five void requests in the assembly"); + + var (count, listed, more) = ParseSkipped(Assert.Single(SkipErrors(typeof(TypedOnlyBehavior<,>)))); + + Assert.Equal(voidRequests.Length, count); + Assert.Equal(5, listed.Length); + Assert.Equal(5, listed.Distinct().Count()); + Assert.All(listed, name => Assert.Contains(name, voidRequests)); + Assert.Equal(voidRequests.Length - 5, more); + } + + [Fact] + public void A_behavior_that_also_requires_a_marker_is_reported_for_exactly_the_void_requests_that_carry_it() + { + var error = Assert.Single(SkipErrors(typeof(AuditedTypedOnlyBehavior<,>))); + + Assert.Equal( + "AuditedTypedOnlyBehavior never runs for 1 void request(s) (AuditedVoidCommand) because it constrains " + + "TRequest to IRequest, which void requests do not implement. Constrain it to IBaseRequest to cover every request.", + error); + } + + [Fact] + public void A_behavior_that_requires_a_reference_type_response_is_not_reported() + { + // It could never run for a Unit response, IRequest or not: excluding void requests is on purpose. + Scanned(cfg => cfg.AddOpenBehavior(typeof(ReferenceResponseBehavior<,>))) + .ValidateSwitchboard(o => o.RequireHandlerForEveryRequest = false); + } + + [Fact] + public void A_behavior_that_requires_a_value_type_response_is_still_reported() + { + // Unit is a value type, so only the IRequest constraint keeps it away from void requests. + var (count, _, _) = ParseSkipped(Assert.Single(SkipErrors(typeof(ValueResponseBehavior<,>)))); + + Assert.Equal(VoidRequests().Length, count); + } + + [Fact] + public void A_behavior_limited_to_value_type_requests_is_not_reported_for_void_requests_that_are_classes() + { + Assert.All(VoidRequests(), t => Assert.False(t.IsValueType)); + + Scanned(cfg => cfg.AddOpenBehavior(typeof(ValueRequestBehavior<,>))) + .ValidateSwitchboard(o => o.RequireHandlerForEveryRequest = false); + } + + [Fact] + public void A_behavior_that_requires_a_parameterless_constructor_counts_only_the_void_requests_that_have_one() + { + var newable = VoidRequests().Count(t => t.GetConstructor(Type.EmptyTypes) is not null); + Assert.True(newable > 0 && newable < VoidRequests().Length, "this test needs void requests with and without a parameterless constructor"); + + var (count, listed, _) = ParseSkipped(Assert.Single(SkipErrors(typeof(NewableRequestBehavior<,>)))); + + Assert.Equal(newable, count); + Assert.DoesNotContain(nameof(VoidWithArgument), listed); + } + + [Fact] + public void A_constraint_on_the_behaviors_own_type_parameter_is_checked_against_each_request() + { + var equatable = VoidRequests().Count(t => typeof(IEquatable<>).MakeGenericType(t).IsAssignableFrom(t)); + Assert.True(equatable > 0, "records implement IEquatable of themselves"); + + var (count, _, _) = ParseSkipped(Assert.Single(SkipErrors(typeof(EquatableRequestBehavior<,>)))); + + Assert.Equal(equatable, count); + } + + [Fact] + public void An_open_generic_service_that_is_not_a_pipeline_behavior_is_never_reported() + { + var services = Scanned().AddTransient(typeof(IRequestAuditor<,>), typeof(TypedRequestAuditor<,>)); + + services.ValidateSwitchboard(o => o.RequireHandlerForEveryRequest = false); + } + + [Fact] + public void ValidateSwitchboard_names_the_null_services_argument() + { + var exception = Assert.Throws(() => SwitchboardValidationExtensions.ValidateSwitchboard(null!)); + + Assert.Equal("services", exception.ParamName); + } +}