Feature/unittest 2

.github/workflows/python-app.yml

@@ -34,7 +34,9 @@ jobs:
         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
-    - name: Test with pytest
+    - name: Run Unit Test via Pytest
       run: |
-        python -m unittest discover -s ./unittests
-
+        coverage run -m unittest discover -v -s .
+    - name: Generate Coverage Report
+      run: |
+        coverage report -m -i

producer.py

@@ -65,7 +65,7 @@ def __len__(self):
         return 0
 
 
-if __name__ == '__main__':
+def main():
     parser = argparse.ArgumentParser()
     parser.add_argument('infile', help='Input file (leave empty to use webcam)', nargs='?', type=str, default=None)
     parser.add_argument('-o', '--output', help='Output stream key name', type=str, default='camera:0')
@@ -118,4 +118,5 @@ def __len__(self):
                 logging.info('Stopping after {} frames.'.format(count))
                 break
 
-
+if __name__ == '__main__':
+    main()

requirements.txt

@@ -8,5 +8,5 @@ flask
 Pillow==8.3.2
 seaborn
 redistimeseries
-
-
+coverage
+mmcv~=1.7.1

server.py

@@ -43,7 +43,7 @@ def get_last(self):
         if tracking_stream and len(tracking_stream[0]) > 0:
             last_frame_refId = tracking_stream[0][1][b'refId'].decode("utf-8")  # Frame reference i
             tracking = json.loads(tracking_stream[0][1][b'tracking'].decode('utf-8'))
-            resp = conn.xread({self.camera: last_frame_refId}, count=1)
+            resp = self.conn.xread({self.camera: last_frame_refId}, count=1)
             key, messages = resp[0]
             frame_last_id, data = messages[0]
 

tracking/tracker.py

@@ -7,7 +7,11 @@
 import mmcv
 from redis import Redis
 import redis
-from Monitor import GPUCalculator , MMTMonitor
+import sys
+if 'tracking.unittests' in sys.modules:
+    from tracking.Monitor import GPUCalculator, MMTMonitor
+else:
+    from Monitor import GPUCalculator, MMTMonitor
 
 redis_client = redis.StrictRedis('redistimeseries', 6379)
 

tracking/unittests/test_monitor.py

@@ -0,0 +1,57 @@
+import time
+import unittest
+
+from unittest import mock
+import sys
+from unittest.mock import patch
+
+from tracking.Monitor import GPUCalculator
+
+redis_mock = mock.MagicMock()
+sys.modules['redis'] = redis_mock
+
+
+class TestMonitor(unittest.TestCase):
+    """
+    This class tests the Monitor functionalities.
+    """
+    def test_MMTMonitor_starttimer(self):
+        redis_mock = mock.MagicMock()
+        import tracking.Monitor as monitor
+        obj = monitor.MMTMonitor(redis_mock, 'key')
+        obj.start_timer()
+        self.assertIsNotNone(obj.start_time)
+
+    def test_MMTMonitor_with_latency(self):
+        redis_mock = mock.MagicMock()
+        import tracking.Monitor as monitor
+        obj = monitor.MMTMonitor(redis_mock, 'key')
+        obj.start_timer()
+        time.sleep(1)
+        obj.end_timer()
+        print(obj.average_latency)
+        self.assertNotEqual(obj.average_latency, 0.0)
+
+    def test_MMTMonitor_counter_15(self):
+        redis_mock = mock.MagicMock()
+        import tracking.Monitor as monitor
+        obj = monitor.MMTMonitor(redis_mock, 'key')
+        obj.start_timer()
+        obj.counter = 14
+        obj.end_timer()
+        print(obj.average_latency)
+        self.assertEqual(obj.average_latency, 0)
+
+    def test_GPUCalculator(self):
+        gpu_calculator = GPUCalculator(redis_mock)
+        gpu_calculator.add()
+        self.assertEqual(gpu_calculator.count, 1)
+
+    @patch('tracking.Monitor.subprocess')
+    def test_GPUCalculator_with_count_15(self, mock_sub):
+        output = '1, 2, 3, 4, 5'.encode('utf-8')
+        mock_sub.check_output.return_value = output
+        gpu_calculator = GPUCalculator(redis_mock)
+        gpu_calculator.count = 14
+        gpu_calculator.add()
+        self.assertEqual(gpu_calculator.count, 15)

tracking/unittests/test_tracker.py

