Merge pull request #5057 from broadinstitute/fix-wait-on-file-iter

Fix wait on file iter

Author: hanars

seqr/management/tests/check_for_new_samples_from_pipeline_tests.py

@@ -800,10 +800,12 @@ class AirtableCheckNewSamplesTest(AnvilAuthenticationTestCase, CheckNewSamplesTe
         ('Fetched 2 AnVIL Seqr Loading Requests Tracking records from airtable', None),
     ]
     VALIDATION_LOGS = [
+        '==> gsutil ls gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-14/validation_errors.json',
         '==> gsutil cat gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-14/validation_errors.json',
         'Fetching AnVIL Seqr Loading Requests Tracking records 0-2 from airtable',
         'Fetched 1 AnVIL Seqr Loading Requests Tracking records from airtable',
         '==> gsutil mv /mock/tmp/* gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-14/',
+        '==> gsutil ls gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-24/validation_errors.json',
         '==> gsutil cat gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-24/validation_errors.json',
         '==> gsutil mv /mock/tmp/* gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-24/',
     ]
@@ -892,22 +894,30 @@ def _set_reloading_loading_files(self):
     def _set_loading_files(self):
         responses.calls.reset()
         self.mock_subprocess.reset_mock()
-        self.mock_subprocess.side_effect = [self.mock_ls_process] + [
-            mock_opened_file(i) for i in range(len(OPENED_RUN_JSON_FILES) - 1)
-        ] + [self.mock_mv_process, mock_opened_file(-1), self.mock_mv_process]
+        subprocesses = [self.mock_ls_process]
+        for i in range(len(OPENED_RUN_JSON_FILES) - 1):
+            subprocesses += [self.mock_mv_process, mock_opened_file(i)]
+        subprocesses += [self.mock_mv_process, self.mock_mv_process,  mock_opened_file(-1), self.mock_mv_process]
+        self.mock_subprocess.side_effect = subprocesses
 
     def _assert_expected_loading_file_calls(self, single_call):
         calls = [
             ('gsutil ls gs://seqr-hail-search-data/v3.1/*/*/runs/*/*', -1),
+            ('gsutil ls gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/auto__2023-08-09/metadata.json', -2),
             ('gsutil cat gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/auto__2023-08-09/metadata.json', -2),
         ]
         if not single_call:
             calls += [
+                ('gsutil ls gs://seqr-hail-search-data/v3.1/GRCh37/SNV_INDEL/runs/manual__2023-11-02/metadata.json', -2),
                 ('gsutil cat gs://seqr-hail-search-data/v3.1/GRCh37/SNV_INDEL/runs/manual__2023-11-02/metadata.json', -2),
+                ('gsutil ls gs://seqr-hail-search-data/v3.1/GRCh38/MITO/runs/auto__2024-08-12/metadata.json', -2),
                 ('gsutil cat gs://seqr-hail-search-data/v3.1/GRCh38/MITO/runs/auto__2024-08-12/metadata.json', -2),
+                ('gsutil ls gs://seqr-hail-search-data/v3.1/GRCh38/GCNV/runs/auto__2024-09-14/metadata.json', -2),
                 ('gsutil cat gs://seqr-hail-search-data/v3.1/GRCh38/GCNV/runs/auto__2024-09-14/metadata.json', -2),
+                ('gsutil ls gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-14/validation_errors.json', -2),
                 ('gsutil cat gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-14/validation_errors.json', -2),
                 ('gsutil mv /mock/tmp/* gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-14/', -2),
+                ('gsutil ls gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-24/validation_errors.json', -2),
                 ('gsutil cat gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-24/validation_errors.json', -2),
                 ('gsutil mv /mock/tmp/* gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-01-24/', -2),
             ]
@@ -916,7 +926,8 @@ def _assert_expected_loading_file_calls(self, single_call):
         )
 
     def _additional_loading_logs(self, data_type, version):
-        return [(f'==> gsutil cat gs://seqr-hail-search-data/v3.1/{data_type.replace("SV", "GCNV")}/runs/{version}/metadata.json', None)]
+        return [(f'==> gsutil ls gs://seqr-hail-search-data/v3.1/{data_type.replace("SV", "GCNV")}/runs/{version}/metadata.json', None),
+                (f'==> gsutil cat gs://seqr-hail-search-data/v3.1/{data_type.replace("SV", "GCNV")}/runs/{version}/metadata.json', None)]
 
     def _assert_expected_airtable_calls(self, has_success_run, single_call):
         # Test request tracking updates for validation errors

seqr/management/tests/load_rna_seq_tests.py

@@ -20,6 +20,10 @@ def setUp(self):
         mock_gzip_open = patcher.start()
         self.mock_gzip_file_iter = mock_gzip_open.return_value.__enter__.return_value.__iter__
         self.addCleanup(patcher.stop)
+        patcher = mock.patch('seqr.utils.file_utils.os')
+        mock_os = patcher.start()
+        mock_os.path.isfile.return_value = True
+        self.addCleanup(patcher.stop)
         patcher = mock.patch('seqr.management.commands.load_rna_seq.open')
         self.mock_open = patcher.start()
         self.addCleanup(patcher.stop)

seqr/management/tests/update_individuals_sample_qc_tests.py

@@ -82,6 +82,8 @@ def setUp(self):
         self.mock_subprocess = patcher.start()
         self.addCleanup(patcher.stop)
         self.mock_ls_process = mock.MagicMock()
