Adding unit test for coverage

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/PravegaSensorCollectorAppTest.java

@@ -0,0 +1,17 @@
+package io.pravega.sensor.collector;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class PravegaSensorCollectorAppTest {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(PravegaSensorCollectorAppTest.class);
+
+    @Test
+    public void testPravegaSensorCollector(){
+        PravegaSensorCollectorApp app = new PravegaSensorCollectorApp();
+        Assertions.assertNotNull(app.toString());
+    }
+}

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/file/FileIngestServiceTest.java

@@ -9,21 +9,56 @@
  */
 package io.pravega.sensor.collector.file;
 
-import io.pravega.sensor.collector.DeviceDriver;
 import io.pravega.sensor.collector.DeviceDriverConfig;
 import io.pravega.sensor.collector.DeviceDriverManager;
-import io.pravega.sensor.collector.file.rawfile.RawFileIngestService;
+import io.pravega.sensor.collector.Parameters;
+import io.pravega.sensor.collector.util.TestUtils;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mock;
 
-import java.sql.SQLException;
+import java.time.Duration;
 import java.util.Map;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.*;
 
+/*
+* Test class for FileIngestService
+*/
 public class FileIngestServiceTest {
+
+    private static final String PREFIX = Parameters.getEnvPrefix();
+    private static final String SEPARATOR = "_";
+    protected DeviceDriverConfig config;
+    static String filename = "./src/test/resources/RawFileIngestService.properties";
+    DeviceDriverConfig deviceDriverConfig;
+    Map<String, String> properties;
+
+    private DeviceDriverManager driverManager;
+
+    @BeforeEach
+    void setUp(){
+        properties = Parameters.getProperties(filename);
+        driverManager = new DeviceDriverManager(properties);
+        deviceDriverConfig = new DeviceDriverConfig("RAW1", "RawFileIngestService",
+                TestUtils.configFromProperties(PREFIX, SEPARATOR, properties), driverManager);
+    }
+
+    @Test
+    public void testFileIngestService(){
+        FileIngestService fileIngestService = new MockFileIngestService(deviceDriverConfig);
+        try {
+            fileIngestService.startAsync();
+            fileIngestService.awaitRunning(Duration.ofSeconds(10));
+        } catch (Exception e){
+            throw new RuntimeException(e);
+        }
+        Assertions.assertTrue(fileIngestService.isRunning());
+        Assertions.assertEquals(fileIngestService.getProperty("PERSISTENT_QUEUE_FILE"),
+                properties.get(PREFIX + "RAW1" + SEPARATOR + "PERSISTENT_QUEUE_FILE"));
+        try {
+            fileIngestService.stopAsync();
+        } catch (Exception e){
+            throw new RuntimeException(e);
+        }
+        Assertions.assertFalse(fileIngestService.isRunning());
+    }
 }

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/file/MockFileIngestService.java

@@ -0,0 +1,21 @@
+package io.pravega.sensor.collector.file;
+
+import io.pravega.sensor.collector.DeviceDriverConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MockFileIngestService extends FileIngestService {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MockFileIngestService.class);
+    public MockFileIngestService(DeviceDriverConfig config) {
+        super(config);
+    }
+
+    /*
+     * Mocking the behaviour of create stream
+     */
+    @Override
+    protected void createStream(String scopeName, String streamName) {
+        LOGGER.info("Do nothing for create stream");
+    }
+}

pravega-sensor-collector/src/test/java/io/pravega/sensor/collector/util/TestUtils.java

@@ -0,0 +1,54 @@
+package io.pravega.sensor.collector.util;
+
+import io.pravega.sensor.collector.DeviceDriverConfig;
+import io.pravega.sensor.collector.Parameters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Class to contain convenient utilities for writing test cases.
+ */
+public final class TestUtils {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(TestUtils.class);
+    private static final String CLASS_KEY = "CLASS";
+    protected DeviceDriverConfig config;
+
+    /*
+    * Utility method that returns a list of DeviceDriverConfig instances from key/value properties.
+    * It removes the prefix read from properties file.
+    * */
+    public static Map<String, String> configFromProperties(String prefix, String sep, Map<String, String> properties) {
+        Map<String, String> instanceProperties = new HashMap<>();
+        // Find instance names.
+        final List<String> instanceNames = properties.keySet().stream().flatMap((key) -> {
+            if (key.startsWith(prefix) && key.endsWith(sep + CLASS_KEY)) {
+                return Stream.of(key.substring(prefix.length(), key.length() - CLASS_KEY.length() - sep.length()));
+            }
+            return Stream.empty();
+        }).collect(Collectors.toList());
+        LOGGER.info("configFromProperties: instanceNames={}", instanceNames);
+        // Copy properties with prefix to keys without a prefix.
+        String instanceName = instanceNames.get(0);
+
+        for(Map.Entry<String, String> e : properties.entrySet()){
+            final String key = e.getKey();
+            final String instancePrefix = prefix + instanceName + sep;
+            if (key.startsWith(instancePrefix)) {
+                LOGGER.info("key"+ key.substring(instancePrefix.length()) + " value " + e.getValue());
+                instanceProperties.put(key.substring(instancePrefix.length()), e.getValue());
+            } else if (key.startsWith(prefix)) {
+                instanceProperties.put(key.substring(prefix.length()), e.getValue());
+            }
+            //LOGGER.info("configFromProperties: instanceProperties={}", instanceProperties);
+        }
+        LOGGER.info("configFromProperties: instanceProperties={}", instanceProperties);
+        return instanceProperties;
+    }
+}

