refactor: introduce extra helper method for tests to count active subprocesses (10a31e4e) · Commits · OSL / MDG HypO / code / org.etsi.osl.hypo.core / org.etsi.osl.hypo.core.sonata

src/test/java/io/sonata/kafka/helpers/TestHelpers.java

+12 −0

Changes for src/test/java/io/sonata/kafka/helpers/TestHelpers.java: 12 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -8,6 +8,7 @@ import io.smallrye.reactive.messaging.memory.InMemorySink;
		import io.sonata.kafka.supplier.ActiveProcessSupplier;
		import io.sonata.kafka.supplier.HasReceivedMessagesSupplier;
		import io.sonata.kafka.supplier.IsProcessCompletedSupplier;
		import io.sonata.kafka.supplier.NumberOfActiveProcessesSupplier;
		import io.sonata.kafka.supplier.NumberOfProcessesSupplier;
		import io.sonata.kafka.supplier.ServiceUpdateChannelSupplier;
		import java.io.File;
		@@ -88,6 +89,17 @@ public final class TestHelpers {
		.last();
		}

		public static Uni<Boolean> pollProcessActiveInstances(
		Process<? extends Model> process, int numberOfInstances) {
		return Multi.createBy()
		.repeating()
		.supplier(new NumberOfActiveProcessesSupplier(process, numberOfInstances))
		.withDelay(Duration.ofMillis(100))
		.whilst(s -> (s == null \|\| !s))
		.collect()
		.last();
		}

		@NotNull
		public static FileReader getFileReader(String fileSubdirectory, String fileName)
		throws FileNotFoundException {

src/test/java/io/sonata/kafka/supplier/NumberOfActiveProcessesSupplier.java

0 → 100644

+26 −0

Changes for src/test/java/io/sonata/kafka/supplier/NumberOfActiveProcessesSupplier.java: 26 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		package io.sonata.kafka.supplier;

		import java.util.function.Supplier;
		import org.kie.kogito.Model;
		import org.kie.kogito.process.Process;
		import org.kie.kogito.process.ProcessInstance;

		public class NumberOfActiveProcessesSupplier implements Supplier<Boolean> {
		final Process<? extends Model> process;
		final int numberOfProcesses;

		public NumberOfActiveProcessesSupplier(Process<? extends Model> process, int numberOfProcesses) {
		this.process = process;
		this.numberOfProcesses = numberOfProcesses;
		}

		@Override
		public Boolean get() {
		long countTotalProcesses = process.instances().stream().count();
		long countActiveProcesses =
		process.instances().stream()
		.filter(processInstance -> processInstance.status() == ProcessInstance.STATE_ACTIVE)
		.count();
		return countTotalProcesses == numberOfProcesses && countActiveProcesses == numberOfProcesses;
		}
		}

src/test/java/sonata/integration/tests/HelmEngineProcessesTest.java

+4 −4

Changes for src/test/java/sonata/integration/tests/HelmEngineProcessesTest.java: 4 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -284,7 +284,7 @@ class HelmEngineProcessesTest {

		incomingChannelFromTmfServiceActivation.send(cloudEventServiceActivationState.toStringAsJson());

		TestHelpers.pollProcessInstances(helmEngineServiceActivationProcess, 1)
		TestHelpers.pollProcessActiveInstances(helmEngineServiceActivationProcess, 1)
		.await()
		.atMost(DURATION);

		@@ -310,7 +310,7 @@ class HelmEngineProcessesTest {
		String kogitoParentProcessInstanceId =
		eventServiceActivationStateReceived.getParentProcessId().toString();

		UUID kogitoProcessInstanceId = eventServiceActivationStateReceived.getProcessInstanceId();
		UUID activationProcessInstanceId = eventServiceActivationStateReceived.getProcessInstanceId();

		// trigger the intermediate step ( mock the Helm Engine Response)

		@@ -323,7 +323,7 @@ class HelmEngineProcessesTest {
		objectMapper.readValue(
		incomingHelmEngineResponseAsJson.toString(),
		CloudEventHelmEngineActivationResultMessage.class);
		cloudEventHelmEngineActivationResultMessage.setProcessReferenceId(kogitoProcessInstanceId);
		cloudEventHelmEngineActivationResultMessage.setProcessReferenceId(activationProcessInstanceId);
		cloudEventHelmEngineActivationResultMessage
		.getHelmEngineActivationResult()
		.setHelmEngineOrderStatus(helmEngineActivationStatus);
		@@ -340,7 +340,7 @@ class HelmEngineProcessesTest {

		// poll until process is completed
		TestHelpers.pollProcessCompletionStatus(
		helmEngineServiceActivationSubprocess, kogitoProcessInstanceId.toString())
		helmEngineServiceActivationSubprocess, activationProcessInstanceId.toString())
		.await()
		.atMost(DURATION);

src/test/java/sonata/integration/tests/K8sAndOssClientProcessesTests.java

+47 −25

Changes for src/test/java/sonata/integration/tests/K8sAndOssClientProcessesTests.java: 47 added lines, 25 removed lines.
Original line number	Diff line number	Diff line
		@@ -320,10 +320,12 @@ class K8sAndOssClientProcessesTests {
		incomingChannelFromTMFExistingK8s.send(k8sServiceOrderMessage.toStringAsJson());

		// assert that the parent process has been triggered
		TestHelpers.pollProcessInstances(k8sStandaloneProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(k8sStandaloneProcess, 1).await().atMost(DURATION);

		// assert that Fabric API client subprocess has been triggered
		TestHelpers.pollProcessInstances(fabricApiServiceConnectionProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(fabricApiServiceConnectionProcess, 1)
		.await()
		.atMost(DURATION);

		// POLL IN ORDER TO "FORCE" THE TEST TO WAIT UNTIL THE PROCESS REACHES THE INTERMEDIATE STEP
		TestHelpers.pollKafkaChannel(outgoingChannelToFabricApiDomainCreation, 1)
		@@ -380,7 +382,7 @@ class K8sAndOssClientProcessesTests {
		.atMost(Duration.ofSeconds(10));

		// poll the helm Engine kubeconfig storage process
		TestHelpers.pollProcessInstances(registryStoreKubernetesInfoProcess, 1)
		TestHelpers.pollProcessActiveInstances(registryStoreKubernetesInfoProcess, 1)
		.await()
		.atMost(Duration.ofSeconds(10));

		@@ -486,10 +488,10 @@ class K8sAndOssClientProcessesTests {
		ServiceOrderState.IN_PROGRESS,
		k8sServiceOrderData);

		TestHelpers.pollProcessInstances(k8sServiceOrderProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(k8sServiceOrderProcess, 1).await().atMost(DURATION);

		// poll registry subprocess instance
		TestHelpers.pollProcessInstances(registryGetOrganizationCredentialsProcess, 1)
		TestHelpers.pollProcessActiveInstances(registryGetOrganizationCredentialsProcess, 1)
		.await()
		.atMost(Duration.ofSeconds(10));

		@@ -499,7 +501,7 @@ class K8sAndOssClientProcessesTests {
		TestHelpers.pollKafkaChannel(outgoingChannelToOssClientK8sOrder, 1).await().atMost(DURATION);

		// test that oss client k8s order subprocess has been generated
		TestHelpers.pollProcessInstances(ossClientOrderK8saasProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(ossClientOrderK8saasProcess, 1).await().atMost(DURATION);

		// retrieve the outgoing oss client messages
		List<? extends Message<String>> receivedToOutgoingChannelToOssClient =
		@@ -600,7 +602,7 @@ class K8sAndOssClientProcessesTests {
		.atMost(DURATION);

		// poll the helm Engine kubeconfig storage process
		TestHelpers.pollProcessInstances(registryStoreKubernetesInfoProcess, 1)
		TestHelpers.pollProcessActiveInstances(registryStoreKubernetesInfoProcess, 1)
		.await()
		.atMost(DURATION);

		@@ -645,7 +647,7 @@ class K8sAndOssClientProcessesTests {

		// poll in order to see that a k8s service termination subprocess has been generated
		// poll the helm Engine kubeconfig storage process
		TestHelpers.pollProcessInstances(k8sServiceTerminationSubprocess, 1)
		TestHelpers.pollProcessActiveInstances(k8sServiceTerminationSubprocess, 1)
		.await()
		.atMost(Duration.ofSeconds(10));

		@@ -686,7 +688,9 @@ class K8sAndOssClientProcessesTests {
		assertk8sServiceTerminationSubprocess();

		// assert Telemetry process
		TestHelpers.pollProcessInstances(serviceTelemetryPlatformService, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(serviceTelemetryPlatformService, 1)
		.await()
		.atMost(DURATION);
		}

		@Test
		@@ -728,12 +732,12 @@ class K8sAndOssClientProcessesTests {

		String processID = processInstance.id();

		TestHelpers.pollProcessInstances(ossServiceOrderProcess, 1)
		TestHelpers.pollProcessActiveInstances(ossServiceOrderProcess, 1)
		.await()
		.atMost(Duration.ofSeconds(10));

		// poll registry subprocess instance
		TestHelpers.pollProcessInstances(registryGetOrganizationCredentialsProcess, 1)
		TestHelpers.pollProcessActiveInstances(registryGetOrganizationCredentialsProcess, 1)
		.await()
		.atMost(Duration.ofSeconds(10));

		@@ -745,7 +749,7 @@ class K8sAndOssClientProcessesTests {
		.atMost(DURATION);

		// test that oss client k8s order subprocess has been generated
		TestHelpers.pollProcessInstances(ossClient5GOrderProcess, 1)
		TestHelpers.pollProcessActiveInstances(ossClient5GOrderProcess, 1)
		.await()
		.atMost(Duration.ofSeconds(10));

		@@ -800,7 +804,9 @@ class K8sAndOssClientProcessesTests {
		.atMost(DURATION);

		// assert Telemetry process
		TestHelpers.pollProcessInstances(serviceTelemetryPlatformService, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(serviceTelemetryPlatformService, 1)
		.await()
		.atMost(DURATION);
		}

		@Test
		@@ -925,7 +931,9 @@ class K8sAndOssClientProcessesTests {
		String stringAsJson = serviceTerminationMessageJson.toString();
		incomingChannelFromTmfKubernetesServiceTermination.send(stringAsJson);

		TestHelpers.pollProcessInstances(k8sServiceTerminationSubprocess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(k8sServiceTerminationSubprocess, 1)
		.await()
		.atMost(DURATION);

		ProcessInstance<? extends Model> k8sServiceTerminationSubprocessProcessInstance =
		getFirst(k8sServiceTerminationSubprocess.instances());
		@@ -992,12 +1000,14 @@ class K8sAndOssClientProcessesTests {
		String stringAsJson = serviceTerminationMessageJson.toString();
		incomingChannelFromTmfKubernetesServiceTermination.send(stringAsJson);

		TestHelpers.pollProcessInstances(k8sServiceTerminationSubprocess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(k8sServiceTerminationSubprocess, 1)
		.await()
		.atMost(DURATION);

		ProcessInstance<? extends Model> k8sServiceTerminationSubprocessProcessInstance =
		getFirst(k8sServiceTerminationSubprocess.instances());

		TestHelpers.pollProcessInstances(registryGetOrganizationCredentialsProcess, 1)
		TestHelpers.pollProcessActiveInstances(registryGetOrganizationCredentialsProcess, 1)
		.await()
		.atMost(DURATION);

		@@ -1074,14 +1084,18 @@ class K8sAndOssClientProcessesTests {
		incomingChannelFromTmfOssServiceTermination.send(stringAsJson);

		// poll registry subprocess instance
		TestHelpers.pollProcessInstances(registryGetOrganizationCredentialsProcess, 1)
		TestHelpers.pollProcessActiveInstances(registryGetOrganizationCredentialsProcess, 1)
		.await()
		.atMost(DURATION);

		assertRegistryCredentialsSubprocess();

		TestHelpers.pollProcessInstances(ossServiceTerminationSubprocess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessInstances(ossClientUpdateTerminationDate, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(ossServiceTerminationSubprocess, 1)
		.await()
		.atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(ossClientUpdateTerminationDate, 1)
		.await()
		.atMost(DURATION);
		}

		@ParameterizedTest
		@@ -1107,16 +1121,20 @@ class K8sAndOssClientProcessesTests {

		incomingChannelFromTmfOssServiceTermination.send(serviceTerminationMessageJson.toString());

		TestHelpers.pollProcessInstances(ossServiceTerminationSubprocess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(ossServiceTerminationSubprocess, 1)
		.await()
		.atMost(DURATION);

		// poll registry subprocess instance
		TestHelpers.pollProcessInstances(registryGetOrganizationCredentialsProcess, 1)
		TestHelpers.pollProcessActiveInstances(registryGetOrganizationCredentialsProcess, 1)
		.await()
		.atMost(DURATION);

		assertRegistryCredentialsSubprocess();

		TestHelpers.pollProcessInstances(ossClientUpdateTerminationDate, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(ossClientUpdateTerminationDate, 1)
		.await()
		.atMost(DURATION);

		assertOssClientFlow(serviceTerminationDataWithDate, ossClientStatus);

		@@ -1263,13 +1281,17 @@ class K8sAndOssClientProcessesTests {
		}

		private void assertk8sServiceTerminationSubprocess() {
		TestHelpers.pollProcessInstances(k8sServiceTerminationSubprocess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(k8sServiceTerminationSubprocess, 1)
		.await()
		.atMost(DURATION);
		}

		private void assertOssClientFlow(
		ServiceTerminationDataWithTimer serviceTerminationDataWithDate, Status ossClientStatus)
		throws JsonProcessingException, FileNotFoundException {
		TestHelpers.pollProcessInstances(ossClientUpdateTerminationDate, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(ossClientUpdateTerminationDate, 1)
		.await()
		.atMost(DURATION);

		TestHelpers.pollKafkaChannel(outgoingChannelOssClientUpdateTerminationDate, 1)
		.await()
		@@ -1314,7 +1336,7 @@ class K8sAndOssClientProcessesTests {

		private void assertFabricApiTearDownProcessFlow(Status fabricApiClientStatus)
		throws JsonProcessingException, FileNotFoundException {
		TestHelpers.pollProcessInstances(fabricApiServiceConnectionTeardownProcess, 1)
		TestHelpers.pollProcessActiveInstances(fabricApiServiceConnectionTeardownProcess, 1)
		.await()
		.atMost(DURATION);

src/test/java/sonata/integration/tests/ServiceDeploymentProcessTests.java

+7 −5

Changes for src/test/java/sonata/integration/tests/ServiceDeploymentProcessTests.java: 7 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -171,7 +171,7 @@ class ServiceDeploymentProcessTests {
		serviceOrderEndUserData);

		// POLL UNTIL THE HELM ENGINE DEPLOYMENT PROCESS HAS BEEN INITIATED
		TestHelpers.pollProcessInstances(helmEngineDeploymentProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(helmEngineDeploymentProcess, 1).await().atMost(DURATION);

		// get the process instance
		ProcessInstance<? extends Model> helmEngineDeploymentProcessInstance =
		@@ -191,7 +191,7 @@ class ServiceDeploymentProcessTests {
		helmEngineDeployment, serviceOrderEndUserData);

		// POLL UNTIL THE HELM ENGINE DEPLOYMENT PROCESS HAS BEEN INITIATED
		TestHelpers.pollProcessInstances(helmEngineDeploymentProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(helmEngineDeploymentProcess, 1).await().atMost(DURATION);

		// POLL KAFKA CHANNEL OUTGOING MESSAGE TO HELM ENGINE DEPLOYMENT
		TestHelpers.pollKafkaChannel(outgoingChannelToHelmEngineDeployment, 1).await().atMost(DURATION);
		@@ -239,15 +239,17 @@ class ServiceDeploymentProcessTests {

		if (helmEngineDeploymentStatus.equals(Status.SUCCESS)) {
		// ensure the service Telemetry Process instance creation
		TestHelpers.pollProcessInstances(serviceTelemetryProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(serviceTelemetryProcess, 1).await().atMost(DURATION);
		// ensure the service Termination Process instance creation
		TestHelpers.pollProcessInstances(endUserServiceTerminationSubprocess, 1)
		TestHelpers.pollProcessActiveInstances(endUserServiceTerminationSubprocess, 1)
		.await()
		.atMost(DURATION);

		if (requestedStartDate != null && requestedStartDate.isAfter(NOW)) {
		// ensure that the Service Activation Process has been created
		TestHelpers.pollProcessInstances(serviceOrderActivationProcess, 1).await().atMost(DURATION);
		TestHelpers.pollProcessActiveInstances(serviceOrderActivationProcess, 1)
		.await()
		.atMost(DURATION);
		}
		// poll channel for Service Order Item Update Message Status: COMPLETED
		TestHelpers.pollKafkaChannel(outgoingChannelForServiceOrderItemUpdate, 2)