-
-
Notifications
You must be signed in to change notification settings - Fork 506
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Improved Postgres Checkpoint store, to make it fully idempotent
Added tests for that
- Loading branch information
1 parent
7f452fa
commit 345e841
Showing
6 changed files
with
189 additions
and
26 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
27 changes: 27 additions & 0 deletions
27
Core.EventStoreDB.Tests/Subscriptions/Checkpoints/PostgresContainerFixture.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| using Npgsql; | ||
| using Testcontainers.PostgreSql; | ||
| using Xunit; | ||
|
|
||
| namespace Core.EventStoreDB.Tests.Subscriptions.Checkpoints; | ||
|
|
||
| public class PostgresContainerFixture: IAsyncLifetime | ||
| { | ||
| private readonly PostgreSqlContainer container = new PostgreSqlBuilder() | ||
| .WithReuse(true) | ||
| .Build(); | ||
|
|
||
| public NpgsqlDataSource DataSource { get; private set; } = default!; | ||
|
|
||
| public async Task InitializeAsync() | ||
| { | ||
| await container.StartAsync(); | ||
| DataSource = new NpgsqlDataSourceBuilder(container.GetConnectionString()).Build(); | ||
| } | ||
|
|
||
| public async Task DisposeAsync() | ||
| { | ||
| await DataSource.DisposeAsync(); | ||
| await container.StopAsync(); | ||
| } | ||
|
|
||
| } |
69 changes: 69 additions & 0 deletions
69
...tStoreDB.Tests/Subscriptions/Checkpoints/PostgresSubscriptionCheckpointRepositoryTests.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| using Core.EventStoreDB.Subscriptions.Checkpoints; | ||
| using Npgsql; | ||
| using Xunit; | ||
|
|
||
| namespace Core.EventStoreDB.Tests.Subscriptions.Checkpoints; | ||
| using static ISubscriptionCheckpointRepository; | ||
|
|
||
| public class PostgresSubscriptionCheckpointRepositoryTests(PostgresContainerFixture fixture) | ||
| : IClassFixture<PostgresContainerFixture> | ||
| { | ||
| private readonly string subscriptionId = Guid.NewGuid().ToString("N"); | ||
|
|
||
| private readonly Func<CancellationToken, ValueTask<NpgsqlConnection>> connectionFactory = | ||
| _ => | ||
| { | ||
| var connection = fixture.DataSource.CreateConnection(); | ||
| connection.Open(); | ||
| return ValueTask.FromResult(connection); | ||
| }; | ||
|
|
||
| [Fact] | ||
| public async Task Store_InitialInsert_Success() | ||
| { | ||
| var checkpointTableCreator = new PostgresSubscriptionCheckpointSetup(fixture.DataSource); | ||
| var repository = new PostgresSubscriptionCheckpointRepository(connectionFactory, checkpointTableCreator); | ||
|
|
||
| var result = await repository.Store(subscriptionId, 1, Checkpoint.None, CancellationToken.None); | ||
|
|
||
| Assert.IsType<StoreResult.Success>(result); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task Store_UpdatePosition_Success() | ||
| { | ||
| var checkpointTableCreator = new PostgresSubscriptionCheckpointSetup(fixture.DataSource); | ||
| var repository = new PostgresSubscriptionCheckpointRepository(connectionFactory, checkpointTableCreator); | ||
|
|
||
| await repository.Store(subscriptionId, 1, Checkpoint.None, CancellationToken.None); | ||
| var result = await repository.Store(subscriptionId, 2, Checkpoint.From(1), CancellationToken.None); | ||
|
|
||
| Assert.IsType<StoreResult.Success>(result); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task Store_IdempotentCheck_ReturnsZero() | ||
| { | ||
| var checkpointTableCreator = new PostgresSubscriptionCheckpointSetup(fixture.DataSource); | ||
| var repository = new PostgresSubscriptionCheckpointRepository(connectionFactory, checkpointTableCreator); | ||
|
|
||
| await repository.Store(subscriptionId, 1, Checkpoint.None, CancellationToken.None); | ||
| await repository.Store(subscriptionId, 2, Checkpoint.From(1), CancellationToken.None); | ||
| var result = await repository.Store(subscriptionId, 2, Checkpoint.From(1), CancellationToken.None); | ||
|
|
||
| Assert.IsType<StoreResult.Ignored>(result); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task Store_InvalidUpdate_Failure() | ||
| { | ||
| var checkpointTableCreator = new PostgresSubscriptionCheckpointSetup(fixture.DataSource); | ||
| var repository = new PostgresSubscriptionCheckpointRepository(connectionFactory, checkpointTableCreator); | ||
|
|
||
| await repository.Store(subscriptionId, 1, Checkpoint.None, CancellationToken.None); | ||
| await repository.Store(subscriptionId, 2, Checkpoint.From(1), CancellationToken.None); | ||
| var result = await repository.Store(subscriptionId, 1, Checkpoint.From(3), CancellationToken.None); | ||
|
|
||
| Assert.IsType<StoreResult.Mismatch>(result); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters