Skip to content

Commit ccf91ae

Browse files
VulthilVulthil
authored
test: cover the broker-outbox seam's filters, parallel dispatch, and end-to-end relay (#344)
1 parent 95c9b1a commit ccf91ae

9 files changed

Lines changed: 818 additions & 0 deletions

‎tests/Vulthil.IntegrationTests/TransactionalOutboxIntegrationTests.cs‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
using Microsoft.EntityFrameworkCore;
22
using Microsoft.Extensions.DependencyInjection;
3+
using Vulthil.Extensions.Testing;
34
using Vulthil.IntegrationTests.Fixtures;
45
using Vulthil.Messaging.Abstractions.Publishers;
56
using Vulthil.Messaging.TestHarness;
7+
using Vulthil.Results;
68
using Vulthil.SharedKernel.Outbox;
79
using Vulthil.SharedKernel.Outbox.EntityFrameworkCore;
810
using Vulthil.TestHost.Data;
@@ -15,6 +17,7 @@ public sealed class TransactionalOutboxIntegrationTests(TestHarnessWebApplicatio
1517
: BaseIntegrationTestCase<TestHarnessWebApplicationFactory, Program>(factory, testOutputHelper), IClassFixture<TestHarnessWebApplicationFactory>
1618
{
1719
private ITestHarness Harness => Factory.Services.GetRequiredService<ITestHarness>();
20+
private IRequester Requester => Factory.Services.GetRequiredService<IRequester>();
1821

1922
[Fact]
2023
public async Task PublishInsideATransactionIsCapturedIntoTheOutboxAndNotSentDirectly()
@@ -44,4 +47,72 @@ await dbContext.ExecuteInTransactionAsync(async cancellationToken =>
4447
capturedDuringTransaction.ShouldBeGreaterThan(0);
4548
sentDuringTransaction.ShouldBeFalse();
4649
}
50+
51+
[Fact]
52+
public async Task CommittingATransactionRelaysTheMessageAndTheConsumerReceivesItWithTheStableMessageId()
53+
{
54+
// Arrange
55+
var id = Guid.NewGuid();
56+
var messageId = Guid.NewGuid().ToString();
57+
var dbContext = ScopedServices.GetRequiredService<TestHostDbContext>();
58+
var publisher = ScopedServices.GetRequiredService<IPublisher>();
59+
60+
// Act — publish inside a committed transaction, then wait for the real relay to dispatch it.
61+
await dbContext.ExecuteInTransactionAsync(async cancellationToken =>
62+
{
63+
await publisher.PublishAsync(
64+
new ProbeCreatedIntegrationEvent(id),
65+
context =>
66+
{
67+
context.SetMessageId(messageId);
68+
return ValueTask.CompletedTask;
69+
},
70+
cancellationToken);
71+
72+
return 0;
73+
}, CancellationToken);
74+
75+
var polled = await Polling.WaitAsync(TimeSpan.FromSeconds(15), async () =>
76+
{
77+
var consumed = Harness.Consumed<ProbeCreatedIntegrationEvent>().FirstOrDefault(message => message.Message.Id == id);
78+
return consumed is not null
79+
? Result.Success(consumed)
80+
: Result.Failure<CapturedMessage<ProbeCreatedIntegrationEvent>>(
81+
Error.NotFound("Probe.NotYetRelayed", "The captured message has not been relayed and consumed yet."));
82+
}, CancellationToken);
83+
84+
// Assert
85+
polled.IsSuccess.ShouldBeTrue();
86+
polled.Value.Envelope.MessageId.ShouldBe(messageId);
87+
88+
var sideEffects = await Requester.RequestAsync<GetProbeSideEffects, List<ProbeSideEffectDto>>(new GetProbeSideEffects(id), CancellationToken);
89+
sideEffects.IsSuccess.ShouldBeTrue();
90+
sideEffects.Value.ShouldContain(sideEffect => sideEffect.ProbeId == id);
91+
}
92+
93+
[Fact]
94+
public async Task RollingBackATransactionNeverRelaysTheCapturedMessage()
95+
{
96+
// Arrange
97+
var id = Guid.NewGuid();
98+
var dbContext = ScopedServices.GetRequiredService<TestHostDbContext>();
99+
var publisher = ScopedServices.GetRequiredService<IPublisher>();
100+
101+
// Act — the operation reports failure, so ExecuteInTransactionAsync rolls back instead of committing.
102+
await dbContext.ExecuteInTransactionAsync(
103+
async cancellationToken =>
104+
{
105+
await publisher.PublishAsync(new ProbeCreatedIntegrationEvent(id), cancellationToken);
106+
return false;
107+
},
108+
static operationSucceeded => operationSucceeded,
109+
CancellationToken);
110+
await Task.Delay(TimeSpan.FromSeconds(1), CancellationToken);
111+
112+
// Assert — fetch every row and filter in memory: Content is mapped to jsonb, which Npgsql cannot LIKE-match server-side.
113+
var outboxMessages = await dbContext.OutboxMessages.AsNoTracking().ToListAsync(CancellationToken);
114+
outboxMessages.ShouldNotContain(message => message.Content.Contains(id.ToString()));
115+
Harness.Published<ProbeCreatedIntegrationEvent>().ShouldNotContain(message => message.Message.Id == id);
116+
Harness.Consumed<ProbeCreatedIntegrationEvent>().ShouldNotContain(message => message.Message.Id == id);
117+
}
47118
}

‎tests/Vulthil.IntegrationTests/Vulthil.IntegrationTests.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
</ItemGroup>
1111

1212
<ItemGroup>
13+
<ProjectReference Include="..\..\src\Vulthil.Extensions.Testing\Vulthil.Extensions.Testing.csproj" />
1314
<ProjectReference Include="..\..\src\Vulthil.Messaging.Inbox.Cosmos\Vulthil.Messaging.Inbox.Cosmos.csproj" />
1415
<ProjectReference Include="..\..\src\Vulthil.Messaging.TestHarness\Vulthil.Messaging.TestHarness.csproj" />
1516
<ProjectReference Include="..\..\src\Vulthil.SharedKernel.Infrastructure.Cosmos\Vulthil.SharedKernel.Infrastructure.Cosmos.csproj" />

‎tests/Vulthil.Messaging.Outbox.Tests/BrokerOutboxDispatcherTests.cs‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,6 @@
1+
using System.Text.Json;
2+
using Vulthil.Messaging.Abstractions.Publishers;
3+
using Vulthil.Messaging.Transport;
14
using Vulthil.SharedKernel.Outbox;
25
using Vulthil.xUnit;
36

@@ -26,4 +29,123 @@ public async Task DispatchAsyncThrowsDescriptiveErrorWhenMessageTypeCannotBeReso
2629
// Assert
2730
exception.Message.ShouldContain("Vulthil.Nonexistent.PhantomMessage");
2831
}
32+
33+
[Fact]
34+
public async Task DispatchAsyncPublishesThePayloadWithTheCapturedMessageIdCorrelationIdRoutingKeyAndHeaders()
35+
{
36+
// Arrange
37+
var jsonOptions = new JsonSerializerOptions();
38+
GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions);
39+
Func<IPublishContext, ValueTask>? capturedConfigure = null;
40+
GetMock<ITransportPublisher>()
41+
.Setup(publisher => publisher.PublishAsync(It.IsAny<TestMessage>(), It.IsAny<Func<IPublishContext, ValueTask>>(), It.IsAny<CancellationToken>()))
42+
.Callback<TestMessage, Func<IPublishContext, ValueTask>, CancellationToken>((_, configure, _) => capturedConfigure = configure)
43+
.Returns(Task.CompletedTask);
44+
var metadata = new BrokerOutboxMetadata
45+
{
46+
MessageId = "stable-message-id",
47+
CorrelationId = "correlation-1",
48+
RoutingKey = "routing-key-1",
49+
Headers = new Dictionary<string, object?> { ["tenant"] = "acme" },
50+
};
51+
var dispatcher = CreateInstance<BrokerOutboxDispatcher>();
52+
var message = new OutboxMessageData(
53+
Id: Guid.NewGuid(),
54+
Type: typeof(TestMessage).FullName!,
55+
Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions),
56+
TraceParent: null,
57+
TraceState: null,
58+
Destination: OutboxDestination.Publish,
59+
Metadata: JsonSerializer.Serialize(metadata, jsonOptions));
60+
61+
// Act
62+
await dispatcher.DispatchAsync(message, CancellationToken);
63+
64+
// Assert
65+
GetMock<ITransportPublisher>().Verify(
66+
publisher => publisher.PublishAsync(It.Is<TestMessage>(m => m.Value == "hello"), It.IsAny<Func<IPublishContext, ValueTask>>(), CancellationToken),
67+
Times.Once);
68+
capturedConfigure.ShouldNotBeNull();
69+
var appliedContext = new PublishContext();
70+
await capturedConfigure!(appliedContext);
71+
appliedContext.MessageId.ShouldBe("stable-message-id");
72+
appliedContext.CorrelationId.ShouldBe("correlation-1");
73+
appliedContext.RoutingKey.ShouldBe("routing-key-1");
74+
appliedContext.Headers["tenant"].ShouldBe("acme");
75+
}
76+
77+
[Fact]
78+
public async Task DispatchAsyncSendsThePayloadToTheCapturedDestinationAddressWithFidelity()
79+
{
80+
// Arrange
81+
var jsonOptions = new JsonSerializerOptions();
82+
GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions);
83+
var sendEndpoint = new Mock<ISendEndpoint>();
84+
Func<IPublishContext, ValueTask>? capturedConfigure = null;
85+
sendEndpoint
86+
.Setup(endpoint => endpoint.SendAsync(It.IsAny<TestMessage>(), It.IsAny<Func<IPublishContext, ValueTask>>(), It.IsAny<CancellationToken>()))
87+
.Callback<TestMessage, Func<IPublishContext, ValueTask>, CancellationToken>((_, configure, _) => capturedConfigure = configure)
88+
.Returns(Task.CompletedTask);
89+
var destinationAddress = new Uri("queue:order-commands");
90+
GetMock<ITransportSendEndpointProvider>()
91+
.Setup(provider => provider.GetSendEndpointAsync(destinationAddress, It.IsAny<CancellationToken>()))
92+
.ReturnsAsync(sendEndpoint.Object);
93+
var metadata = new BrokerOutboxMetadata
94+
{
95+
MessageId = "stable-send-id",
96+
DestinationAddress = destinationAddress.ToString(),
97+
Headers = new Dictionary<string, object?> { ["tenant"] = "acme" },
98+
};
99+
var dispatcher = CreateInstance<BrokerOutboxDispatcher>();
100+
var message = new OutboxMessageData(
101+
Id: Guid.NewGuid(),
102+
Type: typeof(TestMessage).FullName!,
103+
Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions),
104+
TraceParent: null,
105+
TraceState: null,
106+
Destination: OutboxDestination.Send,
107+
Metadata: JsonSerializer.Serialize(metadata, jsonOptions));
108+
109+
// Act
110+
await dispatcher.DispatchAsync(message, CancellationToken);
111+
112+
// Assert
113+
sendEndpoint.Verify(
114+
endpoint => endpoint.SendAsync(It.Is<TestMessage>(m => m.Value == "hello"), It.IsAny<Func<IPublishContext, ValueTask>>(), CancellationToken),
115+
Times.Once);
116+
capturedConfigure.ShouldNotBeNull();
117+
var appliedContext = new PublishContext();
118+
await capturedConfigure!(appliedContext);
119+
appliedContext.MessageId.ShouldBe("stable-send-id");
120+
appliedContext.Headers["tenant"].ShouldBe("acme");
121+
}
122+
123+
[Fact]
124+
public async Task DispatchAsyncThrowsDescriptiveErrorWhenASendMessageHasNoDestinationAddress()
125+
{
126+
// Arrange
127+
var jsonOptions = new JsonSerializerOptions();
128+
GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions);
129+
var dispatcher = CreateInstance<BrokerOutboxDispatcher>();
130+
var message = new OutboxMessageData(
131+
Id: Guid.NewGuid(),
132+
Type: typeof(TestMessage).FullName!,
133+
Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions),
134+
TraceParent: null,
135+
TraceState: null,
136+
Destination: OutboxDestination.Send,
137+
Metadata: null);
138+
139+
// Act
140+
var exception = await Should.ThrowAsync<InvalidOperationException>(
141+
() => dispatcher.DispatchAsync(message, CancellationToken));
142+
143+
// Assert
144+
exception.Message.ShouldContain("missing its destination address");
145+
GetMock<ITransportSendEndpointProvider>().Verify(
146+
provider => provider.GetSendEndpointAsync(It.IsAny<Uri>(), It.IsAny<CancellationToken>()),
147+
Times.Never);
148+
}
149+
150+
private sealed record TestMessage(string Value);
29151
}

