Packages

commanded

0.16.0-rc.0
1.4.10 1.4.9 1.4.8 1.4.7 1.4.6 1.4.3 1.4.2 1.4.1 1.4.0 1.4.0-rc.0 1.3.1 1.3.0 1.2.0 1.1.1 1.1.0 1.0.1 1.0.0 1.0.0-rc.1 1.0.0-rc.0 0.19.1 0.19.0 0.18.1 0.18.0 0.17.5 0.17.4 0.17.3 0.17.2 0.17.1 0.17.0 0.16.0 0.16.0-rc.1 0.16.0-rc.0 0.15.1 0.15.0 0.14.0 0.14.0-rc.0 0.13.0 0.12.0 0.11.0 0.10.0 0.9.0 0.8.5 0.8.4 0.8.3 0.8.1 0.8.0 0.7.1 0.6.2 0.6.1 0.6.0 0.4.0 0.3.1 0.3.0 0.2.1 0.2.0 0.1.0

Use Commanded to build your own Elixir applications following the CQRS/ES pattern.

HexDocs

Current section

Files

Jump to
Readme
56 Versions
3 Dependencies
Dependants
Files
Activity
commanded test aggregates event_persistence_test.exs
Raw

test/aggregates/event_persistence_test.exs 

defmodule Commanded.Aggregates.EventPersistenceTest do

  use Commanded.StorageCase



  import Commanded.Enumerable, only: [pluck: 2]



  alias Commanded.Aggregates.{Aggregate,AppendItemsHandler,ExampleAggregate}

  alias Commanded.Aggregates.ExampleAggregate.Commands.{AppendItems,NoOp}

  alias Commanded.EventStore

  alias Commanded.Helpers.ProcessHelper

  alias Commanded.Aggregates.ExecutionContext



  test "should persist pending events in order applied" do

    aggregate_uuid = UUID.uuid4



    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)



    {:ok, 10, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})

    assert length(events) == 10



    recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()



    assert recorded_events |> pluck(:data) |> pluck(:index) == Enum.to_list(1..10)

    assert pluck(recorded_events, :event_number) == Enum.to_list(1..10)



    Enum.each(recorded_events, fn recorded_event ->

      assert recorded_event.stream_id == aggregate_uuid

    end)

  end



  test "should not persist events when command returns no events" do

    aggregate_uuid = UUID.uuid4



    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)



    {:ok, 1, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle})

    assert length(events) == 1



    {:ok, 1, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %NoOp{}, handler: ExampleAggregate, function: :noop})

    assert length(events) == 0



    recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()

    assert length(recorded_events) == 1

  end



  test "should persist event metadata" do

    aggregate_uuid = UUID.uuid4



    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)



    metadata = %{"ip_address" => "127.0.0.1"}

    context = %ExecutionContext{command: %AppendItems{count: 10}, metadata: metadata, handler: AppendItemsHandler, function: :handle}



    {:ok, 10, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, context)

    assert length(events) == 10



    recorded_events = EventStore.stream_forward(aggregate_uuid, 0) |> Enum.to_list()



    Enum.each(recorded_events, fn recorded_event ->

      assert recorded_event.metadata == metadata

    end)

  end



  test "should reload persisted events when restarting aggregate process" do

    aggregate_uuid = UUID.uuid4



    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)



    {:ok, 10, events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 10}, handler: AppendItemsHandler, function: :handle})

    assert length(events) == 10



    ProcessHelper.shutdown_aggregate(ExampleAggregate, aggregate_uuid)



    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)



    assert Aggregate.aggregate_version(ExampleAggregate, aggregate_uuid) == 10

    assert Aggregate.aggregate_state(ExampleAggregate, aggregate_uuid) == %ExampleAggregate{

      items: 1..10 |> Enum.to_list(),

      last_index: 10,

    }

  end



  test "should reload persisted events in batches when restarting aggregate process" do

    aggregate_uuid = UUID.uuid4



    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)



    {:ok, 100, _events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})

    {:ok, 200, _events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 100}, handler: AppendItemsHandler, function: :handle})

    {:ok, 201, _events} = Aggregate.execute(ExampleAggregate, aggregate_uuid, %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle})



    ProcessHelper.shutdown_aggregate(ExampleAggregate, aggregate_uuid)



    {:ok, ^aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, aggregate_uuid)



    assert Aggregate.aggregate_version(ExampleAggregate, aggregate_uuid) == 201

    assert Aggregate.aggregate_state(ExampleAggregate, aggregate_uuid) == %ExampleAggregate{

      items: 1..201 |> Enum.to_list,

      last_index: 201,

    }

  end



  test "should prefix stream UUID with aggregate indentity prefix" do

    aggregate_uuid = UUID.uuid4()

    prefix = "example-prefix-"

    prefixed_aggregate_uuid = prefix <> aggregate_uuid



    {:ok, ^prefixed_aggregate_uuid} = Commanded.Aggregates.Supervisor.open_aggregate(ExampleAggregate, prefixed_aggregate_uuid)



    context = %ExecutionContext{command: %AppendItems{count: 1}, handler: AppendItemsHandler, function: :handle}



    {:ok, 1, events} = Aggregate.execute(ExampleAggregate, prefixed_aggregate_uuid, context)

    assert length(events) == 1



    recorded_events = EventStore.stream_forward(prefixed_aggregate_uuid, 0) |> Enum.to_list()

    assert length(recorded_events) == 1



    assert {:error, :stream_not_found} == EventStore.stream_forward(aggregate_uuid, 0)

  end

end