Merge pull request #26 from cibinsb/testing_taskmaster

More unittesting

.gitignore

@@ -23,6 +23,7 @@ wheels/
 *.egg-info/
 .installed.cfg
 *.egg
+.idea
 MANIFEST
 
 # PyInstaller

tests/test_job.py

@@ -1,14 +1,171 @@
 import unittest
-
+import json
+import os
+import sys
+import datetime
+from unittest.mock import patch
+from dateutil.tz import tzutc
+from tesk_core import taskmaster
 from tesk_core.job import Job
+from argparse import Namespace
+from datetime import timezone
+
+START_TIME = datetime.datetime.now(timezone.utc)
+class MockObject(object):
+    def __init__(self, dictionary):
+        for k, v in dictionary.items():
+            if isinstance(v, dict):
+                self.__dict__[k] = MockObject(v)
+            else:
+                self.__dict__[k] = v
+
+def read_namespaced_job_error(name, namespace):
+    return_value = {'active': 1,
+                     'completion_time': None,
+                     'conditions': None,
+                     'failed': None,
+                     'start_time': START_TIME - datetime.timedelta(minutes=1),
+                     'succeeded': None}
+    return MockObject({"status":return_value})
+
+
+def read_namespaced_job_success(name, namespace):
+    return_value = {'active': None,
+                 'completion_time': datetime.datetime(2020, 7, 20, 5, 12, 42, tzinfo=tzutc()),
+                 'conditions': [MockObject({'last_probe_time': datetime.datetime(2020, 7, 20, 5, 12, 42, tzinfo=tzutc()),
+                                 'last_transition_time': datetime.datetime(2020, 7, 20, 5, 12, 42, tzinfo=tzutc()),
+                                 'message': None,
+                                 'reason': None,
+                                 'status': 'True',
+                                 'type': 'Complete'})],
+                 'failed': None,
+                 'start_time': datetime.datetime(2020, 7, 20, 5, 12, 35, tzinfo=tzutc()),
+                 'succeeded': 1}
+    return  MockObject({"status":return_value})
 
+def list_namespaced_pod_error(namespace, label_selector):
+    return_value =  {"status": {"conditions": [{"last_probe_time": None,
+                            "last_transition_time": START_TIME- datetime.timedelta(minutes=1),
+                            "message": None,
+                            "reason": None,
+                            "status": "True",
+                            "type": "Initialized"},
+                           {"last_probe_time": None,
+                            "last_transition_time": START_TIME- datetime.timedelta(minutes=1),
+                            "message": "containers with unready status: "
+                                       "[task-1000-ex-00]",
+                            "reason": "ContainersNotReady",
+                            "status": "False",
+                            "type": "Ready"},
+                           {"last_probe_time": None,
+                            "last_transition_time": START_TIME- datetime.timedelta(minutes=1),
+                            "message": "containers with unready status: "
+                                       "[task-1000-ex-00]",
+                            "reason": "ContainersNotReady",
+                            "status": "False",
+                            "type": "ContainersReady"},
+                           {"last_probe_time": None,
+                            "last_transition_time": START_TIME- datetime.timedelta(minutes=1),
+                            "message": None,
+                            "reason": None,
+                            "status": "True",
+                            "type": "PodScheduled"}],
+            "container_statuses": [MockObject({"container_id": None,
+                                    "image": "ubuntu_mock_test_image",
+                                    "image_id": "",
+                                    "last_state": {"running": None,
+                                                   "terminated": None,
+                                                   "waiting": None},
+                                    "name": "task-1000-ex-00",
+                                    "ready": False,
+                                    "restart_count": 0,
+                                    "state": {"running": None,
+                                              "terminated": None,
+                                              "waiting": {"message": "Back-off "
+                                                                     "pulling "
+                                                                     "image "
+                                                                     "ubuntu_mock_test_image",
+                                                          "reason": "ImagePullBackOff"}}})],
+            "host_ip": "192.168.99.100",
+            "init_container_statuses": None,
+            "message": None,
+            "nominated_node_name": None,
+            "phase": "Pending",
+            "pod_ip": "172.17.0.5",
+            "qos_class": "BestEffort",
+            "reason": None,
+            "start_time": START_TIME- datetime.timedelta(minutes=1)}}
+    return MockObject({"items":[MockObject(return_value)]})
 
 class JobTestCase(unittest.TestCase):
+    def setUp(self):
+        self.data = json.loads(open(os.path.join(os.path.dirname(__file__), "resources/inputFile.json")).read())
+        taskmaster.args = Namespace(debug=False, file=None, filer_version='v0.1.9', json='json'
+                                    , namespace='default', poll_interval=5, state_file='/tmp/.teskstate'
+                                    , localKubeConfig=False, pull_policy_always=False
+                                    , filer_name="eu.gcr.io/tes-wes/filer", pod_timeout=240
+                                    )
     def test_job(self):
         job = Job({'metadata': {'name': 'test'}})
         self.assertEqual(job.name, 'task-job')
         self.assertEqual(job.namespace, 'default')
 
