Added validate_zip tests

validation.py

@@ -1,11 +1,11 @@
 import json
 from pathlib import Path
 from zipfile import ZipFile
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Union
 from tempfile import TemporaryDirectory
 
 
-def validate_zip(submission_track: str, submission_zip: str):
+def validate_zip(submission_track: str, submission_zip: Union[Path, str]):
     """
     Validates the submission format and contents
     Args:
@@ -35,20 +35,19 @@ def validate_notsofar_submission(submission_dir: Path):
     Raises:
         ValueError: if the submission zip is invalid
     """
-    submission_file_names = ['tc_orc_wer_hyp.json', 'tcp_wer_hyp.json']
+    submission_file_names = ['tcp_wer_hyp.json']
+    optional_file_names = ['tc_orc_wer_ref.json']
     fields = ['session_id', 'words', 'speaker', 'start_time', 'end_time']
 
-    for file_name in submission_file_names:
+    for file_name in submission_file_names + optional_file_names:
         file_path = submission_dir / file_name
         if not file_path.exists():
-            raise ValueError(f'Missing {file_name}')
-        with open(file_path, 'r') as json_file:
-            json_data: List[Dict[str, Any]] = json.load(json_file)
-            if not isinstance(json_data, list):
-                raise ValueError(f'Invalid `{file_name}` format, expecting a list of entries')
-            for data in json_data:
-                if not all(field in data for field in fields):
-                    raise ValueError(f'Invalid `{file_name}` format, fields: {fields} are required in each entry')
+            if file_name in submission_file_names:
+                raise ValueError(f'Missing {file_name}')
+            else:
+                continue
+
+        validate_json_file_structure(file_path, fields)
 
 
 def validate_dasr_submission(submission_dir: Path):
@@ -64,16 +63,82 @@ def validate_dasr_submission(submission_dir: Path):
     fields = ['session_id', 'words', 'speaker', 'start_time', 'end_time']
 
     if not (submission_dir / 'dev').exists():
-        raise ValueError('Missing dev directory, expecting a directory named `dev` with the submission files in it.')
+        raise ValueError('Missing `dev` directory, expecting a directory named `dev` with the submission files in it.')
 
     for file_name in submission_file_names:
         file_path = submission_dir / 'dev' / file_name
         if not file_path.exists():
             raise ValueError(f'Missing {file_name}')
-        with open(file_path, 'r') as json_file:
-            json_data: List[Dict[str, Any]] = json.load(json_file)
-            if not isinstance(json_data, list):
-                raise ValueError(f'Invalid `{file_name}` format, expecting a list of entries')
-            for data in json_data:
-                if not all(field in data for field in fields):
-                    raise ValueError(f'Invalid `{file_name}` format, fields: {fields} are required in each entry')
+
+        validate_json_file_structure(file_path, fields)
+
+
+def validate_json_file_structure(file_path: Path, fields: List[str]):
+    """
+    Validates the structure of a json file
+    Args:
+        file_path: path to the json file
+        fields: list of fields that are required in each entry
+    Raises:
+        ValueError: if the json file is invalid
+
+    """
+    with open(file_path, 'r') as json_file:
+        json_data: List[Dict[str, Any]] = json.load(json_file)
+        if not isinstance(json_data, list):
+            raise ValueError(f'Invalid `{file_path.name}` format, expecting a list of entries')
+        for data in json_data:
+            if not all(field in data for field in fields):
+                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')
+
+
+if __name__ == '__main__':
+    test_validate_zip()