Merge pull request #39 from gioscarda/missing_tests

[Fixes #38] Added missing tests for actions and resource_cleaner

test/continoustest.bat

@@ -33,6 +33,8 @@ python test_bus_independent_messages.py
 python test_bot.py
 python test_path.py
 python test_ftp_uploads.py
+python test_action.py
+python test_resource_cleaner.py
 python resource_cleaner.py
 
 IF "%1"=="" GOTO exit

test/test_action.py

@@ -0,0 +1,53 @@
+# (c) 2019 Open Source Geospatial Foundation - all rights reserved
+# (c) 2014 - 2015 Centre for Maritime Research and Experimentation (CMRE)
+# (c) 2013 - 2014 German Aerospace Center (DLR)
+# This code is licensed under the GPL 2.0 license, available at the root
+# application directory.
+
+import unittest
+import os
+from wpsremote.action import CopyFile, CopyINIFileAddParam
+import wpsremote.configInstance as configInstance
+
+__author__ = "Alessio Fabiani"
+__copyright__ = "Copyright 2019 Open Source Geospatial Foundation - all rights reserved"
+__license__ = "GPL"
+
+
+class TestAction(unittest.TestCase):
+
+    def test_copy_file(self):
+        copy_path = "./src/wpsremote/xmpp_data/test/test_dir/copy_of_test_file"
+        param_dict = {
+            "source": "./src/wpsremote/xmpp_data/test/test_dir/test_file",
+            "target": copy_path
+        }
+        cf = CopyFile(param_dict)
+        self.assertFalse(os.path.isfile(copy_path))
+        cf.execute("blah")
+        self.assertTrue(os.path.isfile(copy_path))
+        os.remove(copy_path)
+
+    def test_copy_INI_file_add_param(self):
+        copy_path = "./src/wpsremote/xmpp_data/test/copy_of_test_service.config"
+        param_dict = {
+            "source": "./src/wpsremote/xmpp_data/test/test_service.config",
+            "target": copy_path,
+            "param_section": "Input3",
+            "param_name": "another_param",
+            "param_value_ref": "input3_another_param"
+        }
+        cifap = CopyINIFileAddParam(param_dict)
+        input_values = {
+            "input3_another_param": "Another value"
+        }
+        self.assertFalse(os.path.isfile(copy_path))
+        cifap.execute(input_values)
+        self.assertTrue(os.path.isfile(copy_path))
+        config = configInstance.create(copy_path)
+        self.assertEqual("Another value", config.get("Input3", "another_param"))
+        os.remove(copy_path)
+
+
+if __name__ == '__main__':
+    unittest.main()

test/test_path.py

@@ -17,6 +17,7 @@ class TestPath(unittest.TestCase):
     LIST_DIRS = [
         path.path('./src/wpsremote/xmpp_data/test/asset_schema.json'),
         path.path('./src/wpsremote/xmpp_data/test/test_dir'),
+        path.path('./src/wpsremote/xmpp_data/test/resource_dir'),
         path.path('./src/wpsremote/xmpp_data/test/wcsOAAonDemand_template.properties.txt'),
         path.path('./src/wpsremote/xmpp_data/test/service.config'),
         path.path('./src/wpsremote/xmpp_data/test/OAAonDemandWrapper_template.properties.txt'),
@@ -29,7 +30,10 @@ class TestPath(unittest.TestCase):
         path.path('./src/wpsremote/xmpp_data/test/CMREOAA_MainConfigFile_template.json'),
         path.path('./src/wpsremote/xmpp_data/test/logger_test.properties')
     ]
-    DIRS = [path.path('./src/wpsremote/xmpp_data/test/test_dir')]
+    DIRS = [
+        path.path('./src/wpsremote/xmpp_data/test/test_dir'),
+        path.path('./src/wpsremote/xmpp_data/test/resource_dir')
+    ]
 
     def test_path_methods(self):
         p = path.path("test/path")
@@ -65,10 +69,10 @@ def test_path_methods(self):
         with self.assertRaises(OSError):
             p.listdir(pattern=None)
         existing_path = path.path("./src/wpsremote/xmpp_data/test")
-        self.assertEqual(len(existing_path.listdir(pattern=None)), 12)
+        self.assertEqual(len(existing_path.listdir(pattern=None)), 13)
         for d in existing_path.listdir(pattern=None):
             self.assertIn(d, TestPath.LIST_DIRS)
-        self.assertEqual(len(existing_path.dirs(pattern=None)), 1)
+        self.assertEqual(len(existing_path.dirs(pattern=None)), 2)
         for d in existing_path.dirs(pattern=None):
             self.assertIn(d, TestPath.DIRS)
         self.assertEqual(len(existing_path.files()), 11)

test/test_resource_cleaner.py

@@ -0,0 +1,46 @@
+# (c) 2019 Open Source Geospatial Foundation - all rights reserved
+# (c) 2014 - 2015 Centre for Maritime Research and Experimentation (CMRE)
+# (c) 2013 - 2014 German Aerospace Center (DLR)
+# This code is licensed under the GPL 2.0 license, available at the root
+# application directory.
+
+import unittest
+import random
+import os
+from wpsremote.path import path
+from wpsremote.resource_cleaner import Resource
+
+__author__ = "Alessio Fabiani"
+__copyright__ = "Copyright 2019 Open Source Geospatial Foundation - all rights reserved"
+__license__ = "GPL"
+
+
+class TestResourceCleaner(unittest.TestCase):
+
+    def test_instance_methods(self):
+        rc = Resource()
+        self.assertIsNone(rc.start_time())
+        self.assertIsNone(rc.cmd_line())
+        self.assertIsNone(rc.unique_id())
+        self.assertIsNone(rc.set_unique_id(1))
+        self.assertIsNone(rc.sendbox_path())
+        self.assertIsNone(rc.processbot_pid())
+        self.assertIsNone(rc.set_processbot_pid(1))
+        self.assertIsNone(rc.spawned_process_pids())
+        self.assertIsNone(rc.spawned_process_cmd())
+        sendbox_root = path("./src/wpsremote/xmpp_data/test/test_dir")
+        unique_id = str(random.randint(1, 1000)).zfill(5)
+        sendbox_path = sendbox_root / str(unique_id)
+        rc.set_from_servicebot(unique_id, sendbox_path)
+        self.assertEqual(rc.sendbox_path(), sendbox_path)
+        self.assertEqual(rc._unique_id, unique_id)
+        rc.set_from_processbot(os.getpid(), [unique_id])
+        self.assertIn(unique_id, rc._spawned_process_pids)
+        try:
+            rc.read()
+        except Exception as e:
+            self.fail(e)
+
+
+if __name__ == '__main__':
+    unittest.main()

test_suite.sh

@@ -24,3 +24,9 @@ coverage run --source=src test/test_path.py
 
 echo "Running... test_ftp_uploads"
 coverage run --source=src test/test_ftp_uploads.py
+
+echo "Running... test_action"
+coverage run --source=src test/test_action.py
+
+echo "Running... test_resource_cleaner"
+coverage run --source=src test/test_resource_cleaner.py