pravega-sensor-collector/src/test/resources/ParquetFileIngest.properties

@@ -12,7 +12,7 @@ PRAVEGA_SENSOR_COLLECTOR_PARQ2_CLASS=io.pravega.sensor.collector.file.parquet.Pa
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_FILE_SPEC="/opt/pravega-sensor-collector/Parquet_Files/A,/opt/pravega-sensor-collector/Parquet_Files/B"
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_FILE_EXTENSION=parquet
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_DELETE_COMPLETED_FILES=false
-PRAVEGA_SENSOR_COLLECTOR_PARQ2_DATABASE_FILE=/opt/pravega-sensor-collector/datafile.db
+PRAVEGA_SENSOR_COLLECTOR_PARQ2_DATABASE_FILE=./datafile.db
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_SAMPLES_PER_EVENT=200
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_PRAVEGA_CONTROLLER_URI=tls://pravega-controller.sdp.cluster1.sdp-demo.org:443
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_SCOPE=project1
@@ -23,6 +23,7 @@ PRAVEGA_SENSOR_COLLECTOR_PARQ2_TRANSACTION_TIMEOUT_MINUTES=2.0
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_MIN_TIME_IN_MILLIS_TO_UPDATE_FILE=5000
 PRAVEGA_SENSOR_COLLECTOR_PARQ2_DELETE_COMPLETED_FILES_INTERVAL_IN_SECONDS=43200
 HADOOP_HOME=${HOME}/dev
+PRAVEGA_SENSOR_COLLECTOR_PARQ2_PERSISTENT_QUEUE_FILE=/tmp/accel1.db
 
 # windows - location of bin/winutils.exe
 export HADOOP_HOME=/opt/dev

pravega-sensor-collector/src/test/resources/RawFileIngestService.properties

@@ -0,0 +1,26 @@
+
+#
+# Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
+#
+# 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
+#
+# This file can be used to manually test RawFileIngestService.
+PRAVEGA_SENSOR_COLLECTOR_RAW1_CLASS=io.pravega.sensor.collector.file.rawfile.RawFileIngestService
+PRAVEGA_SENSOR_COLLECTOR_RAW1_PERSISTENT_QUEUE_FILE=/tmp/accel1.db
+PRAVEGA_SENSOR_COLLECTOR_RAW1_FILE_SPEC=../parquet-file-sample-data/test
+PRAVEGA_SENSOR_COLLECTOR_RAW1_FILE_EXTENSION=parquet
+PRAVEGA_SENSOR_COLLECTOR_RAW1_DATABASE_FILE=./test.db
+#PRAVEGA_SENSOR_COLLECTOR_RAW1_PRAVEGA_CONTROLLER_URI=tls://pravega-controller.texas-twister.ns.sdp.hop.lab.emc.com:443
+PRAVEGA_SENSOR_COLLECTOR_RAW1_PRAVEGA_CONTROLLER_URI=tcp://127.0.0.1:9090
+PRAVEGA_SENSOR_COLLECTOR_RAW1_SCOPE=test-psc
+PRAVEGA_SENSOR_COLLECTOR_RAW1_STREAM=test-psc-stream
+PRAVEGA_SENSOR_COLLECTOR_RAW1_ROUTING_KEY=$(hostname)
+PRAVEGA_SENSOR_COLLECTOR_RAW1_DELETE_COMPLETED_FILES=true
+PRAVEGA_SENSOR_COLLECTOR_RAW1_TRANSACTION_TIMEOUT_MINUTES=1.0
+PRAVEGA_SENSOR_COLLECTOR_RAW1_CREATE_SCOPE=true
+PRAVEGA_SENSOR_COLLECTOR_RAW1_MIN_TIME_IN_MILLIS_TO_UPDATE_FILE=5000
+PRAVEGA_SENSOR_COLLECTOR_RAW1_DELETE_COMPLETED_FILES_INTERVAL_IN_SECONDS=15