Integrate cds actor service provider

* Add request enrichment with aai service instance id and vnf id
* Implement cds grpc request
* Implement VfwControlLoopCdsTest

Issue-ID: POLICY-2088
Signed-off-by: Bruno Sakoto <bruno.sakoto@bell.ca>
Change-Id: Ib44d447d6a3a70ff800a5760032b676fdfa32d9c

8 files changed, 490 insertions, 1 deletions

diff --git a/controlloop/templates/archetype-cl-amsterdam/src/main/resources/archetype-resources/src/main/resources/__closedLoopControlName__.drl b/controlloop/templates/archetype-cl-amsterdam/src/main/resources/archetype-resources/src/main/resources/__closedLoopControlName__.drl
index 57ebd7089..4cbdd49e5 100644
--- a/controlloop/templates/archetype-cl-amsterdam/src/main/resources/archetype-resources/src/main/resources/__closedLoopControlName__.drl
+++ b/controlloop/templates/archetype-cl-amsterdam/src/main/resources/archetype-resources/src/main/resources/__closedLoopControlName__.drl
@@ -3,6 +3,7 @@
  * ONAP
  * ================================================================================
  * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2019 Bell Canada.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,6 +21,8 @@
 
 package org.onap.policy.controlloop;
 
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
+
 import org.onap.policy.controlloop.VirtualControlLoopEvent;
 import org.onap.policy.controlloop.VirtualControlLoopNotification;
 import org.onap.policy.controlloop.ControlLoopEventStatus;
@@ -32,6 +35,9 @@ import org.onap.policy.controlloop.eventmanager.ControlLoopEventManager;
 import org.onap.policy.controlloop.eventmanager.ControlLoopEventManager.NewEventStatus;
 import org.onap.policy.controlloop.eventmanager.ControlLoopOperationManager;
 import org.onap.policy.controlloop.actor.so.SoActorServiceProvider;
+import org.onap.policy.controlloop.actor.cds.CdsActorServiceProvider;
+import org.onap.policy.controlloop.actor.cds.CdsActorServiceProvider.CdsActorServiceManager;
+import org.onap.policy.controlloop.actor.cds.constants.CdsActorConstants;
 import org.onap.policy.aai.AaiNqResponseWrapper;
 import org.onap.policy.aai.AaiCqResponse;
 import org.onap.policy.appc.Request;
@@ -42,6 +48,9 @@ import org.onap.policy.appclcm.LcmResponseWrapper;
 import org.onap.policy.appclcm.LcmRequest;
 import org.onap.policy.appclcm.LcmResponse;
 import org.onap.policy.appclcm.LcmCommonHeader;
+import org.onap.policy.cds.CdsResponse;
+import org.onap.policy.cds.client.CdsProcessorGrpcClient;
+import org.onap.policy.cds.properties.CdsServerProperties;
 import org.onap.policy.sdnr.PciRequestWrapper;
 import org.onap.policy.sdnr.PciResponseWrapper;
 import org.onap.policy.sdnr.PciRequest;
@@ -636,6 +645,34 @@ rule "${policyName}.EVENT.MANAGER.OPERATION.LOCKED.GUARD_PERMITTED"
                       PolicyEngineConstants.getManager().deliver("SDNR-CL", request);
                   }
                   break;
