tsai
/
squidex
mirror of https://github.com/Squidex/squidex.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Code cleanup.

pull/733/head
Sebastian 5 years ago
parent
4606e50d5a
commit
e70063df33
1 changed files with 46 additions and 59 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 105
      backend/tests/Squidex.Infrastructure.Tests/EventSourcing/Grains/EventConsumerGrainTests.cs

105
backend/tests/Squidex.Infrastructure.Tests/EventSourcing/Grains/EventConsumerGrainTests.cs
View File

@ -58,10 +58,11 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
private readonly IGrainState<EventConsumerState> grainState = A.Fake<IGrainState<EventConsumerState>>(); private readonly IGrainState<EventConsumerState> grainState = A.Fake<IGrainState<EventConsumerState>>();
private readonly IEventConsumer eventConsumer = A.Fake<IEventConsumer>(); private readonly IEventConsumer eventConsumer = A.Fake<IEventConsumer>();
private readonly IEventDataFormatter formatter = A.Fake<IEventDataFormatter>();
private readonly IEventStore eventStore = A.Fake<IEventStore>(); private readonly IEventStore eventStore = A.Fake<IEventStore>();
private readonly IEventSubscription eventSubscription = A.Fake<IEventSubscription>(); private readonly IEventSubscription eventSubscription = A.Fake<IEventSubscription>();
private readonly ISemanticLog log = A.Fake<ISemanticLog>(); private readonly ISemanticLog log = A.Fake<ISemanticLog>();
private readonly IEventDataFormatter formatter = A.Fake<IEventDataFormatter>(); private readonly StoredEvent storedEvent;
private readonly EventData eventData = new EventData("Type", new EnvelopeHeaders(), "Payload"); private readonly EventData eventData = new EventData("Type", new EnvelopeHeaders(), "Payload");
private readonly Envelope<IEvent> envelope = new Envelope<IEvent>(new MyEvent()); private readonly Envelope<IEvent> envelope = new Envelope<IEvent>(new MyEvent());
private readonly MyEventConsumerGrain sut; private readonly MyEventConsumerGrain sut;
@ -95,7 +96,9 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
A.CallTo(() => eventSubscription.Sender) A.CallTo(() => eventSubscription.Sender)
.Returns(eventSubscription); .Returns(eventSubscription);
A.CallTo(() => formatter.ParseIfKnown(A<StoredEvent>.That.Matches(x => x.Data == eventData))) storedEvent = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
A.CallTo(() => formatter.ParseIfKnown(storedEvent))
.Returns(envelope); .Returns(envelope);
sut = new MyEventConsumerGrain( sut = new MyEventConsumerGrain(
@ -116,7 +119,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = true, Position = initialPosition, Error = null }); AssetGrainState(isStopped: true, position: initialPosition);
A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._)) A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._))
.MustNotHaveHappened(); .MustNotHaveHappened();
@ -130,7 +133,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = initialPosition, Error = null }); AssetGrainState(isStopped: false, position: initialPosition);
A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._)) A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._))
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -146,7 +149,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = initialPosition, Error = null }); AssetGrainState(isStopped: false, position: initialPosition);
A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._)) A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._))
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -160,7 +163,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = initialPosition, Error = null }); AssetGrainState(isStopped: false, position: initialPosition);
A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._)) A.CallTo(() => eventStore.CreateSubscription(A<IEventSubscriber>._, A<string>._, A<string>._))
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -176,7 +179,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = true, Position = initialPosition, Error = null }); AssetGrainState(isStopped: true, position: initialPosition);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -195,7 +198,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = null, Error = null }); AssetGrainState(isStopped: false, position: null);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappened(2, Times.Exactly); .MustHaveHappened(2, Times.Exactly);
@ -216,16 +219,14 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
[Fact] [Fact]
public async Task Should_invoke_and_update_position_if_event_received() public async Task Should_invoke_and_update_position_if_event_received()
{ {
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = @event.EventPosition, Error = null, Count = 1 }); AssetGrainState(isStopped: false, position: storedEvent.EventPosition, count: 1);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -237,23 +238,21 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
[Fact] [Fact]
public async Task Should_invoke_and_update_position_if_event_received_one_by_one() public async Task Should_invoke_and_update_position_if_event_received_one_by_one()
{ {
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
A.CallTo(() => eventConsumer.BatchSize) A.CallTo(() => eventConsumer.BatchSize)
.Returns(1); .Returns(1);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = @event.EventPosition, Error = null, Count = 5 }); AssetGrainState(isStopped: false, position: storedEvent.EventPosition, count: 5);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappened(5, Times.Exactly); .MustHaveHappened(5, Times.Exactly);
@ -265,23 +264,21 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
[Fact] [Fact]
public async Task Should_invoke_and_update_position_if_event_received_batched() public async Task Should_invoke_and_update_position_if_event_received_batched()
{ {
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
A.CallTo(() => eventConsumer.BatchSize) A.CallTo(() => eventConsumer.BatchSize)
.Returns(100); .Returns(100);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = @event.EventPosition, Error = null, Count = 5 }); AssetGrainState(isStopped: false, position: storedEvent.EventPosition, count: 5);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -293,19 +290,17 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
[Fact] [Fact]
public async Task Should_not_invoke_but_update_position_if_consumer_does_not_want_to_handle() public async Task Should_not_invoke_but_update_position_if_consumer_does_not_want_to_handle()
{ {
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData); A.CallTo(() => eventConsumer.Handles(storedEvent))
A.CallTo(() => eventConsumer.Handles(@event))
.Returns(false); .Returns(false);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = @event.EventPosition, Error = null, Count = 0 }); AssetGrainState(isStopped: false, position: storedEvent.EventPosition);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -320,16 +315,14 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
A.CallTo(() => formatter.ParseIfKnown(A<StoredEvent>.That.Matches(x => x.Data == eventData))) A.CallTo(() => formatter.ParseIfKnown(A<StoredEvent>.That.Matches(x => x.Data == eventData)))
.Returns(null); .Returns(null);
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = @event.EventPosition, Error = null, Count = 0 }); AssetGrainState(isStopped: false, position: storedEvent.EventPosition);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -341,16 +334,14 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
[Fact] [Fact]
public async Task Should_not_invoke_and_update_position_if_event_is_from_another_subscription() public async Task Should_not_invoke_and_update_position_if_event_is_from_another_subscription()
{ {
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(A.Fake<IEventSubscription>(), @event); await OnEventAsync(A.Fake<IEventSubscription>(), storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = initialPosition, Error = null }); AssetGrainState(isStopped: false, position: initialPosition);
A.CallTo(() => eventConsumer.On(envelope)) A.CallTo(() => eventConsumer.On(envelope))
.MustNotHaveHappened(); .MustNotHaveHappened();
@ -368,7 +359,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = true, Position = initialPosition, Error = ex.ToString() }); AssetGrainState(isStopped: true, position: initialPosition, error: ex.ToString());
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -389,7 +380,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = initialPosition, Error = null }); AssetGrainState(isStopped: false, position: initialPosition);
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustNotHaveHappened(); .MustNotHaveHappened();
@ -422,7 +413,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = true, Position = initialPosition, Error = ex.ToString() }); AssetGrainState(isStopped: true, position: initialPosition, error: ex.ToString());
A.CallTo(() => grainState.WriteAsync()) A.CallTo(() => grainState.WriteAsync())
.MustHaveHappenedOnceExactly(); .MustHaveHappenedOnceExactly();
@ -439,16 +430,14 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
A.CallTo(() => eventConsumer.On(envelope)) A.CallTo(() => eventConsumer.On(envelope))
.Throws(ex); .Throws(ex);
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = true, Position = initialPosition, Error = ex.ToString() }); AssetGrainState(isStopped: true, position: initialPosition, error: ex.ToString());
A.CallTo(() => eventConsumer.On(envelope)) A.CallTo(() => eventConsumer.On(envelope))
.MustHaveHappened(); .MustHaveHappened();
@ -468,16 +457,14 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
A.CallTo(() => formatter.ParseIfKnown(A<StoredEvent>.That.Matches(x => x.Data == eventData))) A.CallTo(() => formatter.ParseIfKnown(A<StoredEvent>.That.Matches(x => x.Data == eventData)))
.Throws(ex); .Throws(ex);
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
AssetGrainState(new EventConsumerState { IsStopped = true, Position = initialPosition, Error = ex.ToString() }); AssetGrainState(isStopped: true, position: initialPosition, error: ex.ToString());
A.CallTo(() => eventConsumer.On(envelope)) A.CallTo(() => eventConsumer.On(envelope))
.MustNotHaveHappened(); .MustNotHaveHappened();
@ -497,12 +484,10 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
A.CallTo(() => eventConsumer.On(envelope)) A.CallTo(() => eventConsumer.On(envelope))
.Throws(exception); .Throws(exception);
var @event = new StoredEvent("Stream", Guid.NewGuid().ToString(), 123, eventData);
await sut.ActivateAsync(consumerName); await sut.ActivateAsync(consumerName);
await sut.ActivateAsync(); await sut.ActivateAsync();
await OnEventAsync(eventSubscription, @event); await OnEventAsync(eventSubscription, storedEvent);
await sut.CompleteAsync(); await sut.CompleteAsync();
@ -510,7 +495,7 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
await sut.StartAsync(); await sut.StartAsync();
await sut.StartAsync(); await sut.StartAsync();
AssetGrainState(new EventConsumerState { IsStopped = false, Position = initialPosition, Error = null }); AssetGrainState(isStopped: false, position: initialPosition);
A.CallTo(() => eventConsumer.On(envelope)) A.CallTo(() => eventConsumer.On(envelope))
.MustHaveHappened(); .MustHaveHappened();
@ -535,9 +520,11 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
return sut.OnEventAsync(subscription, ev); return sut.OnEventAsync(subscription, ev);
} }
private void AssetGrainState(EventConsumerState state) private void AssetGrainState(bool isStopped = false, string? position = null, string? error = null, int count = 0)
{ {
grainState.Value.Should().BeEquivalentTo(state); var expected = new EventConsumerState { IsStopped = isStopped, Position = position, Error = error, Count = count };
grainState.Value.Should().BeEquivalentTo(expected);
} }
} }
} }

Write Preview
Loading…
Cancel
Save