Add CDS simulator to policy-models

Added CDS simulator to policy-models.  Added a test case to the CDS
Actor to verify that it works with the simulator.

Issue-ID: POLICY-2676
Change-Id: I9b10de3bde93c69e82df983f77eecc253de8a1a2
Signed-off-by: Jim Hahn <jrh3@att.com>

2 files changed, 74 insertions, 2 deletions

diff --git a/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/GrpcOperation.java b/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/GrpcOperation.java
index 820f4de34..0a882ce93 100644
--- a/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/GrpcOperation.java
+++ b/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/GrpcOperation.java
@@ -51,6 +51,7 @@ import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
 import org.onap.policy.controlloop.actorserviceprovider.Util;
 import org.onap.policy.controlloop.actorserviceprovider.impl.OperationPartial;
 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
 import org.onap.policy.controlloop.policy.TargetType;
 
 /**
@@ -202,13 +203,26 @@ public class GrpcOperation extends OperationPartial {
     @Override
     protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
 
+        /*
+         * construct the request first so that we don't have to clean up the "client" if
+         * an exception is thrown
+         */
+        ExecutionServiceInput request = constructRequest(params);
+
         CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
+
         client = new CdsProcessorGrpcClient(new CdsActorServiceManager(outcome, future),
                         config.getCdsServerProperties());
 
-        ExecutionServiceInput request = constructRequest(params);
         client.sendRequest(request);
-        return future;
+
+        // arrange to shutdown the client when the request completes
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+        controller.wrap(future).whenCompleteAsync(controller.delayedComplete(), params.getExecutor())
+                        .whenCompleteAsync((arg1, arg2) -> client.close(), getBlockingExecutor());
+
+        return controller;
     }
 
     /**
diff --git a/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/GrpcOperationTest.java b/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/GrpcOperationTest.java
index 9477a1585..fe8d42814 100644
--- a/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/GrpcOperationTest.java
+++ b/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/GrpcOperationTest.java
@@ -35,16 +35,20 @@ import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.AfterClass;
 import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.onap.aai.domain.yang.GenericVnf;
 import org.onap.aai.domain.yang.ServiceInstance;
 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
 import org.onap.policy.aai.AaiCqResponse;
 import org.onap.policy.cds.client.CdsProcessorGrpcClient;
 import org.onap.policy.cds.properties.CdsServerProperties;
@@ -63,6 +67,8 @@ import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOp
 import org.onap.policy.controlloop.policy.PolicyResult;
 import org.onap.policy.controlloop.policy.Target;
 import org.onap.policy.controlloop.policy.TargetType;
+import org.onap.policy.simulators.CdsSimulator;
+import org.onap.policy.simulators.Util;
 
 public class GrpcOperationTest {
     private static final String TARGET_ENTITY = "entity";
@@ -74,6 +80,14 @@ public class GrpcOperationTest {
     private static final UUID REQUEST_ID = UUID.randomUUID();
     private static final Coder coder = new StandardCoder();
 
+    protected static final Executor blockingExecutor = command -> {
+        Thread thread = new Thread(command);
+        thread.setDaemon(true);
+        thread.start();
+    };
+
+    private static CdsSimulator sim;
+
     @Mock
     private CdsProcessorGrpcClient cdsClient;
     private CdsServerProperties cdsProps;
@@ -82,6 +96,16 @@ public class GrpcOperationTest {
     private Target target;
     private GrpcOperation operation;
 
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+        sim = Util.buildCdsSim();
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() {
+        sim.stop();
+    }
+
     /**
      * Sets up the fields.
      */
@@ -112,6 +136,40 @@ public class GrpcOperationTest {
         target.setResourceID(RESOURCE_ID);
     }
 
+    /**
+     * Tests "success" case with simulator.
+     */
+    @Test
+    public void testSuccess() throws Exception {
+        ControlLoopEventContext context = new ControlLoopEventContext(onset);
+        loadCqData(context);
+
+        Map<String, Object> payload = Map.of("artifact_name", "my_artifact", "artifact_version", "1.0");
+
+        final ControlLoopOperationParams params = ControlLoopOperationParams.builder()
+                        .actor(CdsActorConstants.CDS_ACTOR).operation("subscribe").context(context)
+                        .actorService(new ActorService()).targetEntity(TARGET_ENTITY).target(target).retry(0)
+                        .timeoutSec(5).executor(blockingExecutor).payload(payload).build();
+
+        cdsProps.setHost("localhost");
+        cdsProps.setPort(sim.getPort());
+        cdsProps.setTimeout(3);
+
+        GrpcConfig config = new GrpcConfig(blockingExecutor, cdsProps);
+
+        operation = new GrpcOperation(params, config) {
+            @Override
+            protected CompletableFuture<OperationOutcome> startGuardAsync() {
+                // indicate that guard completed successfully
+                return CompletableFuture.completedFuture(params.makeOutcome());
+            }
+        };
+
+        OperationOutcome outcome = operation.start().get();
+        assertEquals(PolicyResult.SUCCESS, outcome.getResult());
+        assertTrue(outcome.getResponse() instanceof ExecutionServiceOutput);
+    }
+
     @Test
     public void testStartPreprocessorAsync() throws InterruptedException, ExecutionException, TimeoutException {