diff --git a/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.cs b/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.cs index 1be69aeae..c2d23d7d7 100644 --- a/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.cs +++ b/src/Core/test/Eventuous.Tests.Application/ServiceTestBase.cs @@ -1,4 +1,3 @@ -using Eventuous.Sut.App; using Eventuous.Sut.Domain; using Eventuous.TestHelpers.TUnit; using Eventuous.Testing; @@ -35,13 +34,6 @@ static ImportBooking CreateCommand() { return new("booking1", "room1", 100, today, today.PlusDays(1), "user"); } - static async Task Seed(ICommandService service) { - var cmd = Helpers.GetBookRoom(); - await service.Handle(cmd, default); - - return cmd; - } - protected ServiceTestBase() { _listener = new(); TypeMap.RegisterKnownEventTypes(typeof(RoomBooked).Assembly); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToAll.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToAll.cs index c9e8f7f18..81e496d22 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToAll.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToAll.cs @@ -59,7 +59,7 @@ protected async Task ShouldUseExistingCheckpoint(CancellationToken cancellationT await fixture.CheckpointStore.StoreCheckpoint(new(fixture.SubscriptionId, last), true, cancellationToken); var l = await fixture.CheckpointStore.GetLastCheckpoint(fixture.SubscriptionId, cancellationToken); - TestContext.Current?.OutputWriter.WriteLine("Last checkpoint: {0}", l.Position!); + WriteLine("Last checkpoint: {0}", l.Position!); await fixture.StartSubscription(); await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs index 70b916ee9..a1d373fe1 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToStream.cs @@ -34,13 +34,13 @@ protected async Task ShouldConsumeProducedEvents(CancellationToken cancellationT } protected async Task ShouldConsumeProducedEventsWhenRestarting(CancellationToken cancellationToken) { - TestContext.Current?.OutputWriter.WriteLine("Phase one"); + WriteLine("Phase one"); await TestConsumptionOfProducedEvents(); - TestContext.Current?.OutputWriter.WriteLine("Resetting handler"); + WriteLine("Resetting handler"); fixture.Handler.Reset(); - TestContext.Current?.OutputWriter.WriteLine("Phase two"); + WriteLine("Phase two"); await TestConsumptionOfProducedEvents(); var checkpoint = await fixture.CheckpointStore.GetLastCheckpoint(fixture.SubscriptionId, cancellationToken); @@ -51,13 +51,13 @@ protected async Task ShouldConsumeProducedEventsWhenRestarting(CancellationToken async Task TestConsumptionOfProducedEvents() { const int count = 10; - TestContext.Current?.OutputWriter.WriteLine("Generating and producing events"); + WriteLine("Generating and producing events"); var testEvents = await GenerateAndProduceEvents(count); - TestContext.Current?.OutputWriter.WriteLine("Starting subscription"); + WriteLine("Starting subscription"); await fixture.StartSubscription(); - await fixture.Handler.AssertCollection(TimeSpan.FromSeconds(2), [..testEvents]).Validate(); - TestContext.Current?.OutputWriter.WriteLine("Stopping subscription"); + await fixture.Handler.AssertCollection(TimeSpan.FromSeconds(2), [..testEvents]).Validate(cancellationToken); + WriteLine("Stopping subscription"); await fixture.StopSubscription(); await Assert.That(fixture.Handler.Count).IsEqualTo(10); } diff --git a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionTestBase.cs b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionTestBase.cs index 2d0fe3894..828aaa09c 100644 --- a/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionTestBase.cs +++ b/src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscriptionTestBase.cs @@ -5,11 +5,19 @@ namespace Eventuous.Tests.Subscriptions.Base; public abstract class SubscriptionTestBase(IStartableFixture fixture) { [Before(Test)] public async Task Startup() { + WriteLine("Starting the fixture"); await fixture.InitializeAsync(); + WriteLine("Fixture started"); } [After(Test)] public async Task Shutdown() { + WriteLine("Stopping the fixture"); await fixture.DisposeAsync(); + WriteLine("Fixture stopped"); } + + protected static void WriteLine(string message) => TestContext.Current?.OutputWriter.WriteLine(message); + + protected static void WriteLine(string message, params object?[] args) => TestContext.Current?.OutputWriter.WriteLine(message, args); } diff --git a/src/EventStore/test/Eventuous.Tests.EventStore/Subscriptions/SubscribeTests.cs b/src/EventStore/test/Eventuous.Tests.EventStore/Subscriptions/SubscribeTests.cs index 2f66b18dd..c939ac27e 100644 --- a/src/EventStore/test/Eventuous.Tests.EventStore/Subscriptions/SubscribeTests.cs +++ b/src/EventStore/test/Eventuous.Tests.EventStore/Subscriptions/SubscribeTests.cs @@ -12,16 +12,19 @@ public class SubscribeToAll() new CatchUpSubscriptionFixture(_ => { }, new("$all"), false) ) { [Test] + [Retry(3)] public async Task Esdb_ShouldConsumeProducedEvents(CancellationToken cancellationToken) { await ShouldConsumeProducedEvents(cancellationToken); } [Test] + [Retry(3)] public async Task Esdb_ShouldConsumeProducedEventsWhenRestarting(CancellationToken cancellationToken) { await ShouldConsumeProducedEventsWhenRestarting(cancellationToken); } [Test] + [Retry(3)] public async Task Esdb_ShouldUseExistingCheckpoint(CancellationToken cancellationToken) { await ShouldUseExistingCheckpoint(cancellationToken); } diff --git a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs index 00c03f3ea..ac188a676 100644 --- a/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs +++ b/src/SqlServer/src/Eventuous.SqlServer/Subscriptions/SqlServerSubscriptionBase.cs @@ -29,7 +29,8 @@ protected SqlServerSubscriptionBase( ? connectionOptions.Schema : options.Schema ); - _connectionString = Ensure.NotEmptyString(Options.ConnectionString); + var connectionString = connectionOptions?.ConnectionString ?? options.ConnectionString; + _connectionString = Ensure.NotEmptyString(connectionString); GetEndOfStream = $"SELECT MAX(StreamPosition) FROM {options.Schema}.Messages"; GetEndOfAll = $"SELECT MAX(GlobalPosition) FROM {options.Schema}.Messages"; } diff --git a/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/ConnectionStringTests.cs b/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/ConnectionStringTests.cs new file mode 100644 index 000000000..19c50cf42 --- /dev/null +++ b/src/SqlServer/test/Eventuous.Tests.SqlServer/Subscriptions/ConnectionStringTests.cs @@ -0,0 +1,160 @@ +using Eventuous.SqlServer.Projections; +using Eventuous.SqlServer.Subscriptions; +using Eventuous.Subscriptions; +using Eventuous.Subscriptions.Checkpoints; +using Eventuous.Subscriptions.Filters; +using Microsoft.Data.SqlClient; +using System.Data; + +namespace Eventuous.Tests.SqlServer.Subscriptions; + +/// +/// Tests for SqlServerSubscriptionBase connection string handling. +/// These tests verify the fix for issue #410 where connection strings were not properly resolved. +/// +public class ConnectionStringTests { + readonly ICheckpointStore _checkpointStore = new NoOpCheckpointStore(); + readonly ConsumePipe _consumePipe = new(); + readonly ILoggerFactory _loggerFactory = LoggerFactory.Create(builder => builder.AddConsole()); + + [Test] + public async Task Should_Use_ConnectionOptions_ConnectionString_When_Provided() { + // Arrange + const string expectedConnectionString = "Server=localhost;Database=Test1;Trusted_Connection=true;"; + const string optionsConnectionString = "Server=localhost;Database=Test2;Trusted_Connection=true;"; + + var options = new TestSubscriptionOptions { + SubscriptionId = "test-subscription-1", + ConnectionString = optionsConnectionString + }; + + var connectionOptions = new SqlServerConnectionOptions(expectedConnectionString, "dbo"); + + // Act & Assert - The constructor should not throw an exception + var subscription = new TestSubscription(options, _checkpointStore, _consumePipe, _loggerFactory, connectionOptions); + + await Assert.That(subscription.IsInitialized).IsTrue(); + } + + [Test] + public async Task Should_Use_Options_ConnectionString_When_ConnectionOptions_Is_Null() { + // Arrange + const string expectedConnectionString = "Server=localhost;Database=Test;Trusted_Connection=true;"; + + var options = new TestSubscriptionOptions { + SubscriptionId = "test-subscription-2", + ConnectionString = expectedConnectionString + }; + + // Act & Assert - The constructor should not throw an exception + var subscription = new TestSubscription(options, _checkpointStore, _consumePipe, _loggerFactory, null); + + await Assert.That(subscription.IsInitialized).IsTrue(); + } + + [Test] + public async Task Should_Use_Options_ConnectionString_When_ConnectionOptions_ConnectionString_Is_Null() { + // Arrange + const string expectedConnectionString = "Server=localhost;Database=Test;Trusted_Connection=true;"; + + var options = new TestSubscriptionOptions { + SubscriptionId = "test-subscription-3", + ConnectionString = expectedConnectionString + }; + + var connectionOptions = new SqlServerConnectionOptions(null!, "dbo"); + + // Act & Assert - The constructor should not throw an exception + var subscription = new TestSubscription(options, _checkpointStore, _consumePipe, _loggerFactory, connectionOptions); + + await Assert.That(subscription.IsInitialized).IsTrue(); + } + + [Test] + public async Task Should_Throw_When_Both_Connection_Strings_Are_Null_Or_Empty() { + // Arrange + var options = new TestSubscriptionOptions { + SubscriptionId = "test-subscription-4", + ConnectionString = null + }; + + var connectionOptions = new SqlServerConnectionOptions(null!, "dbo"); + + // Act & Assert + await Assert.That(() => new TestSubscription(options, _checkpointStore, _consumePipe, _loggerFactory, connectionOptions)).Throws(); + } + + [Test] + public async Task Should_Throw_When_Both_Connection_Strings_Are_Empty() { + // Arrange + var options = new TestSubscriptionOptions { + SubscriptionId = "test-subscription-5", + ConnectionString = "" + }; + + var connectionOptions = new SqlServerConnectionOptions("", "dbo"); + + // Act & Assert + await Assert.That(() => new TestSubscription(options, _checkpointStore, _consumePipe, _loggerFactory, connectionOptions)).Throws(); + } + + [Test] + public async Task Should_Throw_When_Options_Connection_String_Is_Null_And_No_ConnectionOptions() { + // Arrange + var options = new TestSubscriptionOptions { + SubscriptionId = "test-subscription-6", + ConnectionString = null + }; + + // Act & Assert + await Assert.That(() => new TestSubscription(options, _checkpointStore, _consumePipe, _loggerFactory, null)).Throws(); + } + + [Test] + public async Task Should_Throw_When_Options_Connection_String_Is_Empty_And_No_ConnectionOptions() { + // Arrange + var options = new TestSubscriptionOptions { + SubscriptionId = "test-subscription-7", + ConnectionString = "" + }; + + // Act & Assert + await Assert.That(() => new TestSubscription(options, _checkpointStore, _consumePipe, _loggerFactory, null)).Throws(); + } +} + +/// +/// Test implementation of SqlServerSubscriptionBase for testing connection string handling +/// +public class TestSubscription( + TestSubscriptionOptions options, + ICheckpointStore checkpointStore, + ConsumePipe consumePipe, + ILoggerFactory loggerFactory, + SqlServerConnectionOptions? connectionOptions + ) + : SqlServerSubscriptionBase( + options, + checkpointStore, + consumePipe, + SubscriptionKind.All, + loggerFactory, + null, + null, + connectionOptions + ) { + public bool IsInitialized { get; } = true; + + protected override SqlCommand PrepareCommand(SqlConnection connection, long start) { + var command = connection.CreateCommand(); + command.CommandType = CommandType.Text; + command.CommandText = "SELECT TOP 0 MessageId, MessageType, StreamPosition, GlobalPosition, JsonData, JsonMetadata, Created, StreamName FROM dbo.Messages WHERE GlobalPosition > @start"; + command.Parameters.Add("@start", SqlDbType.BigInt).Value = start; + return command; + } +} + +/// +/// Test subscription options for testing connection string handling +/// +public record TestSubscriptionOptions : SqlServerSubscriptionBaseOptions;