EventSourcing.NetCore/Core.EventStoreDB.Tests/Subscriptions/Checkpoints/PostgresSubscriptionCheckpointRepositoryTests.cs at 0f0409cd6bf18b3a5e5022ca30c01e192052825e · oskardudycz/EventSourcing.NetCore · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
using Core.EventStoreDB.Subscriptions.Checkpoints;
using Core.EventStoreDB.Subscriptions.Checkpoints.Postgres;
using Core.Testing.Fixtures;
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.CreateVersion7().ToString("N");

    [Fact]
    public async Task Store_InitialInsert_Success()
    {
        var checkpointTableCreator = new PostgresSubscriptionCheckpointSetup(fixture.DataSource);
        await checkpointTableCreator.EnsureStoreExists(CancellationToken.None);
        await using var connection = fixture.DataSource.CreateConnection();
        var repository = new PostgresSubscriptionCheckpointRepository(connection);

        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);
        await checkpointTableCreator.EnsureStoreExists(CancellationToken.None);
        await using var connection = fixture.DataSource.CreateConnection();
        var repository = new PostgresSubscriptionCheckpointRepository(connection);

        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);
        await checkpointTableCreator.EnsureStoreExists(CancellationToken.None);
        await using var connection = fixture.DataSource.CreateConnection();
        var repository = new PostgresSubscriptionCheckpointRepository(connection);

        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);
        await checkpointTableCreator.EnsureStoreExists(CancellationToken.None);
        await using var connection = fixture.DataSource.CreateConnection();
        var repository = new PostgresSubscriptionCheckpointRepository(connection);

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