Adding initial cut virtual fire-alarm policy publish scenario

modules/integration/tests-common/integration-common/pom.xml

@@ -140,6 +140,10 @@
             <artifactId>org.wso2.carbon.integration.common.utils</artifactId>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

modules/integration/tests-common/integration-common/src/main/java/org/wso2/iot/integration/common/Constants.java

@@ -451,6 +451,9 @@ public final class Constants {
         public static final String PAYLOAD_FILE = "virtual-fire-alarm-enrollment-payloads.json";
         public static final String ANALYTICS_ARTIFACTS_DEPLOYMENT_ENDPOINT =
                 "/api/device-mgt/v1.0/admin/devicetype/1.0.0/deploy/virtual_firealarm";
+        public static final String POLICY_DATA = "POLICY_DATA";
+        public static final String ACTIVATE_POLICY_ENDPOINT = "/api/device-mgt/v1.0/policies/activate-policy";
+        public static final String APPLY_CHANGES_ENDPOINT = "/api/device-mgt/v1.0/policies/apply-changes";
 
     }
 }

modules/integration/tests-common/integration-common/src/main/java/org/wso2/iot/integration/common/MqttSubscriberClient.java

@@ -0,0 +1,57 @@
+package org.wso2.iot.integration.common;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+
+import java.util.ArrayList;
+
+/**
+ * This is a mqtt subscriber client to receive .
+ */
+public class MqttSubscriberClient implements MqttCallback {
+    private static Log log = LogFactory.getLog(MqttSubscriberClient.class);
+    private ArrayList<MqttMessage> mqttMessages;
+
+    @Override
+    public void connectionLost(Throwable throwable) {
+        // Not implemented
+    }
+
+    @Override
+    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
+        mqttMessages.add(mqttMessage);
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+        // Not implemented
+    }
+
+    /**
+     *
+     * @param serverAddress Mqtt broker address
+     * @param clientId Client ID
+     * @param topicName Topic Name
+     * @throws MqttException Mqtt Exception
+     */
+    public MqttSubscriberClient(String serverAddress, String clientId, String topicName, String accessToken) throws
+            MqttException {
+        mqttMessages = new ArrayList<>();
+        MemoryPersistence persistence = new MemoryPersistence();
+        MqttClient client = new MqttClient(serverAddress, clientId, persistence);
+        MqttConnectOptions connectOptions = new MqttConnectOptions();
+        client.setCallback(this);
+        connectOptions.setUserName(accessToken);
+        connectOptions.setPassword("".toCharArray());
+        connectOptions.setCleanSession(true);
+        client.connect(connectOptions);
+        client.subscribe(topicName);
+    }
+
+    public ArrayList<MqttMessage> getMqttMessages() {
+        return mqttMessages;
+    }
+
+}

modules/integration/tests-integration/src/test/java/org/wso2/iot/integration/samples/VirtualFireAlarmTestCase.java

@@ -21,6 +21,7 @@ package org.wso2.iot.integration.samples;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
+import com.google.gson.JsonPrimitive;
 import junit.framework.Assert;
 import org.apache.commons.httpclient.HttpStatus;
 import org.apache.commons.logging.Log;
@@ -35,10 +36,13 @@ import org.testng.annotations.Test;
 import org.wso2.carbon.automation.engine.context.TestUserMode;
 import org.wso2.carbon.automation.test.utils.http.client.HttpResponse;
 import org.wso2.iot.integration.common.Constants;
+import org.wso2.iot.integration.common.MqttSubscriberClient;
 import org.wso2.iot.integration.common.PayloadGenerator;
 import org.wso2.iot.integration.common.RestClient;
 import org.wso2.iot.integration.common.TestBase;
 
+import java.util.ArrayList;
+
 /**
  * This class tests the functionality of the virtual fire alarm.
  */