‎tests/Vulthil.Messaging.Outbox.Tests/TransactionalConsumeFilterTests.cs‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
using Vulthil.Messaging.Abstractions.Consumers;
2+
using Vulthil.Messaging.Transport;
3+
using Vulthil.SharedKernel.Application.Data;
4+
using Vulthil.xUnit;
5+
6+
namespace Vulthil.Messaging.Outbox.Tests;
7+
8+
public sealed class TransactionalConsumeFilterTests : BaseUnitTestCase
9+
{
10+
[Fact]
11+
public async Task ConsumeAsyncRunsTheNextDelegateInsideUnitOfWorkExecuteInTransactionAsync()
12+
{
13+
// Arrange
14+
SetUpUnitOfWorkToInvokeTheOperation();
15+
var filter = CreateInstance<TransactionalConsumeFilter<TestMessage>>();
16+
var context = NewContext();
17+
var nextInvoked = false;
18+
ConsumeDelegate<TestMessage> next = receivedContext =>
19+
{
20+
nextInvoked = ReferenceEquals(receivedContext, context);
21+
return Task.CompletedTask;
22+
};
23+
24+
// Act
25+
await filter.ConsumeAsync(context, next);
26+
27+
// Assert
28+
nextInvoked.ShouldBeTrue();
29+
GetMock<IUnitOfWork>().Verify(
30+
unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny<Func<CancellationToken, Task<bool>>>(), context.CancellationToken),
31+
Times.Once);
32+
}
33+
34+
[Fact]
35+
public async Task ConsumeAsyncNeverRunsTheNextDelegateWhenTheUnitOfWorkDoesNotInvokeTheOperation()
36+
{
37+
// Arrange
38+
GetMock<IUnitOfWork>()
39+
.Setup(unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny<Func<CancellationToken, Task<bool>>>(), It.IsAny<CancellationToken>()))
40+
.ReturnsAsync(false);
41+
var filter = CreateInstance<TransactionalConsumeFilter<TestMessage>>();
42+
var context = NewContext();
43+
var nextInvoked = false;
44+
ConsumeDelegate<TestMessage> next = _ =>
45+
{
46+
nextInvoked = true;
47+
return Task.CompletedTask;
48+
};
49+
50+
// Act
51+
await filter.ConsumeAsync(context, next);
52+
53+
// Assert
54+
nextInvoked.ShouldBeFalse();
55+
}
56+
57+
[Fact]
58+
public async Task ConsumeAsyncPropagatesAnExceptionThrownByTheNextDelegate()
59+
{
60+
// Arrange
61+
SetUpUnitOfWorkToInvokeTheOperation();
62+
var filter = CreateInstance<TransactionalConsumeFilter<TestMessage>>();
63+
var context = NewContext();
64+
ConsumeDelegate<TestMessage> next = _ => throw new InvalidOperationException("handler failed");
65+
66+
// Act & Assert
67+
await Should.ThrowAsync<InvalidOperationException>(() => filter.ConsumeAsync(context, next));
68+
}
69+
70+
private void SetUpUnitOfWorkToInvokeTheOperation() =>
71+
GetMock<IUnitOfWork>()
72+
.Setup(unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny<Func<CancellationToken, Task<bool>>>(), It.IsAny<CancellationToken>()))
73+
.Returns((Func<CancellationToken, Task<bool>> operation, CancellationToken token) => operation(token));
74+
75+
private static MessageContext<TestMessage> NewContext() => new()
76+
{
77+
Message = new TestMessage("hello"),
78+
RoutingKey = string.Empty,
79+
Headers = new Dictionary<string, object?>(),
80+
CancellationToken = CancellationToken,
81+
};
82+
83+
private sealed record TestMessage(string Value);
84+
}

