Test KubeMQ Spring Applications

The kubemq-spring-boot-starter-test module gives you everything you need to test KubeMQ producers and consumers without standing up infrastructure by hand. It ships a @KubeMQTest annotation that layers KubeMQ test wiring on top of @SpringBootTest, an in-process MockKubeMQServer for fast unit-style tests, and TestContainers support for realistic integration tests. The annotation chooses the strategy for you, so your test class never has to start, configure, or tear down a broker.

Prerequisites

• kubemq-spring-boot-starter already in the application, in a Spring Boot 3.2.0+ project (see Getting Started with Spring Boot)
• kubemq-spring-boot-starter-test added to the test scope (see Add the Test Dependency below)
• Docker available if you use EMBEDDED Mode; a running KubeMQ broker if you use EXTERNAL Mode

Add the Test Dependency

The test module is published under the io.kubemq group, version 1.0.0. Add it to your test scope alongside the starter.

dependencies {
    testImplementation("io.kubemq:kubemq-spring-boot-starter-test:1.0.0")
}

<dependency>
    <groupId>io.kubemq</groupId>
    <artifactId>kubemq-spring-boot-starter-test</artifactId>
    <version>1.0.0</version>
    <scope>test</scope>
</dependency>

The module brings in spring-boot-starter-test, TestContainers, gRPC in-process transport, and Awaitility transitively, so no other test libraries are required to get started.

The @KubeMQTest Annotation

@KubeMQTest is meta-annotated with @SpringBootTest, so it bootstraps a full application context and adds KubeMQ test infrastructure on top. The mode attribute selects the testing strategy from the KubeMQTestMode enum:

public enum KubeMQTestMode {
    MOCK,      // gRPC InProcess mock server — no Docker, sub-100ms startup
    EMBEDDED,  // TestContainers with a real KubeMQ broker
    EXTERNAL   // Connect to a pre-existing broker (staging/CI)
}

Because @KubeMQTest aliases @SpringBootTest, you can pass properties and classes straight through:

@KubeMQTest(
    mode = KubeMQTestMode.MOCK,
    properties = "kubemq.client-id=order-test",
    classes = OrderTestConfig.class
)
class OrderEventTest {
    // ...
}

MOCK Mode

In MOCK mode the test framework starts a gRPC MockKubeMQServer on an in-process transport — there is no socket, no Docker, and no real broker. The server registers a MockKubeMQService that implements the KubeMQ gRPC contract, captures every message your application sends, and returns canned success responses. Startup is sub-100ms, which makes this the right mode for unit-style tests of template logic, serialization, and error handling.

The framework wires the mock into the context and points the application's kubemq.address at the in-process server, so your beans connect to the mock transparently. It registers several beans you can inject:

Inject the MockKubeMQServer (or the MockKubeMQService directly) and assert against the captured traffic:

import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate;
import io.kubemq.spring.boot.test.KubeMQTest;
import io.kubemq.spring.boot.test.KubeMQTestMode;
import io.kubemq.spring.boot.test.MockKubeMQServer;
import kubemq.Kubemq;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;

import java.nio.charset.StandardCharsets;
import java.util.List;

import static org.assertj.core.api.Assertions.assertThat;

@KubeMQTest(mode = KubeMQTestMode.MOCK)
class EventCaptureTest {

    @Autowired
    KubeMQTemplate template;

    @Autowired
    MockKubeMQServer mockServer;

    @Test
    void capturesPublishedEvent() {
        template.sendEvent("orders", "order-123");

        List<Kubemq.Event> events = mockServer.getMockService().getReceivedEvents();
        assertThat(events).hasSize(1);
        assertThat(events.get(0).getChannel()).isEqualTo("orders");
        assertThat(events.get(0).getBody().toStringUtf8())
                .contains("order-123");
    }
}

You can also drive negative paths: MockKubeMQService lets you stage responses and inject failures before the call under test runs.

import io.grpc.Status;
import io.grpc.StatusRuntimeException;

mockServer.getMockService()
        .setErrorOnNextCall(new StatusRuntimeException(Status.UNAVAILABLE));
// the next template call now sees the broker as unavailable

Asserting on Captured Messages

The raw accessors on MockKubeMQService return protobuf types (Kubemq.Event, Kubemq.Request, Kubemq.QueueMessage). For assertions that are decoupled from protobuf internals, the module provides immutable record wrappers that expose exactly what the application sent:

Each record has a static from(...) factory that adapts the protobuf message into the test-friendly view, so you can map the captured list and assert on plain fields:

