fix: testing aggregate 'Then' with identifier filters out the specifi…

…c aggregate
Informatievlaanderen · Apr 9, 2021 · 01d79f7 · 01d79f7
1 parent d5b2c7f
commit 01d79f7
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 4 deletions.
diff --git a/...nderen.Basisregisters.AggregateSource.Testing.SqlStreamStore/StreamStoreFactRepository.cs b/...nderen.Basisregisters.AggregateSource.Testing.SqlStreamStore/StreamStoreFactRepository.cs
@@ -3,7 +3,6 @@ namespace Be.Vlaanderen.Basisregisters.AggregateSource.Testing.SqlStreamStore
     using System;
     using System.Collections.Generic;
     using System.Linq;
-    using System.Reflection;
     using System.Threading.Tasks;
     using EventHandling;
     using global::SqlStreamStore;
@@ -31,7 +30,21 @@ public StreamStoreFactRepository(
         public async Task<Fact[]> RetrieveFacts(long fromPositionExclusive)
         {
             var results = new List<Fact>();
-            var page = await _streamStore.ReadAllForwards(fromPositionExclusive<0?Position.Start:fromPositionExclusive, 10);
+            var page = await _streamStore.ReadAllForwards(fromPositionExclusive < 0 ? Position.Start : fromPositionExclusive, 10);
+            results.AddRange(page.Messages.Where(m => m.Position != fromPositionExclusive).Select(MapToFact));
+            while (!page.IsEnd)
+            {
+                page = await page.ReadNext();
+                results.AddRange(page.Messages.Where(m => m.Position != fromPositionExclusive).Select(MapToFact));
+            }
+
+            return results.ToArray();
+        }
+
+        public async Task<Fact[]> RetrieveFactsByStream(long fromPositionExclusive, string aggregateIdentifier)
+        {
+            var results = new List<Fact>();
+            var page = await _streamStore.ReadStreamForwards(new StreamId(aggregateIdentifier), fromPositionExclusive < 0 ? Convert.ToInt32(Position.Start) : Convert.ToInt32(fromPositionExclusive), 10);
             results.AddRange(page.Messages.Where(m => m.Position != fromPositionExclusive).Select(MapToFact));
             while (!page.IsEnd)
             {
@@ -57,7 +70,7 @@ public async Task<long> PersistFacts(Fact[] facts)
                             _eventMapping.GetEventName(e.Event.GetType()),
                             _eventSerializer.SerializeObject(e.Event))).ToArray());
 
-            return result?.CurrentPosition??await _streamStore.ReadHeadPosition();
+            return result?.CurrentPosition ?? await _streamStore.ReadHeadPosition();
         }
 
         private Fact MapToFact(StreamMessage streamMessage)

diff --git a/....Vlaanderen.Basisregisters.AggregateSource.Testing/EventCentricTestSpecificationRunner.cs b/....Vlaanderen.Basisregisters.AggregateSource.Testing/EventCentricTestSpecificationRunner.cs
@@ -32,7 +32,10 @@ private async Task<EventCentricTestResult> RunAsync(EventCentricTestSpecificatio
             if (result.HasValue)
                 return spec.Fail(result.Value);
 
-            var actualEvents = await _factReader.RetrieveFacts(position);
+            var actualEvents =
+                spec.Thens.Length == 0
+                ? await _factReader.RetrieveFacts(position)
+                : await _factReader.RetrieveFactsByStream(position, spec.Thens.First().Identifier);
 
             return actualEvents.SequenceEqual(spec.Thens, new WrappedFactComparerEqualityComparer(_comparer))
                 ? spec.Pass(actualEvents)

diff --git a/src/Be.Vlaanderen.Basisregisters.AggregateSource.Testing/IFactReader.cs b/src/Be.Vlaanderen.Basisregisters.AggregateSource.Testing/IFactReader.cs
@@ -5,5 +5,6 @@ namespace Be.Vlaanderen.Basisregisters.AggregateSource.Testing
     public interface IFactReader
     {
         Task<Fact[]> RetrieveFacts(long fromPositionExclusive);
+        Task<Fact[]> RetrieveFactsByStream(long fromPositionExclusive, string aggregateIdentifier);
     }
 }