v7: Obvs.Testing - in-memory transport and test harness

.github/workflows/dotnet.yml

@@ -108,6 +108,7 @@ jobs:
         dotnet pack --no-build --configuration Release --include-symbols --output nupkgs -p:PackageVersion=${{ steps.gitversion.outputs.semVer }} Obvs.Serialization.ProtoBuf
         dotnet pack --no-build --configuration Release --include-symbols --output nupkgs -p:PackageVersion=${{ steps.gitversion.outputs.semVer }} Obvs.Serialization.Xml
         dotnet pack --no-build --configuration Release --include-symbols --output nupkgs -p:PackageVersion=${{ steps.gitversion.outputs.semVer }} Obvs.Serialization.Yaml
+        dotnet pack --no-build --configuration Release --include-symbols --output nupkgs -p:PackageVersion=${{ steps.gitversion.outputs.semVer }} Obvs.Testing
     - name: NuGet Login
       if: github.event_name == 'push' && (github.ref == 'refs/heads/master' || startsWith(github.ref, 'refs/tags/'))
       id: nuget

Directory.Packages.props

@@ -9,8 +9,10 @@
         <PackageVersion Include="Confluent.Kafka" Version="2.16.0"/>
         <PackageVersion Include="Elastic.Clients.Elasticsearch" Version="9.5.3"/>
         <PackageVersion Include="MessagePack" Version="3.1.12"/>
+        <PackageVersion Include="Microsoft.Bcl.TimeProvider" Version="10.0.13"/>
         <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.13"/>
         <PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="10.0.13"/>
+        <PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.13"/>
         <PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.13"/>
         <PackageVersion Include="MsgPack.Cli" Version="1.0.1"/>
         <PackageVersion Include="NATS.Client.Core" Version="3.3.1"/>
@@ -27,7 +29,7 @@
     <!-- Tests -->
     <ItemGroup>
         <PackageVersion Include="FakeItEasy" Version="9.0.1"/>
-        <PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.13"/>
+        <PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="10.11.0"/>
         <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="18.11.0"/>
         <!-- Keep in step with System.Reactive -->
         <PackageVersion Include="Microsoft.Reactive.Testing" Version="7.0.0"/>

Obvs.Testing.Tests/Messages.cs

@@ -0,0 +1,26 @@
+using Obvs.Types;
+
+namespace Obvs.Testing.Tests
+{
+    public interface IOrderMessage : IMessage { }
+
+    public interface IShippingMessage : IMessage { }
+
+    public record PlaceOrder(string OrderId) : IOrderMessage, ICommand;
+
+    public record OrderPlaced(string OrderId) : IOrderMessage, IEvent;
+
+    public record GetQuote(string Sku) : IOrderMessage, IRequest
+    {
+        public string RequestId { get; set; }
+        public string RequesterId { get; set; }
+    }
+
+    public record Quote(string Sku, decimal Price) : IOrderMessage, IResponse
+    {
+        public string RequestId { get; set; }
+        public string RequesterId { get; set; }
+    }
+
+    public record ShipOrder(string OrderId) : IShippingMessage, ICommand;
+}

Obvs.Testing.Tests/Obvs.Testing.Tests.csproj

@@ -0,0 +1,20 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.TimeProvider.Testing" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" />
+    <PackageReference Include="Microsoft.Reactive.Testing" />
+    <PackageReference Include="xunit.v3" />
+    <PackageReference Include="xunit.runner.visualstudio">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Obvs.Testing\Obvs.Testing.csproj" />
+  </ItemGroup>
+</Project>

Obvs.Testing.Tests/TestInMemoryTransport.cs

