Added validate_zip tests

validation.py

@@ -1,4 +1,6 @@
+import os
 import json
+import unittest
 from pathlib import Path
 from zipfile import ZipFile
 from typing import List, Dict, Any, Union
@@ -92,53 +94,114 @@ def validate_json_file_structure(file_path: Path, fields: List[str]):
                 raise ValueError(f'Invalid `{file_path.name}` format, fields: {fields} are required in each entry')
 
 
-def test_validate_zip(data_samples: int = 10):
-    import os
-    with TemporaryDirectory() as temp_dir:
-        submission_zip = Path(temp_dir) / 'submission.zip'
-        valid_data = [{'session_id': 'session_id', 'words': 'words', 'speaker': 'speaker',
-                       'start_time': 0.0, 'end_time': 1.0} for _ in range(data_samples)]
-        invalid_data = [{'session_id': 'session_id', 'words': 'words', 'start_time': 0.0} for _ in range(data_samples)]
-
-        def create_test_data(submission_track: str, data: List[Dict[str, Any]], json_file_names: List[str],
-                             parent_zip_dir: str = None):
-            submission_dir = Path(temp_dir) / submission_track
-            os.makedirs(submission_dir, exist_ok=True)
-            with ZipFile(submission_zip, 'w') as submission_zip_file:
-                for json_file_name in json_file_names:
-                    if parent_zip_dir:
-                        json_file_name = str(Path(parent_zip_dir) / json_file_name)
-                    submission_zip_file.writestr(json_file_name, json.dumps(data))
-            return submission_track, submission_zip
-
-        def test(track: str, data: List[Dict[str, Any]], json_file_names: List[str], expected_error: bool,
-                 parent_zip_dir=None):
-            try:
-                validate_zip(*create_test_data(track, data, json_file_names, parent_zip_dir))
-                assert not expected_error, f'Expected error for {track}'
-            except ValueError as e:
-                assert expected_error, f'Unexpected error for {track}'
-
-        # NOTSOFAR-SC
-        test('NOTSOFAR-SC', valid_data, ['tcp_wer_hyp.json'], False)
-        test('NOTSOFAR-SC', valid_data, ['tcp_wer_hyp.json', 'tc_orc_wer_ref.json'], False)
-        test('NOTSOFAR-SC', invalid_data, ['tcp_wer_hyp.json'], True)
-        test('NOTSOFAR-SC', invalid_data, ['tcp_wer_hyp.json', 'tc_orc_wer_ref.json'], True)
-
-        # NOTSOFAR-MC
-        test('NOTSOFAR-MC', valid_data, ['tcp_wer_hyp.json'], False)
-        test('NOTSOFAR-MC', valid_data, ['tcp_wer_hyp.json', 'tc_orc_wer_ref.json'], False)
-        test('NOTSOFAR-MC', invalid_data, ['tcp_wer_hyp.json'], True)
-        test('NOTSOFAR-MC', invalid_data, ['tcp_wer_hyp.json', 'tc_orc_wer_ref.json'], True)
-
-        # DASR-Constrained-LM
-        test('DASR-Constrained-LM', valid_data, ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json'], False, 'dev')
-        test('DASR-Constrained-LM', invalid_data, ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json'], True, 'dev')
-
-        # DASR-Unconstrained-LM
-        test('DASR-Unconstrained-LM', valid_data, ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json'], False, 'dev')
-        test('DASR-Unconstrained-LM', invalid_data, ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json'], True, 'dev')
+####################################################################################################
+# Tests
+####################################################################################################
+
+class TestValidateZip(unittest.TestCase):
+    DATA_SAMPLES = 10
+
+    @classmethod
+    def setUpClass(cls):
+        cls.valid_data = [{'session_id': 'session_id', 'words': 'words', 'speaker': 'speaker',
+                           'start_time': 0.0, 'end_time': 1.0} for _ in range(cls.DATA_SAMPLES)]
+        cls.invalid_data = [{'session_id': 'session_id', 'words': 'words',
+                             'start_time': 0.0} for _ in range(cls.DATA_SAMPLES)]
+
+    def setUp(self):
+        self.temp_dir = TemporaryDirectory()
+        self.submission_zip = Path(self.temp_dir.name) / 'submission.zip'
+
+    def create_test_data(self, submission_track: str, data: List[Dict[str, Any]], json_file_names: List[str],
+                         parent_zip_dir: str = None):
+        submission_dir = Path(self.temp_dir.name) / submission_track
+        os.makedirs(submission_dir, exist_ok=True)
+        with ZipFile(self.submission_zip, 'w') as submission_zip_file:
+            for json_file_name in json_file_names:
+                if parent_zip_dir:
+                    json_file_name = str(Path(parent_zip_dir) / json_file_name)
+                submission_zip_file.writestr(json_file_name, json.dumps(data))
+        return submission_track, self.submission_zip
+
+    def tearDown(self):
+        self.temp_dir.cleanup()
+
+    def test_NOTSOFAR_SC_valid_data_tcp(self):
+        self.assertEqual(validate_zip(*self.create_test_data(
+            'NOTSOFAR-SC', self.valid_data, ['tcp_wer_hyp.json'])), None)
+
+    def test_NOTSOFAR_SC_valid_data_tcp_and_tcorc(self):
+        self.assertEqual(validate_zip(*self.create_test_data(
+            'NOTSOFAR-SC', self.valid_data, ['tcp_wer_hyp.json', 'tc_orc_wer_ref.json'])), None)
+
+    def test_NOTSOFAR_SC_missing_tcp_file(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data(
+                'NOTSOFAR-SC', self.valid_data, ['tc_orc_wer_ref.json']))
+
+    def test_NOTSOFAR_SC_invalid_data(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data(
+                'NOTSOFAR-SC', self.invalid_data, ['tcp_wer_hyp.json']))
+
+    def test_NOTSOFAR_MC_valid_data_tcp(self):
+        self.assertEqual(validate_zip(*self.create_test_data(
+            'NOTSOFAR-MC', self.valid_data, ['tcp_wer_hyp.json'])), None)
+
+    def test_NOTSOFAR_MC_valid_data_tcp_and_tcorc(self):
+        self.assertEqual(validate_zip(*self.create_test_data(
+            'NOTSOFAR-MC', self.valid_data, ['tcp_wer_hyp.json', 'tc_orc_wer_ref.json'])), None)
+
+    def test_NOTSOFAR_MC_missing_tcp_file(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data(
+                'NOTSOFAR-MC', self.valid_data, ['tc_orc_wer_ref.json']))
+
+    def test_NOTSOFAR_MC_invalid_data(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data(
+                'NOTSOFAR-MC', self.invalid_data, ['tcp_wer_hyp.json']))
+
+    def test_DASR_Constrained_LM_valid_data(self):
+        self.assertEqual(validate_zip(*self.create_test_data('DASR-Constrained-LM', self.valid_data,
+                                                             ['chime6.json', 'dipco.json', 'mixer6.json',
+                                                              'notsofar1.json'], 'dev')), None)
+
+    def test_DASR_Constrained_LM_invalid_data(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data('DASR-Constrained-LM', self.invalid_data,
+                                                ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json'], 'dev'))
+
+    def test_DASR_Constrained_LM_missing_dev_dir(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data('DASR-Constrained-LM', self.valid_data,
+                                                ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json']))
+
+    def test_DASR_Constrained_LM_missing_json_file(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data('DASR-Constrained-LM', self.valid_data,
+                                                ['chime6.json', 'dipco.json', 'mixer6.json'], 'dev'))
+
+    def test_DASR_Unconstrained_LM_valid_data(self):
+        self.assertEqual(validate_zip(*self.create_test_data('DASR-Unconstrained-LM', self.valid_data,
+                                                             ['chime6.json', 'dipco.json', 'mixer6.json',
+                                                              'notsofar1.json'], 'dev')), None)
+
+    def test_DASR_Unconstrained_LM_invalid_data(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data('DASR-Unconstrained-LM', self.invalid_data,
+                                                ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json'], 'dev'))
+
+    def test_DASR_Unconstrained_LM_missing_dev_dir(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data('DASR-Unconstrained-LM', self.valid_data,
+                                                ['chime6.json', 'dipco.json', 'mixer6.json', 'notsofar1.json']))
+
+    def test_DASR_Unconstrained_LM_missing_json_file(self):
+        with self.assertRaises(ValueError):
+            validate_zip(*self.create_test_data('DASR-Unconstrained-LM', self.valid_data,
+                                                ['chime6.json', 'dipco.json', 'mixer6.json'], 'dev'))
 
 
 if __name__ == '__main__':
-    test_validate_zip()
+    unittest.main()