@@ -46,7 +50,12 @@ public class VirtualFireAlarmTestCase extends TestBase {
     private static Log log = LogFactory.getLog(VirtualFireAlarmTestCase.class);
     public static String deviceId1;
     public static String deviceId2;
+    public static String tenantDeviceId1;
+    public static String tenantDeviceId2;
     public static long currentTime;
+    private String broker = "tcp://localhost:1886";
+    private String DEVICE_TYPE = "virtual_firealarm";
+    private RestClient restClient;
 
     @Factory(dataProvider = "userModeProvider")
     public VirtualFireAlarmTestCase(TestUserMode userMode) {
@@ -55,10 +64,10 @@ public class VirtualFireAlarmTestCase extends TestBase {
 
     @BeforeClass(alwaysRun = true)
     public void initTest() throws Exception {
-        super.init(userMode);;
+        super.init(userMode);
-        RestClient client = new RestClient(backendHTTPSURL, Constants.APPLICATION_JSON, accessTokenString);
+        restClient = new RestClient(backendHTTPSURL, Constants.APPLICATION_JSON, accessTokenString);
         if (userMode == TestUserMode.TENANT_ADMIN) {
-            HttpResponse response = client
+            HttpResponse response = restClient
                     .post(Constants.VirtualFireAlarmConstants.ANALYTICS_ARTIFACTS_DEPLOYMENT_ENDPOINT, "");
             Assert.assertEquals(HttpStatus.SC_CREATED, response.getResponseCode());
             // Time for deploying the carbon apps
@@ -85,20 +94,24 @@ public class VirtualFireAlarmTestCase extends TestBase {
         JsonArray jsonArray = new JsonParser().parse(response.getData()).getAsJsonObject().getAsJsonArray("devices");
         Assert.assertEquals("Virtual fire alarm enrollment failed ", 2, jsonArray.size());
 
+        if (userMode != TestUserMode.TENANT_ADMIN) {
             deviceId1 = jsonArray.get(0).getAsJsonObject().getAsJsonPrimitive("deviceIdentifier").getAsString();
             deviceId2 = jsonArray.get(1).getAsJsonObject().getAsJsonPrimitive("deviceIdentifier").getAsString();
+        } else {
+            tenantDeviceId1 = jsonArray.get(0).getAsJsonObject().getAsJsonPrimitive("deviceIdentifier").getAsString();
+            tenantDeviceId2 = jsonArray.get(1).getAsJsonObject().getAsJsonPrimitive("deviceIdentifier").getAsString();
+        }
     }
 
     @Test(description = "Test whether the publishing to a mqtt broker works fine without exceptions", dependsOnMethods =
             {"testEnrollment"} )
     public void testEventPublishing() throws Exception {
-        String DEVICE_TYPE = "virtual_firealarm";
+        String deviceId1 = userMode == TestUserMode.TENANT_ADMIN ? tenantDeviceId1 : VirtualFireAlarmTestCase.deviceId1;
-
+        String deviceId2 = userMode == TestUserMode.TENANT_ADMIN ? tenantDeviceId2 : VirtualFireAlarmTestCase.deviceId2;
         // Publishing message as a device with simple agent (device 1)
         String topic = automationContext.getContextTenant().getDomain() + "/" + DEVICE_TYPE + "/" + deviceId1 +
                 "/temperature";
         int qos = 2;
-        String broker = "tcp://localhost:1886";
         String clientId = deviceId1 + ":" + DEVICE_TYPE;
         MemoryPersistence persistence = new MemoryPersistence();
         MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
@@ -154,4 +167,35 @@ public class VirtualFireAlarmTestCase extends TestBase {
         log.info("Mqtt Client is Disconnected");
         currentTime = System.currentTimeMillis();
     }
+
+    @Test(description = "Test whether the policy publishing from the server to device works", dependsOnMethods =
+            {"testEnrollment"} )
+    public void testPolicyPublishing() throws Exception {
+        String deviceId2 = userMode == TestUserMode.TENANT_ADMIN ? tenantDeviceId2 : VirtualFireAlarmTestCase.deviceId2;
+        String topic = automationContext.getContextTenant().getDomain() + "/" + DEVICE_TYPE + "/" + deviceId2 + "/#";
+        String clientId = deviceId2 + ":" + DEVICE_TYPE;
+        HttpResponse response = restClient.post("/api/device-mgt/v1.0/policies", PayloadGenerator
+                .getJsonPayload(Constants.VirtualFireAlarmConstants.PAYLOAD_FILE,
+                        Constants.VirtualFireAlarmConstants.POLICY_DATA).toString());
+        Assert.assertEquals("Policy creation for virtual fire alarm failed", HttpStatus.SC_CREATED,
+                response.getResponseCode());
+        JsonObject jsonObject = new JsonParser().parse(response.getData()).getAsJsonObject();
+        String policyId = jsonObject.getAsJsonPrimitive("id").getAsString();
+        JsonArray jsonArray = new JsonArray();
+        jsonArray.add(new JsonPrimitive(policyId));
+        response = restClient.post(Constants.VirtualFireAlarmConstants.ACTIVATE_POLICY_ENDPOINT,
+                jsonArray.toString());
+        Assert.assertEquals("Policy activation for virtual fire alarm failed", HttpStatus.SC_OK,
+                response.getResponseCode());
+        MqttSubscriberClient mqttSubscriberClient = new MqttSubscriberClient(broker, clientId, topic, accessToken);
+        response = restClient.put(Constants.VirtualFireAlarmConstants.APPLY_CHANGES_ENDPOINT, "");
+        Assert.assertEquals("Applying changes to policy for virtual fire alarm failed", HttpStatus.SC_OK,
+                response.getResponseCode());
+        // Allow some time for message delivery
+        Thread.sleep(20000);
+        ArrayList<MqttMessage> mqttMessages = mqttSubscriberClient.getMqttMessages();
+        Assert.assertEquals("Policy published message is not received by the mqtt listener. ", 1, mqttMessages.size());
+
+
+    }
 }

modules/integration/tests-integration/src/test/resources/payloads/virtual-fire-alarm-enrollment-payloads.json

@@ -10,5 +10,21 @@
         "temperature": 30
       }
     }
+  },
+  "POLICY_DATA" : {
+     "policyName":"testpolicy",
+    "description":"","compliance":"enforce","ownershipType":null,"active":true,
+    "profile":{
+      "profileName":"dvdv",
+      "deviceType":"virtual_firealarm",
+      "profileFeaturesList":[{
+        "featureCode":"CONFIG",
+        "deviceType":"virtual_firealarm",
+        "content":{
+            "policyDefinition":"define stream fireAlarmEventStream (deviceID string, temp int);\nfrom fireAlarmEventStream#window.time(30 sec)\nselect deviceID, max(temp) as maxValue\ngroup by deviceID\ninsert into analyzeStream for expired-events;\nfrom analyzeStream[maxValue > 50]\nselect maxValue\ninsert into bulbOnStream;\nfrom fireAlarmEventStream[temp < 50]\nselect deviceID, temp\ninsert into bulbOffStream;"
+        }
+      }]
+    },
+    "roles":["ANY"]
   }
 }

pom.xml

@@ -1512,7 +1512,7 @@
 
         <!-- Carbon Device Management -->
 
-        <carbon.device.mgt.version>2.0.65</carbon.device.mgt.version>
+        <carbon.device.mgt.version>2.0.68</carbon.device.mgt.version>
         <carbon.device.mgt.version.range>[2.0.0, 3.0.0)</carbon.device.mgt.version.range>
 
         <!-- IOT Device Management -->
@@ -1520,7 +1520,7 @@
 
         <!-- Carbon Device Management Plugins-->
 
-        <carbon.device.mgt.plugin.version>3.0.35</carbon.device.mgt.plugin.version>
+        <carbon.device.mgt.plugin.version>3.0.36</carbon.device.mgt.plugin.version>
 
         <!-- API Management -->
         <carbon.api.mgt.version>6.1.80</carbon.api.mgt.version>