@@ -0,0 +1,154 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive.Linq;
+using System.Reactive.Threading.Tasks;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Reactive.Testing;
+using Obvs.Hosting;
+using Obvs.Types;
+using Xunit;
+
+namespace Obvs.Testing.Tests
+{
+    // Exercises the transport on its own, through AddObvs and the bus, without the harness
+    public class TestInMemoryTransport
+    {
+        private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
+
+        [Fact]
+        public async Task ShouldDeliverCommandsFromClientToServer()
+        {
+            using var host = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory()));
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+
+            var received = bus.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            await bus.SendAsync(new PlaceOrder("o-1"));
+
+            Assert.Equal("o-1", (await received.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+        }
+
+        [Fact]
+        public async Task ShouldDeliverEventsFromServerToClient()
+        {
+            using var host = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory()));
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+
+            var received = bus.Events.OfType<OrderPlaced>().FirstAsync().ToTask();
+            await bus.PublishAsync(new OrderPlaced("o-1"));
+
+            Assert.Equal("o-1", (await received.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+        }
+
+        [Fact]
+        public async Task ShouldReturnResponsesToTheRequester()
+        {
+            using var host = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory()));
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+            using var server = bus.Requests.OfType<GetQuote>()
+                .Subscribe(request => _ = bus.ReplyAsync(request, new Quote(request.Sku, 9.99m)));
+
+            var quote = await bus.GetResponses(new GetQuote("sku-1")).OfType<Quote>().FirstAsync().ToTask()
+                .WaitAsync(Timeout, TestContext.Current.CancellationToken);
+
+            Assert.Equal(new Quote("sku-1", 9.99m), quote with { RequestId = null, RequesterId = null });
+        }
+
+        [Fact]
+        public async Task ShouldOnlyDeliverMessagesToTheirOwnService()
+        {
+            // The bus merges every service's commands, so a command delivered to both services would arrive twice.
+            // The test scheduler makes delivery deterministic, so "exactly once" needs no waiting.
+            var scheduler = new TestScheduler();
+            var broker = new InMemoryBroker(scheduler);
+            using var host = await StartHostAsync(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service.UseInMemory(broker))
+                .AddService<IShippingMessage>("shipping", service => service.UseInMemory(broker)));
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+
+            var received = new List<ICommand>();
+            using var subscription = bus.Commands.Subscribe(received.Add);
+            await bus.SendAsync(new ShipOrder("o-1"));
+            scheduler.Start();
+
+            Assert.Equal(new ICommand[] { new ShipOrder("o-1") }, received);
+        }
+
+        [Fact]
+        public async Task ShouldDeliverMessagesInTheOrderTheyWereSent()
+        {
+            using var host = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory()));
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+            var ids = Enumerable.Range(1, 100).Select(i => $"o-{i}").ToArray();
+
+            var received = bus.Commands.OfType<PlaceOrder>().Take(ids.Length).Select(command => command.OrderId).ToArray().ToTask();
+            foreach (var id in ids)
+            {
+                await bus.SendAsync(new PlaceOrder(id));
+            }
+
+            Assert.Equal(ids, await received.WaitAsync(Timeout, TestContext.Current.CancellationToken));
+        }
+
+        [Fact]
+        public async Task ShouldGiveEachContainerItsOwnBroker()
+        {
+            using var hostA = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory()));
+            using var hostB = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory()));
+            var busA = hostA.Services.GetRequiredService<IServiceBus>();
+            var busB = hostB.Services.GetRequiredService<IServiceBus>();
+
+            Assert.NotSame(hostA.Services.GetRequiredService<InMemoryBroker>(), hostB.Services.GetRequiredService<InMemoryBroker>());
+
+            var receivedByB = busB.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            var receivedByA = busA.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            await busA.SendAsync(new PlaceOrder("o-1"));
+
+            await receivedByA.WaitAsync(Timeout, TestContext.Current.CancellationToken);
+            Assert.False(receivedByB.IsCompleted);
+        }
+
+        [Fact]
+        public async Task ShouldConnectHostsThatShareABroker()
+        {
+            var broker = new InMemoryBroker();
+            using var clientHost = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory(broker).AsClient()));
+            using var serverHost = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory(broker).AsServer()));
+
+            var received = serverHost.Services.GetRequiredService<IServiceBus>().Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            await clientHost.Services.GetRequiredService<IServiceBusClient>().SendAsync(new PlaceOrder("o-1"));
+
+            Assert.Equal("o-1", (await received.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+        }
+
+        [Fact]
+        public async Task ShouldDeliverOnTheBrokersScheduler()
+        {
+            var scheduler = new TestScheduler();
+            var broker = new InMemoryBroker(scheduler);
+            using var host = await StartHostAsync(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory(broker)));
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+
+            PlaceOrder received = null;
+            using var subscription = bus.Commands.OfType<PlaceOrder>().Subscribe(command => received = command);
+            await bus.SendAsync(new PlaceOrder("o-1"));
+
+            Assert.Null(received);
+
+            scheduler.Start();
+
+            Assert.Equal("o-1", received?.OrderId);
+        }
+
+        private static async Task<IHost> StartHostAsync(Action<ObvsBuilder> configure)
+        {
+            var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
+            builder.Services.AddObvs(configure);
+            var host = builder.Build();
+            await host.StartAsync(TestContext.Current.CancellationToken);
+            return host;
+        }
+    }
+}

Obvs.Testing.Tests/TestObvsTestHarness.cs