+
+             case "CDS":
+
+                  if(request instanceof ExecutionServiceInput) {
+
+                      // Instantiate cds actor, service manager and grpc properties
+
+                      CdsActorServiceProvider cdsActor = new CdsActorServiceProvider();
+                      CdsActorServiceProvider.CdsActorServiceManager cdsActorServiceManager = cdsActor.new CdsActorServiceManager();
+
+                      CdsServerProperties props = new CdsServerProperties();
+                      props.setHost(PolicyEngineConstants.getManager().getEnvironmentProperty("cds.grpcHost"));
+                      props.setPort(Integer.parseInt(PolicyEngineConstants.getManager().getEnvironmentProperty("cds.grpcPort")));
+                      props.setUsername(PolicyEngineConstants.getManager().getEnvironmentProperty("cds.grpcUsername"));
+                      props.setPassword(PolicyEngineConstants.getManager().getEnvironmentProperty("cds.grpcPassword"));
+                      props.setTimeout(Integer.parseInt(PolicyEngineConstants.getManager().getEnvironmentProperty("cds.grpcTimeout")));
+
+                      // Send cds grpc request
+                      try (CdsProcessorGrpcClient client = new CdsProcessorGrpcClient(cdsActorServiceManager, props)) {
+                          CdsResponse response =
+                                cdsActorServiceManager.sendRequestToCds(client, props, (ExecutionServiceInput) request);
+                          logger.info("CDS response: {}", response);
+                          insert(response);
+                      }
+
+                  }
+                  break;
+
           }
         } else {
           //
@@ -1429,6 +1466,86 @@ rule "${policyName}.SDNC.RESPONSE"
 
 end
 
+
+/**
+ * This rule responds to CDS Response Events
+ */
+rule "${policyName}.CDS.RESPONSE"
+    when
+        $params : Params( getClosedLoopControlName() == "${closedLoopControlName}" )
+        $event : VirtualControlLoopEvent( closedLoopControlName == $params.getClosedLoopControlName(),
+            closedLoopEventStatus == ControlLoopEventStatus.ONSET )
+        $manager : ControlLoopEventManager( closedLoopControlName == $event.getClosedLoopControlName(),
+            requestId == $event.getRequestId() )
+        $operation : ControlLoopOperationManager( onset.closedLoopControlName == $event.getClosedLoopControlName(),
+            onset.getRequestId() == $event.getRequestId() )
+        $opTimer : ControlLoopTimer( closedLoopControlName == $event.getClosedLoopControlName(),
+            requestId == $event.getRequestId().toString(), timerType == "Operation", !expired )
+        $lock : TargetLock (requestId == $event.getRequestId())
+        $response : CdsResponse( requestId == $event.getRequestId().toString() )
+
+    then
+
+        Logger logger = LoggerFactory.getLogger(drools.getRule().getPackage());
+        logger.info("{}: {}", $params.getClosedLoopControlName(), drools.getRule().getName());
+        logger.debug("{}: {}: event={} manager={} operation={} lock={} opTimer={} response={}",
+                    $params.getClosedLoopControlName(), drools.getRule().getName(),
+                    $event, $manager, $operation, $lock, $operation, $opTimer, $response);
+
+        // Get the result of the operation
+        PolicyResult policyResult = $operation.onResponse($response);
+
+        if (policyResult != null) {
+            logger.debug("{}: {}: operation finished - result={}",
+                    $params.getClosedLoopControlName(), drools.getRule().getName(),
+                    policyResult);
+
+            // The operation has completed, construct a notification showing our results
+            VirtualControlLoopNotification notification = new VirtualControlLoopNotification($event);
+            notification.setFrom("policy");
+            notification.setPolicyName(drools.getRule().getName());
+            notification.setPolicyScope("${policyScope}");
+            notification.setPolicyVersion("${policyVersion}");
+            notification.setMessage($operation.getOperationHistory());
+            notification.setHistory($operation.getHistory());
+            notification.setNotification(
+                ($response != null && CdsActorConstants.SUCCESS.equals($response.getStatus()))
+                    ? ControlLoopNotificationType.OPERATION_SUCCESS : ControlLoopNotificationType.OPERATION_FAILURE);
+
+            // Send the notification
+            PolicyEngineConstants.getManager().deliver("POLICY-CL-MGT", notification);
+
+            // Ensure the operation is complete
+            if ($operation.isOperationComplete()) {
+
+                // It is complete, remove it from memory
+                retract($operation);
+
+                // We must also retract the timer object
+                // NOTE: We could write a Rule to do this
+                retract($opTimer);
+
+                // Complete the operation
+                modify($manager) {finishOperation($operation)};
+
+            } else {
+                // Just doing this will kick off the LOCKED rule again
+                modify($operation) {};
+            }
+        } else {
+            // Its not finished yet (i.e. expecting more Response objects)
+            // Or possibly it is a leftover response that we timed the request out previously
+            logger.info(
+                "policyResult is null"
+                    + "\nIt's not finished yet (i.e. expecting more Response objects)"
+                    + "\nOr possibly it is a leftover response that we timed the request out previously");
+        }
+
+        // We are going to retract these objects from memory
+        retract($response);
+
+end
+
 /*
 *
 * This manages a single timer.
diff --git a/controlloop/templates/template.demo/pom.xml b/controlloop/templates/template.demo/pom.xml
index 548bb76b3..57edba6f0 100644
--- a/controlloop/templates/template.demo/pom.xml
+++ b/controlloop/templates/template.demo/pom.xml
@@ -4,6 +4,7 @@
   ================================================================================
   Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
   Modifications Copyright (C) 2019 Nordix Foundation.
+  Modifications Copyright (C) 2019 Bell Canada.
   ================================================================================
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
@@ -30,6 +31,11 @@
 
   <artifactId>template.demo</artifactId>
 
+  <properties>
+    <protobuf.version>3.6.1</protobuf.version>
+    <grpc.version>1.17.1</grpc.version>
+  </properties>
+
   <dependencies>
     <dependency>
       <groupId>org.eclipse.persistence</groupId>
@@ -185,6 +191,28 @@
       <scope>provided</scope>
     </dependency>
     <dependency>
+      <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
+      <artifactId>actor.cds</artifactId>
+      <version>${policy.models.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.onap.ccsdk.cds.components</groupId>
+      <artifactId>proto-definition</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
+      <artifactId>cds</artifactId>
+      <version>${policy.models.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+      <version>${protobuf.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
@@ -207,6 +235,12 @@
       <scope>test</scope>
     </dependency>
     <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-testing</artifactId>
+      <version>${grpc.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
       <groupId>org.onap.policy.drools-pdp</groupId>
       <artifactId>policy-management</artifactId>
       <version>${version.policy.drools-pdp}</version>
diff --git a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopBase.java b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopBase.java
index ab6ec5a3a..5817dc930 100644
--- a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopBase.java
+++ b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopBase.java
@@ -3,6 +3,7 @@
  * demo
  * ================================================================================
  * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2019 Bell Canada.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -84,6 +85,7 @@ public class ControlLoopBase {
         SupportUtil.setSoProps();
         SupportUtil.setVfcProps();
         SupportUtil.setPuProp();
+        SupportUtil.setCdsProps();
 
         LoggerUtil.setLevel(LoggerUtil.ROOT_LOGGER, "INFO");
 
diff --git a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopFailureTest.java b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopFailureTest.java
index bc5a9828b..0793b8825 100644
--- a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopFailureTest.java
+++ b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/ControlLoopFailureTest.java
@@ -252,6 +252,7 @@ public class ControlLoopFailureTest extends ControlLoopBase implements TopicList
         event.setClosedLoopAlarmStart(Instant.now());
         event.setAai(new HashMap<>());
         event.getAai().put("generic-vnf.vnf-id", target);
+        event.getAai().put("vserver.vserver-name", "OzVServer");
         event.setClosedLoopEventStatus(status);
         kieSession.insert(event);
     }
diff --git a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/SupportUtil.java b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/SupportUtil.java
index ab57fdf86..878f94be1 100644
--- a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/SupportUtil.java
+++ b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/SupportUtil.java
@@ -3,6 +3,7 @@
  * demo
  * ================================================================================
  * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2019 Bell Canada.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -44,6 +45,7 @@ import org.kie.api.builder.Results;
 import org.kie.api.runtime.KieContainer;
 import org.kie.api.runtime.KieSession;
 import org.onap.policy.common.endpoints.http.server.HttpServletServer;
+import org.onap.policy.common.utils.network.NetworkUtil;
 import org.onap.policy.controlloop.policy.ControlLoopPolicy;
 import org.onap.policy.controlloop.policy.guard.ControlLoopGuard;
 import org.onap.policy.drools.system.PolicyEngineConstants;
@@ -58,6 +60,16 @@ public final class SupportUtil {
     private static final String OPSHISTPUPROP = "OperationsHistoryPU";
     private static final Logger logger = LoggerFactory.getLogger(SupportUtil.class);
 
+    static final int GRPC_SERVER_PORT;
+
+    static {
+        try {
+            GRPC_SERVER_PORT = NetworkUtil.allocPort();
+        } catch (IOException e) {
+            throw new RuntimeException("Socket cannot be created for grpc server port", e);
+        }
+    }
+
     public static class Pair<A, B> {
         public final A first;
         public final B second;
@@ -370,6 +382,17 @@ public final class SupportUtil {
     }
 
     /**
+     * Set cds properties.
+     */
+    public static void setCdsProps() {
+        PolicyEngineConstants.getManager().setEnvironmentProperty("cds.grpcHost", "localhost");
+        PolicyEngineConstants.getManager().setEnvironmentProperty("cds.grpcPort", Integer.toString(GRPC_SERVER_PORT));
+        PolicyEngineConstants.getManager().setEnvironmentProperty("cds.grpcUsername", "grpc-username");
+        PolicyEngineConstants.getManager().setEnvironmentProperty("cds.grpcPassword", "grpc-password");
+        PolicyEngineConstants.getManager().setEnvironmentProperty("cds.grpcTimeout", "5");
+    }
+
+    /**
      * Rule specification.
      */
     public static class RuleSpec {
diff --git a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/VcpeControlLoopTest.java b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/VcpeControlLoopTest.java
index 1cd431f50..25bef2ebb 100644
--- a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/VcpeControlLoopTest.java
+++ b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/VcpeControlLoopTest.java
@@ -58,6 +58,7 @@ public class VcpeControlLoopTest extends ControlLoopBase implements TopicListene
             "service=ServiceDemo;resource=Res1Demo;type=operational",
             "CL_vCPE",
             "org.onap.closed_loop.ServiceDemo:VNFS:1.0.0");
+        SupportUtil.setCustomQuery("false");
     }
 
     @Test