import io.kubemq.spring.boot.test.CapturedEvent;
import java.nio.charset.StandardCharsets;

CapturedEvent captured = CapturedEvent.from(
        mockServer.getMockService().getReceivedEvents().get(0));

assertThat(captured.channel()).isEqualTo("orders");
assertThat(new String(captured.body(), StandardCharsets.UTF_8)).contains("order-123");
assertThat(captured.tags()).containsEntry("region", "eu-west-1");

EMBEDDED Mode

EMBEDDED mode spins up a real KubeMQ broker in a Docker container through TestContainers. The framework starts a KubeMQContainer — a wrapper around the europe-docker.pkg.dev/kubemq/images/kubemq image — before the Spring context loads, then injects the container's mapped gRPC address into kubemq.address so your beans connect to the live broker. This exercises the full gRPC round-trip and is the right mode for integration tests that need real broker semantics across all five messaging patterns.

The container exposes the gRPC port (50000), the REST API port (9090), and the dashboard/API port (8080), and waits on an HTTP health check before the test starts. Because Docker assigns random host ports, the connection details are wired automatically through Spring Boot's service-connection mechanism — KubeMQContainerConnectionDetailsFactory produces a KubeMQConnectionDetails from the running container so the address resolves to the correct mapped port without any manual configuration.

import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate;
import io.kubemq.spring.boot.test.KubeMQTest;
import io.kubemq.spring.boot.test.KubeMQTestMode;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;

@KubeMQTest(mode = KubeMQTestMode.EMBEDDED)
class OrderIntegrationTest {

    @Autowired
    KubeMQTemplate template;

    @Test
    void sendsAgainstRealBroker() {
        // The TestContainers KubeMQ broker is live; this is a real gRPC send.
        template.sendEvent("orders", "order-123");
    }
}

EXTERNAL Mode

EXTERNAL mode starts no infrastructure at all. The framework leaves your context untouched and your application connects to whatever broker kubemq.address already points at. This is the right mode for CI pipelines or staging environments that run a dedicated, shared broker, and it has zero per-test startup overhead.

Point the test at the broker through configuration — for example a test profile or an environment variable:

kubemq:
  address: ${KUBEMQ_ADDRESS:localhost:50000}
  client-id: order-external-test

@KubeMQTest(mode = KubeMQTestMode.EXTERNAL)
class OrderExternalTest {
    // Connects to the broker configured in kubemq.address.
}

If you need a broker locally for an EXTERNAL run, start one with Docker. The starter speaks gRPC on 50000; map 9090 as well if your tests also hit the shared HTTP/REST endpoints:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Test Harness Helpers

For assertions that must account for asynchronous delivery, inject the KubeMQTestHarness. It wraps the MockKubeMQService and provides a fluent, Awaitility-backed API that polls until a condition is met or a timeout elapses — so you do not have to sprinkle sleeps through your tests. It also exposes reset() to clear captured traffic between cases.

import io.kubemq.spring.boot.test.KubeMQTestHarness;
import java.time.Duration;

@Autowired
KubeMQTestHarness harness;

@Test
void eventArrivesOnChannel() {
    template.sendEvent("events.orders", "order-123");

    harness.expectEvent()
        .onChannel("events.orders")
        .withBodyContaining("order-123")
        .receivedWithin(Duration.ofSeconds(5));
}

The harness covers every pattern with expectEvent(), expectCommand(), expectQuery(), and expectQueueMessage(), plus expectNoEvents(Duration) for asserting that nothing was published.

KubeMQTestUtils rounds out the helpers with static utilities for setup and isolation. uniqueChannel("orders") generates a collision-free channel name like orders-a1b2c3d4 so parallel or repeated runs do not interfere, and waitForBrokerReady(address, timeout) blocks until a broker is accepting connections — handy in EXTERNAL mode before the first send.

import io.kubemq.spring.boot.test.KubeMQTestUtils;
import java.time.Duration;

String channel = KubeMQTestUtils.uniqueChannel("orders");
KubeMQTestUtils.waitForBrokerReady("localhost:50000", Duration.ofSeconds(10));

Choosing a Mode

Pick MOCK for fast Docker-free tests, EMBEDDED for a real broker via TestContainers, or EXTERNAL for a shared CI/staging broker.

A common layout is to run the bulk of your suite in MOCK mode for speed, keep a smaller set of EMBEDDED tests for full round-trip coverage, and reserve EXTERNAL for pipeline stages that already provision a broker.

Related

• Getting Started with Spring Boot — add the starter and send your first message
• Spring Boot integration overview — modules, architecture, and messaging patterns