@@ -0,0 +1,153 @@
+using System;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Time.Testing;
+using Obvs.Hosting;
+using Xunit;
+
+namespace Obvs.Testing.Tests
+{
+    public class TestObvsTestHarness
+    {
+        [Fact]
+        public async Task ShouldReturnAMessagePublishedAfterWaitingStarts()
+        {
+            await using var harness = await StartOrdersHarnessAsync();
+
+            var waiting = harness.Published<OrderPlaced>();
+            await harness.Bus.PublishAsync(new OrderPlaced("o-1"));
+
+            Assert.Equal("o-1", (await waiting).OrderId);
+        }
+
+        [Fact]
+        public async Task ShouldReturnAMessagePublishedBeforeWaitingStarts()
+        {
+            await using var harness = await StartOrdersHarnessAsync();
+            await harness.Bus.PublishAsync(new OrderPlaced("o-1"));
+
+            Assert.Equal("o-1", (await harness.Published<OrderPlaced>()).OrderId);
+        }
+
+        [Fact]
+        public async Task ShouldReturnOnlyAMessageMatchingThePredicate()
+        {
+            await using var harness = await StartOrdersHarnessAsync();
+            await harness.Bus.PublishAsync(new OrderPlaced("o-1"));
+            await harness.Bus.PublishAsync(new OrderPlaced("o-2"));
+
+            Assert.Equal("o-2", (await harness.Published<OrderPlaced>(placed => placed.OrderId == "o-2")).OrderId);
+        }
+
+        [Fact]
+        public async Task ShouldTestAServiceEndToEnd()
+        {
+            await using var harness = await StartOrdersHarnessAsync();
+            using var service = harness.Bus.Commands.OfType<PlaceOrder>()
+                .Subscribe(command => _ = harness.Bus.PublishAsync(new OrderPlaced(command.OrderId)));
+
+            await harness.Client.SendAsync(new PlaceOrder("o-1"));
+
+            await harness.Published<OrderPlaced>(placed => placed.OrderId == "o-1");
+        }
+
+        [Fact]
+        public async Task ShouldSeeCommandsSentByAClientOnlyApp()
+        {
+            await using var harness = await ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service.UseInMemory().AsClient())),
+                cancellationToken: TestContext.Current.CancellationToken);
+
+            await harness.Client.SendAsync(new PlaceOrder("o-1"));
+
+            Assert.Equal("o-1", (await harness.Sent<PlaceOrder>()).OrderId);
+        }
+
+        [Fact]
+        public async Task ShouldSeeRequestsAndReplies()
+        {
+            await using var harness = await StartOrdersHarnessAsync();
+            using var server = harness.Bus.Requests.OfType<GetQuote>()
+                .Subscribe(request => _ = harness.Bus.ReplyAsync(request, new Quote(request.Sku, 9.99m)));
+
+            using var requester = harness.Client.GetResponses(new GetQuote("sku-1")).Subscribe(_ => { });
+
+            Assert.Equal("sku-1", (await harness.Requested<GetQuote>()).Sku);
+            Assert.Equal(9.99m, (await harness.Replied<Quote>(quote => quote.Sku == "sku-1")).Price);
+        }
+
+        [Fact]
+        public async Task ShouldRecordEveryMessageInOrder()
+        {
+            await using var harness = await StartOrdersHarnessAsync();
+            await harness.Client.SendAsync(new PlaceOrder("o-1"));
+            await harness.Bus.PublishAsync(new OrderPlaced("o-1"));
+            await harness.Published<OrderPlaced>();
+
+            Assert.Equal(new object[] { new PlaceOrder("o-1"), new OrderPlaced("o-1") }, harness.Messages);
+        }
+
+        [Fact]
+        public async Task ShouldListTheMessagesSeenWhenWaitingTimesOut()
+        {
+            await using var harness = await StartOrdersHarnessAsync();
+            await harness.Bus.PublishAsync(new OrderPlaced("o-1"));
+            await harness.Published<OrderPlaced>();
+
+            var exception = await Assert.ThrowsAsync<ObvsTestTimeoutException>(
+                () => harness.Published<OrderPlaced>(placed => placed.OrderId == "o-2", TimeSpan.FromMilliseconds(100)));
+
+            Assert.Contains("OrderPlaced", exception.Message);
+            Assert.Contains("o-1", exception.Message);
+        }
+
+        [Fact]
+        public async Task ShouldUseTheDefaultTimeoutFromTheOptions()
+        {
+            await using var harness = await ObvsTestHarness.StartAsync(
+                services => services.AddObvs(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory())),
+                options => options.DefaultTimeout = TimeSpan.FromMilliseconds(100),
+                TestContext.Current.CancellationToken);
+
+            await Assert.ThrowsAsync<ObvsTestTimeoutException>(() => harness.Published<OrderPlaced>())
+                .WaitAsync(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken);
+        }
+
+        [Fact]
+        public async Task ShouldTimeOutOnTheTimeProvider()
+        {
+            var time = new FakeTimeProvider();
+            await using var harness = await ObvsTestHarness.StartAsync(
+                services => services.AddObvs(obvs => obvs.AddService<IOrderMessage>("orders", service => service.UseInMemory())),
+                options => options.TimeProvider = time,
+                TestContext.Current.CancellationToken);
+
+            var waiting = harness.Published<OrderPlaced>(timeout: TimeSpan.FromMinutes(5));
+            time.Advance(TimeSpan.FromMinutes(6));
+
+            // Completes because virtual time passed, long before five real minutes
+            await Assert.ThrowsAsync<ObvsTestTimeoutException>(() => waiting)
+                .WaitAsync(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken);
+        }
+
+        [Fact]
+        public async Task ShouldStopTheHostWhenDisposed()
+        {
+            var harness = await StartOrdersHarnessAsync();
+            var lifetime = harness.Services.GetRequiredService<IHostApplicationLifetime>();
+
+            await harness.DisposeAsync();
+
+            Assert.True(lifetime.ApplicationStopped.IsCancellationRequested);
+        }
+
+        private static Task<ObvsTestHarness> StartOrdersHarnessAsync()
+        {
+            return ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service.UseInMemory())),
+                cancellationToken: TestContext.Current.CancellationToken);
+        }
+    }
+}