-
-
Notifications
You must be signed in to change notification settings - Fork 506
/
StreamLoading.cs
100 lines (80 loc) · 2.79 KB
/
StreamLoading.cs
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
using FluentAssertions;
using Marten.Integration.Tests.TestsInfrastructure;
using Xunit;
namespace Marten.Integration.Tests.EventStore.Stream;
public class StreamLoading(MartenFixture fixture): MartenTest(fixture.PostgreSqlContainer)
{
public record IssueCreated(
Guid IssueId,
string Description
);
public record IssueUpdated(
Guid IssueId,
string Description
);
private readonly Guid issueId = Guid.NewGuid();
private Guid GetExistingStreamId()
{
var @event = new IssueCreated(issueId, "Description");
var streamId = EventStore.StartStream(@event).Id;
Session.SaveChanges();
return streamId;
}
[Fact]
public void GivenExistingStream_WithOneEventWhenStreamIsLoaded_ThenItLoadsOneEvent()
{
//Given
var streamId = GetExistingStreamId();
//When
var events = EventStore.FetchStream(streamId);
//Then
events.Count.Should().Be(1);
events.First().Version.Should().Be(1);
}
[Fact]
public void GivenExistingStreamWithOneEvent_WhenEventIsAppended_ThenItLoadsTwoEvents()
{
//Given
var streamId = GetExistingStreamId();
//When
EventStore.Append(streamId, new IssueUpdated(issueId, "New Description"));
Session.SaveChanges();
//Then
var events = EventStore.FetchStream(streamId);
events.Count.Should().Be(2);
events.Last().Version.Should().Be(2);
}
[Fact]
public void GivenExistingStreamWithOneEvent_WhenStreamIsLoadedByEventType_ThenItLoadsOneEvent()
{
//Given
var streamId = GetExistingStreamId();
var eventId = EventStore.FetchStream(streamId).Single().Id;
//When
var @event = EventStore.Load<IssueCreated>(eventId);
//Then
@event.Should().NotBeNull();
@event!.Id.Should().Be(eventId);
}
[Fact]
public async Task
GivenExistingStreamWithMultipleEvents_WhenEventsAreQueriedOrderedDescending_ThenLastEventIsLoaded()
{
var streamId = Guid.NewGuid();
Session.Events.Append(streamId,
new IssueCreated(streamId, "Description"),
new IssueUpdated(streamId, "Description"),
new IssueUpdated(streamId, "The Last One")
);
await Session.SaveChangesAsync();
var lastEvent = Session.Events.QueryAllRawEvents()
.Where(e => e.StreamId == streamId)
.OrderByDescending(e => e.Version)
.FirstOrDefault();
lastEvent.Should().NotBeNull();
lastEvent!.Data.Should().BeOfType<IssueUpdated>();
var lastUpdatedEvent = (IssueUpdated)lastEvent.Data;
lastUpdatedEvent.IssueId.Should().Be(streamId);
lastUpdatedEvent.Description.Should().Be("The Last One");
}
}