+        self.mock_does_exist_process = mock.MagicMock()
+        self.mock_does_exist_process.wait.return_value = 0
         self.mock_metadata_file = mock.MagicMock()
         super().setUp()
 
@@ -100,14 +102,14 @@ def test_command(self):
         self.mock_ls_process.communicate.return_value = b'gs://seqr-hail-search-data/v3.1/GRCh38/SNV_INDEL/runs/manual__2025-02-24/metadata.json\n', b''
         self.mock_metadata_file.stdout = [json.dumps({}).encode()]
         self.mock_metadata_file.wait.return_value = 0
-        self.mock_subprocess.side_effect = [self.mock_ls_process, self.mock_metadata_file]
+        self.mock_subprocess.side_effect = [self.mock_ls_process, self.mock_does_exist_process, self.mock_metadata_file]
 
         with self.assertRaises(CommandError):
             call_command('update_individuals_sample_qc', 'SNV_INDEL', 'GRCh38', 'manual__2025-02-24')
 
         # Test valid case
         self.mock_metadata_file.stdout = [json.dumps(METADATA_JSON).encode()]
-        self.mock_subprocess.side_effect = [self.mock_ls_process, self.mock_metadata_file]
+        self.mock_subprocess.side_effect = [self.mock_ls_process, self.mock_does_exist_process, self.mock_metadata_file]
         call_command('update_individuals_sample_qc', 'SNV_INDEL', 'GRCh38', 'manual__2025-02-24')
 
         # Individual model properly updated with sample qc results

seqr/utils/file_utils.py

@@ -43,7 +43,7 @@ def does_file_exist(file_path, user=None):
         success = process.wait() == 0
         if not success:
             errors = [line.decode('utf-8').strip() for line in process.stdout]
-            logger.info(' '.join(errors), user)
+            logger.warning(' '.join(errors), user)
         return success
     return os.path.isfile(file_path)
 
@@ -57,6 +57,8 @@ def list_files(wildcard_path, user, check_subfolders=False, allow_missing=True):
 
 
 def file_iter(file_path, byte_range=None, raw_content=False, user=None, **kwargs):
+    if not does_file_exist(file_path, user=user):
+        raise FileNotFoundError(f'Could not access file {file_path}')
     if is_google_bucket_file_path(file_path):
         for line in _google_bucket_file_iter(file_path, byte_range=byte_range, raw_content=raw_content, user=user, **kwargs):
             yield line
@@ -82,7 +84,7 @@ def file_iter(file_path, byte_range=None, raw_content=False, user=None, **kwargs
 def _google_bucket_file_iter(gs_path, byte_range=None, raw_content=False, user=None, **kwargs):
     """Iterate over lines in the given file"""
     range_arg = ' -r {}-{}'.format(byte_range[0], byte_range[1]) if byte_range else ''
-    process = run_gsutil_with_wait(
+    process = _run_gsutil_command(
         'cat{}'.format(range_arg), gs_path, gunzip=gs_path.endswith("gz") and not raw_content, user=user, **kwargs)
     for line in process.stdout:
         if not raw_content:

seqr/utils/vcf_utils.py

@@ -76,15 +76,17 @@ def _get_vcf_meta_info(line):
 def validate_vcf_and_get_samples(data_path, user, genome_version, path_name=None, dataset_type=None):
     allowed_exts = DATA_TYPE_FILE_EXTS.get(dataset_type)
 
-    vcf_filename = _validate_vcf_exists(data_path, user, path_name, allowed_exts)
+    vcf_filename = _validate_valid_vcf_name(data_path, user, allowed_exts)
 
-    if allowed_exts and vcf_filename.endswith(allowed_exts):
+    if vcf_filename is None:
         return None
 
     byte_range = None if vcf_filename.endswith('.vcf') else (0, BLOCK_SIZE)
     meta = defaultdict(dict)
     try:
-        header_line = next(_get_vcf_header_line(file_iter(vcf_filename, byte_range=byte_range), meta))
+        header_line = next(_get_vcf_header_line(file_iter(vcf_filename, byte_range=byte_range, user=user), meta))
+    except FileNotFoundError:
+        raise ErrorsWarningsException([f'Data file or path {path_name or data_path} is not found.'], [])
     except StopIteration:
         raise ErrorsWarningsException(['No header found in the VCF file.'], [])
     except UnicodeDecodeError:
@@ -117,23 +119,22 @@ def _get_vcf_header_line(vcf_file, meta):
                     meta[meta_info['field']].update({meta_info['id']: meta_info['type']})
 
 
-def _validate_vcf_exists(data_path, user, path_name, allowed_exts):
+def _validate_valid_vcf_name(data_path, user, allowed_exts):
     file_extensions = (allowed_exts or ()) + VCF_FILE_EXTENSIONS
     if not data_path.endswith(file_extensions):
         raise ErrorsWarningsException([
             'Invalid VCF file format - file path must end with {}'.format(' or '.join(file_extensions))
         ])
 
-    file_to_check = None
+    file_to_check = data_path
     if '*' in data_path:
         files = list_files(data_path, user)
         if files:
             file_to_check = files[0]
-    elif does_file_exist(data_path, user=user):
-        file_to_check = data_path
-
-    if not file_to_check:
-        raise ErrorsWarningsException(['Data file or path {} is not found.'.format(path_name or data_path)])
+    elif allowed_exts and data_path.endswith(allowed_exts):
+        if not does_file_exist(data_path, user=user):
+            raise ErrorsWarningsException([f'Data file or path {data_path} is not found.'])
+        file_to_check = None
 
     return file_to_check