Fetch the connection string from the connection options

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<Commands.BookRoom> Seed(ICommandService<BookingState> service) {
-        var cmd = Helpers.GetBookRoom();
-        await service.Handle(cmd, default);
-
-        return cmd;
-    }
-
     protected ServiceTestBase() {
         _listener = new();
         TypeMap.RegisterKnownEventTypes(typeof(RoomBooked).Assembly);

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);

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);
         }

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);
 }

src/EventStore/test/Eventuous.Tests.EventStore/Subscriptions/SubscribeTests.cs

@@ -12,16 +12,19 @@ public class SubscribeToAll()
         new CatchUpSubscriptionFixture<AllStreamSubscription, AllStreamSubscriptionOptions, TestEventHandler>(_ => { }, 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);
     }

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";
     }

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;
+
+/// <summary>
+/// Tests for SqlServerSubscriptionBase connection string handling.
+/// These tests verify the fix for issue #410 where connection strings were not properly resolved.
+/// </summary>
+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<ArgumentException>();
+    }
+
+    [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<ArgumentException>();
+    }
+
+    [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<ArgumentException>();
+    }
+
+    [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<ArgumentException>();
+    }
+}
+
+/// <summary>
+/// Test implementation of SqlServerSubscriptionBase for testing connection string handling
+/// </summary>
+public class TestSubscription(
+        TestSubscriptionOptions     options,
+        ICheckpointStore            checkpointStore,
+        ConsumePipe                 consumePipe,
+        ILoggerFactory              loggerFactory,
+        SqlServerConnectionOptions? connectionOptions
+    )
+    : SqlServerSubscriptionBase<TestSubscriptionOptions>(
+        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;
+    }
+}
+
+/// <summary>
+/// Test subscription options for testing connection string handling
+/// </summary>
+public record TestSubscriptionOptions : SqlServerSubscriptionBaseOptions;