Integration testing with event sourcing systems

Question

I'm working on a PoC where we use CQRS in combination with Event Sourcing. We use the Axon framework and Axon server as toolset.

We have some microservices (Maven packages) with some business logic.

A simple overview of the application flow:

We post a xml message (with REST) to service 1 that will result in an event (with Aggregate). Service 2 handles the event "fired" by service 1 and starts a saga flow. Part of the sage flow is for example to send a mail message.

I can do some tests with Axon Test to test the aggregate from service 1 or the saga from service 2. But is there a good option to do a real integration test where we start with posting a message to the REST interface and check all the operations in the aggregate and saga (inclusive sending mail and so on)

Maybe this kind of integration test is overdone and it's better to test each component on it's own. I doubt what's needed / the best solution to test this type of system.

Answer 1

I suggest to have a look at testcontainers (https://www.testcontainers.org/)

It provides a very convenient way to start up and cleanly tear down docker containers in JUnit tests. This feature is very useful for integration testing of applications against real databases and any other resource (for example Axon Server) for which a docker image is available (https://hub.docker.com/r/axoniq/axonserver/).

I'm sharing some code snippets from JUnit 4 test class (Kotlin). Hopefully this can help you to get started and evolve your specific test strategy (integration should cover smaller scope then end-to-end tests). My opinion is that integration test should focus on Axon messaging API components and REST API components separately/independently. End-to-end should cover all components in your microservice/s.

@RunWith(SpringRunner::class)
@SpringBootTest
@ContextConfiguration(initializers = [DrestaurantCourierCommandMicroServiceIT.Initializer::class])
internal class DrestaurantCourierCommandMicroServiceIT {

    @Autowired
    lateinit var eventStore: EventStore

    @Autowired
    lateinit var commandGateway: CommandGateway

    companion object {

        // An Axon Server container
        @ClassRule
        @JvmField
        var axonServerTestContainer = KGenericContainer(
                "axoniq/axonserver")
                .withExposedPorts(8024, 8124)
                .waitingFor(Wait.forHttp("/actuator/info").forPort(8024))
                .withStartupTimeout(Duration.of(60L, ChronoUnit.SECONDS))

        // A PostgreSQL container is being started up using a JUnit Class Rule which gets triggered before any of the tests are run:

        @ClassRule
        @JvmField
        var postgreSQLContainer = KPostgreSQLContainer(
                "postgres:latest")
                .withDatabaseName("drestaurant")
                .withUsername("demouser")
                .withPassword("thepassword")
                .withStartupTimeout(Duration.of(60L, ChronoUnit.SECONDS))
    }

    // Pass details on the application as properties BEFORE Spring starts creating a test context for the test to run in:
    class Initializer : ApplicationContextInitializer<ConfigurableApplicationContext> {

        override fun initialize(configurableApplicationContext: ConfigurableApplicationContext) {
            val values = TestPropertyValues.of(
                    "spring.datasource.url=" + postgreSQLContainer.jdbcUrl,
                    "spring.datasource.username=" + postgreSQLContainer.username,
                    "spring.datasource.password=" + postgreSQLContainer.password,
                    "spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.PostgreSQLDialect",
                    "axon.axonserver.servers=" + axonServerTestContainer.containerIpAddress + ":" + axonServerTestContainer.getMappedPort(8124)
            )
            values.applyTo(configurableApplicationContext)
        }
    }


    @Test
    fun `restaurant command microservice integration test - happy scenario`() {

        val who = "johndoe"
        val auditEntry = AuditEntry(who, Calendar.getInstance().time)
        val maxNumberOfActiveOrders = 5
        val name = PersonName("Ivan", "Dugalic")
        val orderId = CourierOrderId("orderId")

        // ******* Sending the `createCourierCommand` ***********
        val createCourierCommand = CreateCourierCommand(name, maxNumberOfActiveOrders, auditEntry)
        commandGateway.sendAndWait<Any>(createCourierCommand)
        await withPollInterval org.awaitility.Duration.ONE_SECOND atMost org.awaitility.Duration.FIVE_SECONDS untilAsserted {
            val latestCourierCreatedEvent = eventStore.readEvents(createCourierCommand.targetAggregateIdentifier.identifier).asStream().toList().last().payload as CourierCreatedEvent
            assertThat(latestCourierCreatedEvent.name).isEqualTo(createCourierCommand.name)
            assertThat(latestCourierCreatedEvent.auditEntry.who).isEqualTo(createCourierCommand.auditEntry.who)
            assertThat(latestCourierCreatedEvent.maxNumberOfActiveOrders).isEqualTo(createCourierCommand.maxNumberOfActiveOrders)
        }

        // ******* Sending the `createCourierOrderCommand` **********
        val createCourierOrderCommand = CreateCourierOrderCommand(orderId, auditEntry)
        commandGateway.sendAndWait<Any>(createCourierOrderCommand)
        await withPollInterval org.awaitility.Duration.ONE_SECOND atMost org.awaitility.Duration.FIVE_SECONDS untilAsserted {
            val latestCourierOrderCreatedEvent = eventStore.readEvents(createCourierOrderCommand.targetAggregateIdentifier.identifier).asStream().toList().last().payload as CourierOrderCreatedEvent
            assertThat(latestCourierOrderCreatedEvent.aggregateIdentifier.identifier).isEqualTo(createCourierOrderCommand.targetAggregateIdentifier.identifier)
            assertThat(latestCourierOrderCreatedEvent.auditEntry.who).isEqualTo(createCourierOrderCommand.auditEntry.who)
        }

        // ******* Assign the courier order to courier **********
        val assignCourierOrderToCourierCommand = AssignCourierOrderToCourierCommand(orderId, createCourierCommand.targetAggregateIdentifier, auditEntry)
        commandGateway.sendAndWait<Any>(assignCourierOrderToCourierCommand)
        await withPollInterval org.awaitility.Duration.ONE_SECOND atMost org.awaitility.Duration.FIVE_SECONDS untilAsserted {
            val latestCourierOrderAssignedEvent = eventStore.readEvents(assignCourierOrderToCourierCommand.targetAggregateIdentifier.identifier).asStream().toList().last().payload as CourierOrderAssignedEvent
            assertThat(latestCourierOrderAssignedEvent.aggregateIdentifier.identifier).isEqualTo(assignCourierOrderToCourierCommand.targetAggregateIdentifier.identifier)
            assertThat(latestCourierOrderAssignedEvent.auditEntry.who).isEqualTo(assignCourierOrderToCourierCommand.auditEntry.who)
            assertThat(latestCourierOrderAssignedEvent.courierId.identifier).isEqualTo(assignCourierOrderToCourierCommand.courierId.identifier)
        }

    }
}

class KGenericContainer(imageName: String) : GenericContainer<KGenericContainer>(imageName)
class KPostgreSQLContainer(imageName: String) : PostgreSQLContainer<KPostgreSQLContainer>(imageName)

Answer 2

Axon developers suggest to go with docker solution as mentioned here. Testcontainers seems to be the best here. My java snippet:

@ActiveProfiles("test")
public class TestContainers {

    private static final int AXON_HTTP_PORT = 8024;
    private static final int AXON_GRPC_PORT = 8124;

    public static void startAxonServer() {
        GenericContainer axonServer = new GenericContainer("axoniq/axonserver:latest")
                .withExposedPorts(AXON_HTTP_PORT, AXON_GRPC_PORT)
                .waitingFor(
                        Wait.forLogMessage(".*Started AxonServer.*", 1)
                );
        axonServer.start();

        System.setProperty("ENV_AXON_GRPC_PORT", String.valueOf(axonServer.getMappedPort(AXON_GRPC_PORT)));
    }

Call startAxonServer method in your @BeforeClass. Now you have to obtain external docker ports (these indicated in withExposedPorts are docker-internal). You can do it in runtime via getMappedPort as shown in my snippet. Remember to provide connection configuration to your test suite. My example on spring boot as follows:

axon:
  axonserver:
    servers: localhost:${ENV_AXON_GRPC_PORT}

Complete working solution can be found on my github project.