@@ -0,0 +1,75 @@
+import argparse
+
+import pickle
+import unittest
+
+from unittest import mock
+import sys
+from unittest.mock import patch
+
+import numpy as np
+
+arg_mock = mock.MagicMock()
+redis_mock = mock.MagicMock()
+sys.modules['redis'] = redis_mock
+sys.modules['argparse'] = arg_mock
+sys.modules['redis'] = redis_mock
+
+sys.modules['mmtracking'] = arg_mock
+sys.modules['mmtracking.mmtrack'] = arg_mock
+sys.modules['mmtracking.mmtrack.apis'] = arg_mock
+
+import tracking.tracker as tracker
+
+
+class TestTracker(unittest.TestCase):
+    """
+    This class tests the tracker functionalities.
+    """
+    @patch('tracking.tracker.results2outs')
+    def test_main_with_no_infile(self, mock_results):
+        ids_mock = mock.MagicMock()
+        mock_results.return_value = ids_mock
+        ids_mock.get.return_value = (1, 2, 3)
+        redis_mock2 = mock.MagicMock()
+        redis_mock.Redis.return_value = redis_mock2
+        redis_mock2.ping.return_value = True
+        redis_mock2.xadd.side_effect = [1, Exception]
+        img = np.array([[[62, 62, 62], [61, 61, 61], [63, 63, 63]]])
+        msg = {
+            b'frameId': '1'.encode('utf-8'),
+            b'image': pickle.dumps(img)
+        }
+        redis_mock2.xread.return_value = [(1, [(1, msg)])]
+        arg_mock2 = mock.MagicMock()
+        arg_mock.ArgumentParser.return_value = arg_mock2
+        arg_mock2.parse_args.return_value = argparse.Namespace(input_stream='camera:0', classId='PERSON',
+                                                               output_stream='camera:0:mot',
+                                                               checkpoint='',
+                                                               device='cuda:0',
+                                                               redis='redis://127.0.0.1:6379',
+                                                               maxlen=3000,
+                                                               config='')
+        with self.assertRaises(Exception):
+            tracker.main()
+
+    def test_results2outs_error(self):
+        bbox = np.zeros((0, 4), dtype=np.float32), np.zeros((0, 4), dtype=np.float32)
+        with self.assertRaises(NotImplementedError):
+            tracker.results2outs(bbox, bbox, bbox)
+
+    def test_results2outs_5(self):
+        bbox = np.ones((1, 5), dtype=np.float32), np.ones((1, 5), dtype=np.float32)
+        actual_response = tracker.results2outs(bbox, bbox, bbox)
+        self.assertEqual(actual_response['labels'].size, np.array([0, 1], dtype=np.int64).size)
+
+    def test_results2outs_6(self):
+        bbox = np.ones((1, 6), dtype=np.float32), np.ones((1, 6), dtype=np.float32)
+        actual_response = tracker.results2outs(bbox, bbox, bbox)
+        self.assertEqual(actual_response['labels'].size, np.array([0, 1], dtype=np.int64).size)
+
+    def test_results2outs_masks(self):
+        bbox = np.ones((1, 6), dtype=np.float32), np.ones((1, 6), dtype=np.float32)
+        actual_response = tracker.results2outs(bbox, bbox, (np.zeros((0, 6), dtype=np.float32),np.zeros((0, 6), dtype=np.float32)))
+        self.assertEqual(actual_response['labels'].size, np.array([0, 1], dtype=np.int64).size)
+

tracklet/unittests/test_tracklet.py

@@ -0,0 +1,48 @@
+import json
+
+import unittest
+from tracklet.tracklet import Tracklet
+
+
+class TestTracklet(unittest.TestCase):
+    """ This class tests the tracklet functionalities. """
+
+    def test_add_box(self):
+        obj = Tracklet('', '')
+        obj.add_box(-1)
+        obj.add_box(-1)
+        obj.add_box(-1)
+        obj.add_box(-1)
+        obj.add_box(-1)
+        self.assertEqual(obj._boxes, [-1, -1, -1, -1, -1])
+
+    def test_increase_skip(self):
+        from tracklet.tracklet import Tracklet
+        obj = Tracklet('', '')
+        obj.increase_skip()
+        self.assertEqual(obj._skipped_frames, 1)
+
+    def test_skipped_frames(self):
+        from tracklet.tracklet import Tracklet
+        obj = Tracklet('', '')
+        self.assertEqual(obj.skipped_frames, 0)
+
+    def test_objectId(self):
+        from tracklet.tracklet import Tracklet
+        obj = Tracklet('', '')
+        self.assertEqual(obj.objectId, '')
+
+    def test_object_class(self):
+        from tracklet.tracklet import Tracklet
+        obj = Tracklet('', '')
+        self.assertEqual(obj.object_class, '')
+
+    def test__repr__(self):
+        from tracklet.tracklet import Tracklet
+        obj = Tracklet('', '')
+        res = obj.__repr__()
+        self.assertEqual(json.loads(res), {
+            "object_id": "",
+            "boxes": [],
+            "skipped_frames": "0"
+        })