+    @patch("kubernetes.client.BatchV1Api.create_namespaced_job")
+    @patch("tesk_core.job.Job.get_status", side_effect=[("Running", True),("Running", True),("Running", True),
+                                                        ("Running", True),("Running", True),("Complete", True)])
+    def test_run_to_completion_success(self, mock_get_status, mock_create_namespaced_job):
+        for executor in self.data['executors']:
+            jobname = executor['metadata']['name']
+            job = Job(executor, jobname, taskmaster.args.namespace)
+            status = job.run_to_completion(taskmaster.args.poll_interval, taskmaster.check_cancelled,
+                                           taskmaster.args.pod_timeout)
+            self.assertEqual(status, "Complete")
+
+    @patch("tesk_core.job.Job.delete")
+    @patch("tesk_core.taskmaster.check_cancelled", return_value=True)
+    @patch("kubernetes.client.BatchV1Api.create_namespaced_job")
+    @patch("tesk_core.job.Job.get_status", side_effect=[("Running", True)])
+    def test_run_to_completion_cancelled(self, mock_get_status, mock_create_namespaced_job, mock_check_cancelled,
+                               mock_job_delete):
+        for executor in self.data['executors']:
+            jobname = executor['metadata']['name']
+            job = Job(executor, jobname, taskmaster.args.namespace)
+            status = job.run_to_completion(taskmaster.args.poll_interval, taskmaster.check_cancelled,
+                                           taskmaster.args.pod_timeout)
+            self.assertEqual(status, "Cancelled")
+
+    @patch("kubernetes.client.CoreV1Api.list_namespaced_pod", side_effect=list_namespaced_pod_error)
+    @patch("kubernetes.client.BatchV1Api.read_namespaced_job", side_effect=read_namespaced_job_error)
+    @patch("tesk_core.job.Job.delete")
+    @patch("tesk_core.taskmaster.check_cancelled", return_value=False)
+    @patch("kubernetes.client.BatchV1Api.create_namespaced_job")
+    def test_run_to_completion_error(self, mock_create_namespaced_job, mock_check_cancelled,
+                                         mock_job_delete,mock_list_namespaced_pod,  mock_read_namespaced_job):
+        for executor in self.data['executors']:
+            jobname = executor['metadata']['name']
+            job = Job(executor, jobname, taskmaster.args.namespace)
+            status = job.run_to_completion(1, taskmaster.check_cancelled,
+                                           120)
+            self.assertEqual(status, "Error")
+
+    @patch("kubernetes.client.BatchV1Api.read_namespaced_job", side_effect=read_namespaced_job_error)
+    @patch("kubernetes.client.CoreV1Api.list_namespaced_pod", side_effect=list_namespaced_pod_error)
+    def test_get_status_error(self,mock_list_namespaced_pod,  mock_read_namespaced_job):
+        executor = self.data['executors'][0]
+        jobname = executor['metadata']['name']
+        job = Job(executor, jobname, taskmaster.args.namespace)
+        job.timeout = 5
+        status, all_pods_running = job.get_status(False)
+        self.assertEqual(status,"Error")
+
+    @patch("kubernetes.client.CoreV1Api.list_namespaced_pod")
+    @patch("kubernetes.client.BatchV1Api.read_namespaced_job", side_effect=read_namespaced_job_success)
+    def test_get_status_success(self, mock_read_namespaced_job, mock_list_namespaced_pod):
+        executor = self.data['executors'][0]
+        jobname = executor['metadata']['name']
+        job = Job(executor, jobname, taskmaster.args.namespace)
+        status, all_pods_running = job.get_status(False)
+        self.assertEqual(status, "Complete")
 
 if __name__ == '__main__':
     unittest.main()

tests/test_taskmaster.py