@@ -122,7 +123,6 @@ public class VcpeControlLoopTest extends ControlLoopBase implements TopicListene
          */
         sendEvent(pair.first, requestId, ControlLoopEventStatus.ONSET, "getFail", false);
 
-
         kieSession.fireUntilHalt();
 
         // allow object clean-up
diff --git a/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/VfwControlLoopCdsTest.java b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/VfwControlLoopCdsTest.java
new file mode 100644
index 000000000..435faf25d
--- /dev/null
+++ b/controlloop/templates/template.demo/src/test/java/org/onap/policy/template/demo/VfwControlLoopCdsTest.java
@@ -0,0 +1,266 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 Bell Canada.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.template.demo;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.stub.StreamObserver;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader;
+import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType;
+import org.onap.ccsdk.cds.controllerblueprints.common.api.Status;
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc;
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.TopicListener;
+import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.onap.policy.controlloop.ControlLoopEventStatus;
+import org.onap.policy.controlloop.ControlLoopNotificationType;
+import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.VirtualControlLoopNotification;
+import org.onap.policy.controlloop.policy.ControlLoopPolicy;
+import org.onap.policy.controlloop.util.Serialization;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test class for vfirewall use case using CDS actor.
+ */
+public class VfwControlLoopCdsTest extends ControlLoopBase implements TopicListener {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(VfwControlLoopCdsTest.class);
+
+    private final AtomicReference<StreamObserver<ExecutionServiceOutput>> responseObserverRef = new AtomicReference<>();
+    private Server server;
+
+    /**
+     * Setup the simulator.
+     */
+    @BeforeClass
+    public static void setUpBeforeClass() {
+        ControlLoopBase.setUpBeforeClass("../archetype-cl-amsterdam/src/main/resources/archetype-resources/src/"
+                                                 + "main/resources/__closedLoopControlName__.drl",
+                "src/test/resources/yaml/policy_ControlLoop_vFW_CDS.yaml",
+                "service=ServiceDemo;resource=Res1Demo;type=operational", "CL_vFW",
+                "org.onap.closed_loop.ServiceDemo:VNFS:1.0.0");
+        SupportUtil.setCustomQuery("true");
+    }
+
+    @Before
+    public void setUp() throws IOException {
+        this.startGrpcServer();
+    }
+
+    @After
+    public void tearDown() {
+        this.stopGrpcServer();
+    }
+
+    private void startGrpcServer() throws IOException {
+
+        BluePrintProcessingServiceGrpc.BluePrintProcessingServiceImplBase cdsBlueprintServerImpl =
+
+            new BluePrintProcessingServiceGrpc.BluePrintProcessingServiceImplBase() {
+
+                @Override
+                public StreamObserver<ExecutionServiceInput> process(
+                        final StreamObserver<ExecutionServiceOutput> responseObserver) {
+
+                    responseObserverRef.set(responseObserver);
+
+                    return new StreamObserver<ExecutionServiceInput>() {
+                        @Override
+                        public void onNext(final ExecutionServiceInput input) {
+                            LOGGER.info("gRPC server onNext() for input: {} ...", input);
+                            ExecutionServiceOutput output =
+                                    ExecutionServiceOutput.newBuilder()
+                                            .setCommonHeader(
+                                                    CommonHeader.newBuilder().setRequestId(
+                                                            input.getCommonHeader().getRequestId()).build())
+                                            .setStatus(
+                                                    Status.newBuilder().setEventType(
+                                                            EventType.EVENT_COMPONENT_EXECUTED).build())
+                                            .build();
+                            responseObserver.onNext(output);
+                        }
+
+                        @Override
+                        public void onError(final Throwable throwable) {
+                            LOGGER.error("gRPC server onError() for throwable: {} ...", throwable);
+                        }
+
+                        @Override
+                        public void onCompleted() {
+                            LOGGER.info("gRPC server onCompleted() ...");
+                            responseObserver.onCompleted();
+                        }
+                    };
+                }
+            };
+
+        server = ServerBuilder.forPort(SupportUtil.GRPC_SERVER_PORT).addService(cdsBlueprintServerImpl).build().start();
+        LOGGER.info("gRPC server is listening for CDS requests on port {}", SupportUtil.GRPC_SERVER_PORT);
+
+    }
+
+    private void stopGrpcServer() {
+        if (server != null) {
+            this.server.shutdown();
+            LOGGER.info("gRPC server handling CDS requests has been successfully shut down.");
+        }
+    }
+
+    @Test
+    public void successTest() {
+
+        /*
+         * Allows the PolicyEngine to callback to this object to notify that there is an event ready
+         * to be pulled from the queue
+         */
+        for (TopicSink sink : noopTopics) {
+            assertTrue(sink.start());
+            sink.register(this);
+        }
+
+        /*
+         * Create a unique requestId
+         */
+        requestId = UUID.randomUUID();
+
+        /*
+         * Simulate an onset event the policy engine will receive from DCAE to kick off processing
+         * through the rules
+         */
+        sendEvent(pair.first, requestId, ControlLoopEventStatus.ONSET);
+
+        kieSession.fireUntilHalt();
+        kieSession.fireAllRules();
+
+        /*
+         * The only fact in memory should be Params
+         */
+        assertEquals(1, kieSession.getFactCount());
+
+        /*
+         * Print what's left in memory
+         */
+        dumpFacts(kieSession);
+    }
+
+
+    /*
+     * @see org.onap.policy.drools.PolicyEngineListener#newEventNotification(java.lang.String)
+     */
+    @Override
+    public void onTopicEvent(CommInfrastructure commType, String topic, String event) {
+        /*
+         * Pull the object that was sent out to DMAAP and make sure it is a ControlLoopNotification
+         * of type active
+         */
+        assertEquals("POLICY-CL-MGT", topic);
+        VirtualControlLoopNotification notification =
+                Serialization.gsonJunit.fromJson(event, VirtualControlLoopNotification.class);
+        assertNotNull(notification);
+        String policyName = notification.getPolicyName();
+        if (policyName.endsWith("EVENT")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            assertEquals(ControlLoopNotificationType.ACTIVE, notification.getNotification());
+        } else if (policyName.endsWith("GUARD_NOT_YET_QUERIED")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            assertEquals(ControlLoopNotificationType.OPERATION, notification.getNotification());
+            assertNotNull(notification.getMessage());
+            assertTrue(notification.getMessage().startsWith("Sending guard query"));
+        } else if (policyName.endsWith("GUARD.RESPONSE")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            assertEquals(ControlLoopNotificationType.OPERATION, notification.getNotification());
+            assertNotNull(notification.getMessage());
+            assertTrue(notification.getMessage().toLowerCase().endsWith("permit"));
+        } else if (policyName.endsWith("GUARD_PERMITTED")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            assertEquals(ControlLoopNotificationType.OPERATION, notification.getNotification());
+            assertNotNull(notification.getMessage());
+            assertTrue(notification.getMessage().startsWith("actor=CDS"));
+        } else if (policyName.endsWith("OPERATION.TIMEOUT")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            kieSession.halt();
+            logger.debug("The operation timed out");
+            fail("Operation Timed Out");
+        } else if (policyName.endsWith("CDS.RESPONSE")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            assertEquals(ControlLoopNotificationType.OPERATION_SUCCESS, notification.getNotification());
+            assertNotNull(notification.getMessage());
+            assertTrue(notification.getMessage().startsWith("actor=CDS"));
+            sendEvent(pair.first, requestId, ControlLoopEventStatus.ABATED);
+        } else if (policyName.endsWith("EVENT.MANAGER")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            if ("error".equals(notification.getAai().get("generic-vnf.vnf-name"))) {
+                assertEquals(ControlLoopNotificationType.FINAL_FAILURE, notification.getNotification());
+                assertEquals("Target vnf-id could not be found", notification.getMessage());
+            } else if ("getFail".equals(notification.getAai().get("generic-vnf.vnf-name"))) {
+                assertEquals(ControlLoopNotificationType.FINAL_FAILURE, notification.getNotification());
+            } else {
+                assertEquals(ControlLoopNotificationType.FINAL_SUCCESS, notification.getNotification());
+            }
+            kieSession.halt();
+        } else if (policyName.endsWith("EVENT.MANAGER.TIMEOUT")) {
+            logger.debug("Rule Fired: " + notification.getPolicyName());
+            kieSession.halt();
+            logger.debug("The control loop timed out");
+            fail("Control Loop Timed Out");
+        }
+    }
+
+    /**
+     * This method is used to simulate event messages from DCAE that start the control loop (onset
+     * message) or end the control loop (abatement message).
+     *
+     * @param policy the controlLoopName comes from the policy
+     * @param requestId the requestId for this event
+     * @param status could be onset or abated
+     */
+    private void sendEvent(ControlLoopPolicy policy, UUID requestId, ControlLoopEventStatus status) {
+        VirtualControlLoopEvent event = new VirtualControlLoopEvent();
+        event.setClosedLoopControlName(policy.getControlLoop().getControlLoopName());
+        event.setRequestId(requestId);
+        event.setTarget("generic-vnf.vnf-name");
+        event.setClosedLoopAlarmStart(Instant.now());
+        event.setAai(new HashMap<>());
+        event.getAai().put("generic-vnf.vnf-name", "testGenericVnfID");
+        event.getAai().put("vserver.vserver-name", "OzVServer");
+        event.setClosedLoopEventStatus(status);
+        kieSession.insert(event);
+    }
+
+}
diff --git a/controlloop/templates/template.demo/src/test/resources/yaml/policy_ControlLoop_vFW_CDS.yaml b/controlloop/templates/template.demo/src/test/resources/yaml/policy_ControlLoop_vFW_CDS.yaml
new file mode 100644
index 000000000..511a9694b
--- /dev/null
+++ b/controlloop/templates/template.demo/src/test/resources/yaml/policy_ControlLoop_vFW_CDS.yaml
@@ -0,0 +1,46 @@
+# Copyright (C) 2019 Bell Canada.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+controlLoop:
+  version: 2.0.0
+  controlLoopName: ControlLoop-vFirewall-d0a1dfc6-94f5-4fd4-a5b5-4630b438850a
+  services:
+    - serviceInvariantUUID: 5cfe6f4a-41bc-4247-8674-ebd4b98e35cc
+      serviceUUID: 0f40bba5-986e-4b3c-803f-ddd1b7b25f24
+      serviceName: 57e66ea7-0ed6-45c7-970f
+  trigger_policy: unique-policy-id-1-modifyConfig
+  timeout: 60
+  abatement: true
+
+policies:
+  - id: unique-policy-id-1-modifyConfig
+    name: modify packet gen config
+    description:
+    actor: CDS
+    recipe: ModifyConfig
+    target:
+      resourceID: bbb3cefd-01c8-413c-9bdd-2b92f9ca3d38
+      type: VNF
+    payload:
+      artifact_name: vfw-cds
+      artifact_version: 1.0.0
+      mode: async
+      data: '{"mapInfo":{"key":"val"},"arrayInfo":["one","two"],"paramInfo":"val"}'
+    retry: 0
+    timeout: 30
+    success: final_success
+    failure: final_failure
+    failure_timeout: final_failure_timeout
+    failure_retries: final_failure_retries
+    failure_exception: final_failure_exception
+    failure_guard: final_failure_guard