tracklet/unittests/test_trackletmanager.py

@@ -0,0 +1,18 @@
+import unittest
+
+from tracklet.trackletmanager import TrackletManager
+
+
+class TestTrackletManager(unittest.TestCase):
+    """
+    This class tests the trackletmanager functionalities.
+    """
+    def test_process_objects_active(self):
+        obj = TrackletManager(3)
+        res = obj.process_objects({'a': 1, 'b': 2})
+        self.assertEqual(res, {'a': 'active', 'b': 'active'})
+
+    def test_process_objects_inactive(self):
+        obj = TrackletManager(1)
+        res = obj.process_objects({'a': None})
+        self.assertEqual(res, {'a': 'inactive'})

unittests/test_producer.py

@@ -0,0 +1,75 @@
+import sys
+import unittest
+from unittest import mock
+
+arg_mock = mock.MagicMock()
+redis_mock = mock.MagicMock()
+sys.modules['redis'] = redis_mock
+sys.modules['argparse'] = arg_mock
+
+import producer
+
+
+class TestProducer(unittest.TestCase):
+    """
+    This class tests the producer functionalities.
+    """
+    def test_main_with_infile(self):
+        redis_mock2 = mock.MagicMock()
+        redis_mock.Redis.return_value = redis_mock2
+        redis_mock2.ping.return_value = True
+        arg_mock2 = mock.MagicMock()
+        arg_mock.ArgumentParser.return_value = arg_mock2
+        arg_mock3 = mock.MagicMock()
+        arg_mock2.parse_args.return_value = arg_mock3
+        arg_mock3.url = 'redis://127.0.0.1:6379'
+        arg_mock3.infile = 'data/race.mp4'
+        arg_mock3.output = 'camera:0'
+        arg_mock3.webcam = 0
+        arg_mock3.verbose = False
+        arg_mock3.count = 5
+        arg_mock3.fmt = '.jpg'
+        arg_mock3.inputFps = 30.0
+        arg_mock3.maxlen = 3000
+        arg_mock3.outputFps = 10.0
+        res = producer.main()
+        self.assertIsNone(res)
+
+    def test_main_exception(self):
+        redis_mock2 = mock.MagicMock()
+        redis_mock.Redis.return_value = redis_mock2
+        redis_mock2.ping.return_value = False
+        arg_mock2 = mock.MagicMock()
+        arg_mock.ArgumentParser.return_value = arg_mock2
+        arg_mock3 = mock.MagicMock()
+        arg_mock2.parse_args.return_value = arg_mock3
+        arg_mock3.url = 'redis://127.0.0.1:6379'
+        arg_mock3.infile = 'data/race.mp4'
+        arg_mock3.output = 'camera:0'
+        arg_mock3.webcam = 0
+        arg_mock3.verbose = False
+        arg_mock3.count = 5
+        arg_mock3.fmt = '.jpg'
+        arg_mock3.inputFps = 30.0
+        arg_mock3.maxlen = 3000
+        arg_mock3.outputFps = 10.0
+        with self.assertRaises(Exception):
+            producer.main()
+
+    def test_video_sample_rate(self):
+        obj = producer.Video(0, 30)
+        res = obj.video_sample_rate(30)
+        self.assertEqual(res, 1)
+
+    def test_cam_release(self):
+        obj = producer.Video(0, 30)
+        res = obj.cam_release()
+        self.assertIsNone(res)
+
+    def test___iter__(self):
+        obj = producer.Video(0, 30)
+        res = obj.__iter__()
+        self.assertEqual(res.count, -1)
+        res1 = obj.__len__()
+        self.assertEqual(res1, 0)
+        obj.isFile = True