@@ -0,0 +1,118 @@
+import json
+import os
+import unittest
+from unittest.mock import patch
+from argparse import Namespace
+from tesk_core import taskmaster
+from tesk_core.filer_class import Filer
+from tesk_core.taskmaster import init_pvc, PVC, run_executor,\
+    generate_mounts, append_mount, dirname, run_task,newParser
+
+
+class TaskmasterTest(unittest.TestCase):
+
+    def setUp(self):
+        self.data = json.loads(open(os.path.join(os.path.dirname(__file__), "resources/inputFile.json")).read())
+        self.task_name = self.data['executors'][0]['metadata']['labels']['taskmaster-name']
+        taskmaster.args = Namespace( debug = False, file = None, filer_version = 'v0.1.9', json = 'json'
+                                    ,namespace='default', poll_interval=5, state_file='/tmp/.teskstate'
+                                    , localKubeConfig=False, pull_policy_always=False
+                                    , filer_name= "eu.gcr.io/tes-wes/filer", pod_timeout = 240
+                                    )
+        self.filer = Filer(self.task_name + '-filer', self.data, taskmaster.args.filer_name,
+                            taskmaster.args.filer_version, taskmaster.args.pull_policy_always)
+        self.pvc = PVC(self.task_name + '-pvc', self.data['resources']['disk_gb'], taskmaster.args.namespace)
+
+        taskmaster.created_jobs = []
+
+    @patch("tesk_core.taskmaster.PVC.create")
+    @patch("tesk_core.taskmaster.Job.run_to_completion", return_value="Complete")
+    @patch("tesk_core.taskmaster.logger")
+    def test_pvc_creation(self, mock_logger, mock_run_to_compl, mock_pvc_create):
+        """
+        Testing to check if the PVC volume was created successfully
+        """
+        self.assertIsInstance(init_pvc(self.data, self.filer), PVC)
+
+
+    @patch("tesk_core.taskmaster.PVC.delete")
+    @patch("tesk_core.taskmaster.PVC.create")
+    @patch("tesk_core.taskmaster.Job.run_to_completion", return_value="error")
+    @patch("tesk_core.taskmaster.logger")
+    def test_pvc_failure(self, mock_logger, run_to_compl, mock_pvc_create, mock_pvc_delete):
+        """
+        Testcase for finding if the PVC creation failed with exit 0
+        """
+
+        self.assertRaises(SystemExit, init_pvc, self.data, self.filer)
+
+    @patch("tesk_core.taskmaster.PVC.delete")
+    @patch("tesk_core.taskmaster.Job.delete")
+    @patch("tesk_core.taskmaster.Job.run_to_completion", return_value="Error")
+    @patch("tesk_core.taskmaster.logger")
+    def test_run_executor_failure(self, mock_logger, mock_run_to_compl, mock_job_delete,  mock_pvc_delete):
+        """
+
+        """
+        self.assertRaises(SystemExit, run_executor, self.data['executors'][0],taskmaster.args.namespace)
+
+    @patch("tesk_core.taskmaster.PVC")
+    @patch("tesk_core.taskmaster.Job.run_to_completion", return_value="Complete")
+    @patch("tesk_core.taskmaster.logger")
+    def test_run_executor_complete(self, mock_logger, mock_run_to_compl, mock_pvc):
+        """
+
+        """
+        self.assertEqual(run_executor(self.data['executors'][0], taskmaster.args.namespace,mock_pvc),None)
+
+
+
+    @patch("tesk_core.taskmaster.logger")
+    def test_generate_mount(self, mock_logger):
+        """
+
+        """
+        self.assertIsInstance(generate_mounts(self.data, self.pvc),list)
+
+    @patch("tesk_core.taskmaster.logger")
+    def test_append_mount(self, mock_logger):
+        """
+
+        """
+        volume_mounts = []
+        task_volume_name = 'task-volume'
+        for aninput in self.data['inputs']:
+            dirnm = dirname(aninput)
+            append_mount(volume_mounts, task_volume_name, dirnm, self.pvc)
+        self.assertEqual(volume_mounts,[{'name': task_volume_name, 'mountPath': '/some/volume', 'subPath': 'dir0'}])
+
+
+    @patch('tesk_core.taskmaster.logger')
+    @patch('tesk_core.taskmaster.PVC.create')
+    @patch('tesk_core.taskmaster.PVC.delete')
+    @patch('tesk_core.taskmaster.Job.run_to_completion', return_value='Complete' )
+    def test_run_task(self, mock_job, mock_pvc_create, mock_pvc_delete, mock_logger):
+        """
+
+        """
+        run_task(self.data, taskmaster.args.filer_name, taskmaster.args.filer_version)
+
+    def test_localKubeConfig(self):
+        """
+
+        """
+        parser = newParser()
+        args = parser.parse_args(['json', '--localKubeConfig'])
+        self.assertEqual(args
+                          , Namespace(debug=False, file=None, filer_version='v0.1.9', json='json', namespace='default',
+                                      poll_interval=5, state_file='/tmp/.teskstate'
+                                      , localKubeConfig=True
+                                      , pull_policy_always=False
+                                      , filer_name='eu.gcr.io/tes-wes/filer'
+                                      , pod_timeout=240
+                                      )
+                          )
+
+if __name__ == "__main__":
+    #import sys;sys.argv = ['', 'Test.testName']
+    unittest.main()