‎tests/Vulthil.Messaging.Outbox.Tests/TransactionalPublishFilterTests.cs‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
using System.Diagnostics;
12
using System.Text.Json;
23
using System.Transactions;
34
using Microsoft.Extensions.Options;
@@ -88,6 +89,73 @@ public async Task PublishingInsideAnEntityFrameworkTransactionCapturesTheMessage
8889
GetMock<IOutboxStore>().Verify(store => store.SaveChangesAsync(It.IsAny<CancellationToken>()), Times.Once);
8990
}
9091

92+
[Fact]
93+
public async Task EnableTracingCapturesTheCurrentActivityOntoTheCapturedRow()
94+
{
95+
// Arrange
96+
GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(true);
97+
Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions { EnableTracing = true }));
98+
Use(TimeProvider.System);
99+
GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions());
100+
var filter = CreateInstance<TransactionalPublishFilter>();
101+
var context = NewContext();
102+
OutboxMessage? captured = null;
103+
GetMock<IOutboxStore>()
104+
.Setup(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()))
105+
.Callback<OutboxMessage>(message => captured = message);
106+
using var activity = new Activity("test-publish");
107+
activity.TraceStateString = "congo=t61rcWkgMzE";
108+
109+
// Act
110+
activity.Start();
111+
try
112+
{
113+
await filter.PublishAsync(context, _ => Task.CompletedTask);
114+
}
115+
finally
116+
{
117+
activity.Stop();
118+
}
119+
120+
// Assert
121+
captured.ShouldNotBeNull();
122+
captured.TraceParent.ShouldBe(activity.Id);
123+
captured.TraceState.ShouldBe(activity.TraceStateString);
124+
}
125+
126+
[Fact]
127+
public async Task EnableTracingDisabledLeavesTraceFieldsNullEvenWithAnActiveActivity()
128+
{
129+
// Arrange
130+
GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(true);
131+
Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions { EnableTracing = false }));
132+
Use(TimeProvider.System);
133+
GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions());
134+
var filter = CreateInstance<TransactionalPublishFilter>();
135+
var context = NewContext();
136+
OutboxMessage? captured = null;
137+
GetMock<IOutboxStore>()
138+
.Setup(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()))
139+
.Callback<OutboxMessage>(message => captured = message);
140+
using var activity = new Activity("test-publish");
141+
142+
// Act
143+
activity.Start();
144+
try
145+
{
146+
await filter.PublishAsync(context, _ => Task.CompletedTask);
147+
}
148+
finally
149+
{
150+
activity.Stop();
151+
}
152+
153+
// Assert
154+
captured.ShouldNotBeNull();
155+
captured.TraceParent.ShouldBeNull();
156+
captured.TraceState.ShouldBeNull();
157+
}
158+
91159
private static PublishFilterContext NewContext() => new()
92160
{
93161
Message = new TestMessage("hello"),

0 commit comments

Comments
 (0)