Testowanie CQRS na przykładzie Axon Framework we Flowee
Flowee to jeden z flagowych projektów Finture – innowacyjna platforma do zarządzania procesami biznesowymi, umożliwiająca ich automatyzację i optymalizację, a przez to redukcję kosztów, zwiększenie efektywności i przepływu dokumentów i informacji. Dzięki temu to, co dzieje się w firmie „od kuchni”, dzieje się płynnie i sprawnie. Tu znajdziesz więcej informacji o tym, do czego można wykorzystać Flowee.
Żeby Flowee mogło skutecznie wspomagać i automatyzować procesy biznesowe – musimy zadbać o jego niezawodność. Zważywszy na to, jednym ze sposobów, w jaki zespół Flowee zapewnia bezpieczeństwo zmian, są testy jednostkowe. Są one szczególnie ważne w przypadku architektury rozproszonej oraz architektury opartej na zdarzeniach. W takich środowiskach nawet niewielka zmiana w modelach komend, zdarzeń czy agregatów może spowodować sporo zamieszania w środowisku produkcyjnym.Testy jednostkowe pozwalają nam wywołać dany element aplikacji, by sprawdzić, czy wynik wywołania (np. zwrócona wartość czy też wyjątek) jest zgodny z oczekiwaniami.
Zatem dobre pokrycie takiego kodu testami pozwala już na etapie dewelopmentu zidentyfikować (a w konsekwencji – wyeliminować) błędy implementacji.
Flowee jest właśnie taką aplikacją – zbudowaną na architekturze mikroserwisowej, w której komunikacja pomiędzy domenami opiera się między innymi na wykorzystaniu zdarzeń, do czego służy nam Axon Framework.
Testowanie CQRS – na czym polega
Nie do końca wiesz, o co chodzi w poprzednim zdaniu? Spróbujmy to rozgryźć:
Wyobraź sobie nowoczesną kuchnię w dużej, ekskluzywnej restauracji. Kluczem do sukcesu jest perfekcyjny plan, organizacja i dbałość o jakość na każdym etapie. W związku z czym wspaniałe doświadczenie, jakim jest zjedzenie posiłku w tej restauracji, wymaga staranności od początku do końca. Jak wiemy, wszystko zaczyna się od idealnie dobranych składników. Następnie liczy się organizacja pracy kuchni i obsługi na sali. Nie mniej ważna jest komunikacja między pracownikami kuchni a obsługą sali.
Architektura rozproszona to sposób organizacji naszej restauracji. Każda osoba w niej pracująca (mikroserwis) ma swoją specjalizację i określone zadania. Pracują niezależnie, wykonując swoje zadania autonomicznie – choć ostateczny efekt zależy od ich synchronizacji i komunikacji między nimi. Na przykład – cukiernik specjalizuje się w deserach, jeden z kucharzy zajmuje się odpowiednim wysmażaniem mięs a młodszy kucharz spędza swój czas na krojeniu warzyw i składników.
Organizacja naszej kuchni to także architektura oparta o zdarzenia – sygnałem do wykonania czynności w jednej części kuchni może być wydarzenie w zupełnie innej części. Gdy pomocnik cukiernika wyjmuje z piekarnika babeczki, to sygnał dla cukiernika, by przygotować dekoracje. Gdy szef kuchni przygotuje na talerzu danie ze składników przygotowanych przez pozostałych kucharzy – obsługa zabiera je, by zanieść do odpowiedniego stolika.
CQRS (Command Query Responsibiliy Segregation) oznacza, że w naszej architekturze rozdzielone są operacje odczytu i zapisu danych. A Axon Framework to narzędzie programistyczne zaprojekowane do budowy skalowalnych i elastycznych aplikacji opartych na architekturze mikroserwisów i wzorcach CQRS.
W naszym artykule przybliżymy nieco architekturę Flowee, by następnie pokazać, jak można jednostkowo testować elementy z trzech głównych obszarów tej architektury.
Jeśli chcesz dowiedzieć się, dlaczego tak ważne są testy jednostkowe, ale nie chcesz zagłębiać się w technikalia testowania przeskocz do ostatniego akapitu!
Architektura przepływu danych we Flowee
Jak już wspomniano – Flowee to aplikacja zbudowana w architekturze mikroserwisowej. Zatem komunikacja pomiędzy domenami opiera się między innymi na wykorzystaniu zdarzeń (do tego wykorzystujemy Axon Framework).
Poniżej przedstawiony jest przykładowy przepływ w interakcji z użytkownikiem.
Gdy użytkownik Flowee chce zaprojektować nowy proces – na początku tworzy sam obiekt procesu, zawierający podstawowe informacje, takie jak: nazwa procesu, kod (identyfikator procesu) i jego opis.
Te dane odbierane są z REST API i opakowywane w komendę (command), a następnie wysyłane za pośrednictwem Axon Framework do serwera. Stamtąd trafiają do odpowiedniego agregatu.
Ponieważ jest to obiekt inicjalny, tworzy się nowa instancja procesu. W efekcie w agregacie powstaje zdarzenie (event), który następnie trafia do metod obsługujących (handlerów). Jedna z tych metod odpowiada za zaktualizowanie stanu agregatu (w naszej implementacji agregat jest „źródłem prawdy”) natomiast inne – aktualizują odpowiednie bazy danych o wpis dotyczący utworzenia nowego procesu.
Użytkownik stworzył właśnie obiekt procesu. Może przejść teraz do tworzenia jego definicji (we Flowee służy do tego dedykowana aplikacja).
Gdy obiekt jest zapisywany – tworzy się komenda informująca o powstaniu nowej wersji definicji procesu. Komenda, na podstawie identyfikatora, trafia do odpowiedniej instancji agregatu. Tam tworzone jest zdarzenie powstania nowej wersji.
Podobnie jak przy tworzeniu obiektu procesu – zdarzenie odbierane jest w samej instancji agregatu, gdzie aktualizuje się jego stan oraz odbierane jest przez zainteresowane bazy danych (przy czym nie muszą to być te same bazy danych, co w pierwszym przypadku).
Testowanie wysłania komendy do agregatu
Chcąc przetestować wysyłanie komendy do agregatu w przypadku pierwszej operacji (utworzenia definicji procesu) z pewnością będziemy chcieli wiedzieć:
- Co się stanie, jeśli użytkownik będzie próbował utworzyć proces o takiej samej nazwie lub kodzie jak już istniejący?
- Co się stanie, jeśli nazwa lub kod nie spełnią warunków walidacji (np. są zbyt krótkie czy zbyt długie)?
- Co się stanie, gdy użytkownik nie wpisze kodu lub nazwy?
- Czy komenda została wysłana?
- Czy zawartość komendy jest rzeczywiście zgodna z zapytaniem użytkownika?
Żeby na te pytania odpowiedzieć, tworzymy scenariusz inicjalny, w którym bazę danych zasilamy przykładowymi procesami. W testach nigdy nie chcemy korzystać z produkcyjnej bazy danych – dlatego wykorzystujemy bibliotekę Testcontainers, która umożliwia jej odwzorowanie w dedykowanym kontenerze tworzonym jedynie na potrzeby testów. Prawidłowość odwzorowania gwarantuje nam wykorzystanie w całym systemie biblioteki Liquibase.
Kolejny krok – to przygotowanie testowego zapytania oraz oczekiwanej komendy.
// given
initScenario();
var givenRequest = CreateProcessRequest.builder()
.code(„testCode”)
.name(„testName”)
.description(„testDescription”)
.build();
var expectedCommand = CreateProcessCommand.builder()
.code(new ProcessId(„testCode”))
.name(„testName”)
.description(„testDescription”)
.build();
Oczekiwanie
Gdy oczekujemy, że operacja ma się zakończyć pozytywnie, określamy, że ma zostać wysłana dana komenda (i tylko ona):
// when
processService.createProcess(givenRequest);
// then
verify(commandGateway).sendAndWait(expectedCommand);
verifyNoInteractions(commandGateway);
CommandGateway to zaślepiony w Mockito interfejs (mock) będący elementem Axon Framework. Służy on do wysyłania komend. ProcessService to usługa, którą testujemy – z wstrzykniętym CommandGateway i bazą skonfigurowaną w Testcontainers.
Jeśli w naszym scenariuszu operacja ma się zakończyć wyrzuceniem wyjątku – określamy, jaki to wyjątek (wykorzystując standardową asercję JUnit) i upewniamy się, że żadna komenda nie została wysłana:
assertThrows(ProcessCreateException.class,
() -> processService.createProcess(givenRequest));
verifyNoInteractions(commandGateway);
W ramach wysyłania komendy możemy też zasymulować, jaką odpowiedź otrzymamy:
- a) na dowolną zawartość:
when(commandGateway.sendAndWait(any())).thenReturn(new SomeAggregateResponse(„response”));
- b) na komendę jakiejś klasy:
when(commandGateway.sendAndWait(any(CreateProcessCommand.class))).thenReturn(new SomeAggregateResponse(„response”));
- c) lub na konkretna implementację komendy (co może być przydatne w testowaniu logiki następującej po wysłaniu komendy:
when(commandGateway.sendAndWait(expectedCommand)).thenReturn(new SomeAggregateResponse(„response”));
Testowanie wysłania komendy do agregatu
Testowanie agregatów polega na sprawdzeniu dwóch głównych obszarów: wyemitowanych zdarzeń oraz stanu agregatu. Oba aspekty sprawdzamy za pomocą jednej z bibliotek Axon Framework o nazwie axon-test.
Głównym elementem agregatu jest jego identyfikator, oznaczony jako pole z adnotacją @AggregateIdentifier. W agregacie znajdują się specjalne klasy oznaczone adnotacjami @CommandHandler oraz @EventSourcingHandler które odpowiadają odpowiednio za interpretację komend wraz z tłumaczeniem ich na zdarzenia oraz za aktualizację stanu agregatu.
Testując agregat, będziemy chcieli dowiedzieć się:
- Czy komenda została odebrana przez agregat?
- Czy komenda została właściwie przetłumaczona na zdarzenie?
- Czy zdarzenia zostały wysłane?
- Czy zdarzenia zostały odebrane przez agregat?
- Czy stan agregatu został prawidłowo zaktualizowany?
Pierwszym krokiem jest przygotowanie testowego agregatu. Wykorzystujemy tu klasę AggregateTestFixture z Axon Framework.
fixture = new AggregateTestFixture<>(ProcessAggregate.class);
Jest to obiekt, który odzwierciedla zachowania agregatu na podstawie przesłanych komend i zdarzeń, bazując na obiekcie typu TestExecutor. Imitację pustego agregatu możemy uzyskać wywołując:
var emptyAggregate = fixture.given();
Przykładowy pozytywny scenariusz odebrania komendy można przedstawić za pomocą następującego kodu:
fixture.given()
.when(createProcessCommand)
.expectSuccessfulHandlerExecution()
.expectEvents(processCreatedEvent);
To bardzo czytelne rozwiązanie. Gdy odbierzemy komendę createProcessCommand mając pusty agregat, z sukcesem utworzymy zdarzenie processCreatedEvent.
Jeśli chcemy operować na agregacie, który już został utworzony i ma za sobą jakąś historię, możemy to określić w następujący sposób:
fixture.given(processCreatedEvent, initialProcessModifiedEvent)
.when(modifyProcessCommand)
.expectSuccessfulHandlerExecution()
.expectEvents(otherProcessModifiedEvent);
Sprawdzamy też, czy w sytuacji, gdy nie chcemy, aby zdarzenie zostało wyemitowane (przykładowo: komenda nie zmienia stanu agregatu), to faktycznie nie zostanie ono wyemitowane:
fixture.given(processCreatedEvent, initialProcessModifiedEvent)
.when(changeNothingInProcessCommand)
.expectSuccessfulHandlerExecution()
.expectNoEvents();
Testowanie konsumpcji
Testując agregat, sprawdzamy nie tylko czy zdarzenie zostało wyemitowane, ale także czy zostało prawidłowo skonsumowane. W naszym przykładzie posługujemy się interfejsem ResultValidator z biblioteki Axon Framework.
Przyjmijmy, że dodajemy drugą wersję procesu:
ResultValidator<ProcessAggregate> aggregate = fixture.given(processCreatedEvent)
.when(addNewVersionOfProcessCommand);
Gdy mamy tak przygotowany agregat, sprawdzamy, czy w ramach odebrania komendy do agregatu zostały przekazane odpowiednie zdarzenia:
aggregate.expectSuccesfulHandlerExecution()
.expectEvents(processVersionAddedEvent);
Sprawdzamy też stan agregatu:
aggregate.expectState(aggregateInstance -> assertThat(aggregateInstance.versions())
.hasSize(2);
Im bardziej złożony jest agregat i logika konsumpcji komendy, tym większy będzie zakres testów. Warto testować pojedyncze zmiany. Ale warto też sprawdzać, jak aplikacja zachowuje się przy złożonych scenariuszach.
Testowanie konsumpcji zdarzeń emitowanych przez agregat
Ostatnim etapem dla naszego przepływu danych – jest odebranie zdarzenia w projekcji. Na tym etapie sprawdzamy:
- Czy zdarzenie zostało odebrane przez metodę?
- Czy dane ze zdarzenia prawidłowo zmieniły stan bazy danych?
- Czy aplikacja zachowa się przy przesłaniu niepełnych lub nieprawidłowych danych?
- Czy aplikacja prawidłowo reaguje na przesłanie duplikatu?
Podobnie jak w przypadku komend i tutaj posługujemy się biblioteką Testcontainers, by odseparować testy od danych, a tym samym zapewnić ich powtarzalność. Komunikacja z bazą danych we Flowee odbywa się przy wykorzystaniu biblioteki JOOQ, której składnia zbliżona jest do czystego SQL i tym samym daje możliwość łatwego tworzenia złożonych zapytań.
// given
initScenario();
var givenEvent = CreateProcessCommand.builder()
.code(new ProcessId(„testCode”))
.name(„testName”)
.description(„testDescription”)
.build();
// when
processProjection.handle(givenEvent);
var result = cotnainer.dsl().selectFrom(PROCESS_TABLE)
.where(PROCESS_TABLE.CODE.eq(„testCode”)
.fetch();
// then
assertThat(result).hasSize(1);
(…)
W zależności od potrzeb, możemy posłużyć się tu dowolnym zestawem asercji.
Testy jednostkowe podstawą funkcjonalnej architektury
Testy jednostkowe na każdym etapie przepływu danych pomagają zapobiegać krytycznym błędom systemu. Ich koszt jest niewielki w porównaniu z korzyściami, szczególnie przy zmianach w modelu agregatu, komend czy zdarzeń. W systemach rozproszonych nawet drobna modyfikacja może mieć wiele konsekwencji, których bez testów trudno zauważyć. Choć aplikacja może działać bez testów, ich stosowanie znacznie ułatwia rozwój i dalszą rozbudowę systemu.
Testowanie CQRS jest kluczowym elementem zapewnienia jakości w systemach wykorzystujących ten wzorzec architektoniczny. Odpowiednie podejście do testów pozwala lepiej kontrolować logikę poleceń i zapytań, a także zwiększa stabilność całego rozwiązania. Dzięki temu testowanie CQRS wspiera rozwój skalowalnych i niezawodnych aplikacji.