feat(environment): create the EZE environment for spec tests

.github/workflows/build.yml

@@ -13,7 +13,7 @@ jobs:
     - name: Java setup
       uses: actions/setup-java@e54a62b3df9364d4b4c1c29c7225e57fe605d7dd # pin@v1
       with:
-        java-version: 11
+        java-version: 17
     - name: Cache
       uses: actions/cache@99d99cd262b87f5f8671407a1e5c1ddfa36ad5ba # pin@v1
       with:

.github/workflows/deploy.yml

@@ -25,7 +25,7 @@ jobs:
     - name: Set up Java environment
       uses: actions/setup-java@e54a62b3df9364d4b4c1c29c7225e57fe605d7dd # pin@v1
       with:
-        java-version: 11
+        java-version: 17
         gpg-private-key: ${{ secrets.MAVEN_CENTRAL_GPG_SIGNING_KEY_SEC }}
         gpg-passphrase: MAVEN_CENTRAL_GPG_PASSPHRASE
     - name: Deploy SNAPSHOT / Release

core/src/main/kotlin/io/zeebe/bpmnspec/SpecRunner.kt

@@ -78,8 +78,9 @@ class SpecRunner(
             resources.joinToString()
         )
         resources.forEach { resourceName ->
-            val resourceStream = resourceResolver.getResource(resourceName)
-            testRunner.deployProcess(resourceName, resourceStream)
+            resourceResolver.getResource(resourceName).use {
+                testRunner.deployProcess(resourceName, it)
+            }
         }
 
         logger.debug(

pom.xml

@@ -32,6 +32,8 @@
 
     <hazelcast.exporter.version>1.0.0</hazelcast.exporter.version>
 
+    <eze.version>1.0.2</eze.version>
+    <zeebe-hazelcast-exporter.version>1.2.1</zeebe-hazelcast-exporter.version>
     <zeebe-test-container.version>3.5.2</zeebe-test-container.version>
     <test-container.version>1.17.5</test-container.version>
 
@@ -79,6 +81,18 @@
         <scope>import</scope>
       </dependency>
 
+      <dependency>
+        <groupId>org.camunda.community</groupId>
+        <artifactId>eze</artifactId>
+        <version>${eze.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>io.zeebe.hazelcast</groupId>
+        <artifactId>zeebe-hazelcast-exporter</artifactId>
+        <version>${zeebe-hazelcast-exporter.version}</version>
+      </dependency>
+
       <dependency>
         <groupId>io.zeebe</groupId>
         <artifactId>zeebe-test-container</artifactId>
@@ -115,6 +129,12 @@
         <version>${junit.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>io.github.microutils</groupId>
+        <artifactId>kotlin-logging-jvm</artifactId>
+        <version>3.0.2</version>
+      </dependency>
+
       <dependency>
         <groupId>org.apache.logging.log4j</groupId>
         <artifactId>log4j-slf4j-impl</artifactId>

zeebe-test-runner/pom.xml

@@ -21,6 +21,16 @@
       <artifactId>zeebe-client-java</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>org.camunda.community</groupId>
+      <artifactId>eze</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>io.zeebe.hazelcast</groupId>
+      <artifactId>zeebe-hazelcast-exporter</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>io.zeebe</groupId>
       <artifactId>zeebe-test-container</artifactId>
@@ -41,6 +51,11 @@
       <artifactId>jackson-module-kotlin</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.github.microutils</groupId>
+      <artifactId>kotlin-logging-jvm</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter-api</artifactId>
@@ -77,6 +92,11 @@
       <artifactId>awaitility-kotlin</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-params</artifactId>
+      <scope>test</scope>
+    </dependency>
 
   </dependencies>
 

zeebe-test-runner/src/main/kotlin/io/zeebe/bpmnspec/runner/zeebe/AbstractTestRunner.kt

@@ -0,0 +1,124 @@
+package io.zeebe.bpmnspec.runner.zeebe
+
+import io.zeebe.bpmnspec.api.ProcessInstanceContext
+import io.zeebe.bpmnspec.api.runner.ElementInstance
+import io.zeebe.bpmnspec.api.runner.Incident
+import io.zeebe.bpmnspec.api.runner.ProcessInstanceState
+import io.zeebe.bpmnspec.api.runner.ProcessInstanceVariable
+import io.zeebe.bpmnspec.api.runner.TestRunner
+import mu.KLogger
+import java.io.InputStream
+
+abstract class AbstractTestRunner(
+    protected val environment: ZeebeEnvironment,
+    protected val reuseEnvironment: Boolean,
+    protected val beforeEachCallback: (ZeebeTestContext) -> Unit,
+    protected val afterEachCallback: (ZeebeTestContext) -> Unit,
+    protected val logger: KLogger
+) : TestRunner {
+    override fun beforeAll() {
+        if (reuseEnvironment) {
+            environment.setup()
+        }
+    }
+
+    override fun beforeEach() {
+        if (!reuseEnvironment || !environment.isRunning) {
+            environment.setup()
+        }
+        val testContext = ZeebeTestContext(zeebeClient = environment.zeebeClient)
+        beforeEachCallback(testContext)
+    }
+
+    override fun afterEach() {
+        val testContext = ZeebeTestContext(zeebeClient = environment.zeebeClient)
+        afterEachCallback(testContext)
+
+        if (!reuseEnvironment) {
+            environment.cleanUp()
+        }
+    }
+
+    override fun afterAll() {
+        if (reuseEnvironment) {
+            environment.cleanUp()
+        }
+    }
+
+    override fun deployProcess(name: String, bpmnXml: InputStream) {
+        logger.debug { "Deploying a BPMN. [name: $name]" }
+
+        environment.zeebeService.deployProcess(name, bpmnXml)
+    }
+
+    override fun createProcessInstance(
+        bpmnProcessId: String,
+        variables: String
+    ): ProcessInstanceContext {
+        logger.debug { "Creating a process instance. [BPMN-process-id: $bpmnProcessId, variables: $variables]" }
+
+        val response = environment.zeebeService.createProcessInstance(bpmnProcessId, variables)
+
+        return ZeebeProcessInstanceContext(processInstanceKey = response)
+    }
+
+    override fun completeTask(jobType: String, variables: String) {
+        logger.debug { "Starting a job worker to complete jobs. [job-type: $jobType, variables: $variables]" }
+
+        environment.zeebeService.completeTask(jobType, variables)
+    }
+
+    override fun publishMessage(messageName: String, correlationKey: String, variables: String) {
+        logger.debug {
+            "Publishing a message. [name: $messageName, correlation-key: $correlationKey, variables: $variables]"
+        }
+
+        environment.zeebeService.publishMessage(messageName, correlationKey, variables)
+    }
+
+    override fun throwError(jobType: String, errorCode: String, errorMessage: String) {
+        logger.debug {
+            "Starting a job worker to throw errors. [job-type: $jobType, error-code: $errorCode, error-message: $errorMessage]"
+        }
+
+        environment.zeebeService.throwError(jobType, errorCode, errorMessage)
+    }
+
+    override fun cancelProcessInstance(context: ProcessInstanceContext) {
+        val wfContext = context as ZeebeProcessInstanceContext
+
+        logger.debug {
+            "Cancelling a process instance. [key: ${wfContext.processInstanceKey}]"
+        }
+
+        environment.zeebeService.cancelProcessInstance(wfContext.processInstanceKey)
+    }
+
+    override fun getProcessInstanceContexts(): List<ProcessInstanceContext> {
+        return environment.zeebeEventRepository.getProcessInstanceKeys().map { ZeebeProcessInstanceContext(it) }
+    }
+
+    override fun getProcessInstanceState(context: ProcessInstanceContext): ProcessInstanceState {
+        val wfContext = context as ZeebeProcessInstanceContext
+
+        return environment.zeebeEventRepository.getProcessInstanceState(wfContext.processInstanceKey)
+    }
+
+    override fun getElementInstances(context: ProcessInstanceContext): List<ElementInstance> {
+        val wfContext = context as ZeebeProcessInstanceContext
+
+        return environment.zeebeEventRepository.getElementInstances(wfContext.processInstanceKey)
+    }
+
+    override fun getProcessInstanceVariables(context: ProcessInstanceContext): List<ProcessInstanceVariable> {
+        val wfContext = context as ZeebeProcessInstanceContext
+
+        return environment.zeebeEventRepository.getProcessInstanceVariables(wfContext.processInstanceKey)
+    }
+
+    override fun getIncidents(context: ProcessInstanceContext): List<Incident> {
+        val wfContext = context as ZeebeProcessInstanceContext
+
+        return environment.zeebeEventRepository.getIncidents(wfContext.processInstanceKey)
+    }
+}

zeebe-test-runner/src/main/kotlin/io/zeebe/bpmnspec/runner/zeebe/DefaultZeebeEnvironment.kt

@@ -0,0 +1,140 @@
+package io.zeebe.bpmnspec.runner.zeebe
+
+import io.camunda.zeebe.client.ZeebeClient
+import io.zeebe.bpmnspec.runner.zeebe.zeeqs.ZeeqsClient
+import io.zeebe.bpmnspec.runner.zeebe.zeeqs.ZeeqsZeebeEventRepository
+import io.zeebe.containers.ZeebeContainer
+import org.slf4j.LoggerFactory
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.containers.Network
+import org.testcontainers.containers.wait.strategy.Wait
+import org.testcontainers.utility.DockerImageName
+
+class DefaultZeebeEnvironment(
+    val zeebeImage: String = System.getProperty(
+        "zeebeImage",
+        "ghcr.io/camunda-community-hub/zeebe-with-hazelcast-exporter"
+    ),
+    val zeebeImageVersion: String = System.getProperty("zeebeImageVersion", "8.0.6"),
+    private val zeebeEventRepositoryProvider: (zeeqsClient: ZeeqsClient) -> ZeebeEventRepository = {
+        ZeeqsZeebeEventRepository(
+            it
+        )
+    },
+    private val zeebeServiceProvider: (zeebeClient: ZeebeClient) -> ZeebeService = { ZeebeClientZeebeService(it) }
+) : ZeebeEnvironment {
+
+    private val logger = LoggerFactory.getLogger(ZeebeTestRunner::class.java)
+
+    val zeeqsImage = "ghcr.io/camunda-community-hub/zeeqs"
+    val zeeqsImageVersion: String = "2.4.0"
+
+    private val zeebeHost = "zeebe"
+    private val hazelcastPort = 5701
+    private val zeeqsGraphqlPort = 9000
+
+    private val closingSteps = mutableListOf<AutoCloseable>()
+
+    override lateinit var zeebeClient: ZeebeClient
+    override lateinit var zeeqsClient: ZeeqsClient
+    override lateinit var zeebeEventRepository: ZeebeEventRepository
+    override lateinit var zeebeService: ZeebeService
+
+    override var isRunning = false
+
+    override fun setup() {
+        val network = Network.newNetwork()!!
+        closingSteps.add(network)
+
+        listOf(this::startZeebeContainer, this::startZeeqsContainer)
+            .parallelStream()
+            .forEach { it.invoke(network) }
+
+        isRunning = true
+    }
+
+    private fun startZeebeContainer(network: Network) {
+        val zeebeImageName = DockerImageName.parse("$zeebeImage:$zeebeImageVersion")
+        val zeebeContainer = ZeebeContainer(zeebeImageName)
+            .withAdditionalExposedPort(hazelcastPort)
+            .withNetwork(network)
+            .withNetworkAliases(zeebeHost)
+
+        logger.debug("Starting the Zeebe container [image: {}]", zeebeContainer.dockerImageName)
+        try {
+            zeebeContainer.start()
+        } catch (e: Exception) {
+            logger.error("Failed to start the Zeebe container", e)
+            logger.debug("Zeebe container output: {}", zeebeContainer.logs)
+
+            throw RuntimeException("Failed to start the Zeebe container", e)
+        }
+
+        logger.debug("Started the Zeebe container")
+        closingSteps.add(zeebeContainer)
+
+        val zeebeGatewayPort = zeebeContainer.externalGatewayAddress
+
+        zeebeClient = ZeebeClient
+            .newClientBuilder()
+            .gatewayAddress(zeebeGatewayPort)
+            .usePlaintext()
+            .build()
+
+        zeebeService = zeebeServiceProvider(zeebeClient)
+
+        closingSteps.add(zeebeClient)
+        closingSteps.add(zeebeService)
+
+        // verify that the client is connected
+        try {
+            zeebeClient.newTopologyRequest().send().join()
+        } catch (e: Exception) {
+            logger.error("Failed to connect the Zeebe client", e)
+            logger.debug("Zeebe container output: {}", zeebeContainer.logs)
+
+            throw RuntimeException("Failed to connect the Zeebe client", e)
+        }
+    }
+
+    private fun startZeeqsContainer(network: Network) {
+        val zeeqsContainer = ZeeqsContainer(zeeqsImage, zeeqsImageVersion)
+            .withEnv("zeebe.client.worker.hazelcast.connection", "$zeebeHost:$hazelcastPort")
+            .withExposedPorts(zeeqsGraphqlPort)
+            .waitingFor(Wait.forHttp("/actuator/health"))
+            .withNetwork(network)
+            .withNetworkAliases("zeeqs")
+
+        logger.debug("Starting the ZeeQS container [image: {}]", zeeqsContainer.dockerImageName)
+        try {
+            zeeqsContainer.start()
+        } catch (e: Exception) {
+            logger.error("Failed to start the ZeeQS container", e)
+            logger.debug("ZeeQS container output: {}", zeeqsContainer.logs)
+
+            throw RuntimeException("Failed to start the ZeeQS container", e)
+        }
+
+        logger.debug("Started the ZeeQS container")
+        closingSteps.add(zeeqsContainer)
+
+        val zeeqsContainerHost = zeeqsContainer.host
+        val zeeqsContainerPort = zeeqsContainer.getMappedPort(zeeqsGraphqlPort)
+
+        zeeqsClient = ZeeqsClient(zeeqsEndpoint = "$zeeqsContainerHost:$zeeqsContainerPort/graphql")
+        zeebeEventRepository = zeebeEventRepositoryProvider(zeeqsClient)
+    }
+
+    override fun cleanUp() {
+        logger.debug("Closing resources")
+        closingSteps.toList().reversed().forEach(AutoCloseable::close)
+
+        logger.debug("Closed resources")
+
+        isRunning = false
+    }
+
+    class ZeeqsContainer(imageName: String, version: String) :
+        GenericContainer<ZeeqsContainer>("$imageName:$version")
+
+}

zeebe-test-runner/src/main/kotlin/io/zeebe/bpmnspec/runner/zeebe/TestEnvironment.kt

@@ -0,0 +1,17 @@
+package io.zeebe.bpmnspec.runner.zeebe
+
+import io.camunda.zeebe.client.ZeebeClient
+import io.zeebe.bpmnspec.runner.zeebe.zeeqs.ZeeqsClient
+
+interface TestEnvironment {
+    val zeebeClient: ZeebeClient
+    val zeeqsClient: ZeeqsClient
+    val zeebeEventRepository: ZeebeEventRepository
+    val zeebeService: ZeebeService
+
+    val isRunning: Boolean
+
+    fun setup()
+
+    fun cleanUp()
+}