Complete testing s3 coverage provider methods.

Author: grosscol

bravo_api/core/s3_coverage_provider.py

@@ -8,15 +8,19 @@
 import re
 import boto3
 from botocore.exceptions import ClientError
+from functools import reduce
 
 
 class S3CoverageProvider(CoverageProvider):
 
     def __init__(self, src):
         self.client = boto3.client('s3')
         self.src = src
+        split_url = urlparse(self.src)
+        self.bucket = split_url.netloc
+        self.prefix = split_url.path.lstrip('/')
         self.validate_source()
-        # self.catalog = self.discover_files()
+        self.catalog = self.discover_files()
 
     def validate_source(self):
         """
@@ -30,8 +34,8 @@ def validate_source(self):
 
         # Check that objects exists in bucket at prefix.
         try:
-            obj_resp = self.client.list_objects_v2(Bucket=split_url.netloc,
-                                                   Prefix=split_url.path.lstrip('/'),
+            obj_resp = self.client.list_objects_v2(Bucket=self.bucket,
+                                                   Prefix=self.prefix,
                                                    MaxKeys=10)
         except ClientError as err:
             msg = err.response['Error']['Message']
@@ -52,16 +56,38 @@ def validate_source(self):
     def evaluate_chrom_representation(self):
         """
         All chromosomes expected to be represented in all coverage bins
+        Return list of messages to the contrary.
+        """
+        messages = []
+        for bin_name, cov_bin in self.catalog.items():
+            missing_chroms = [chrom for chrom in self._chroms if chrom not in cov_bin.keys()]
+            if(missing_chroms):
+                messages.append(f'Coverage {bin_name} missing chroms: {missing_chroms}')
+        return(messages)
+
+    @staticmethod
+    def simple_dict_flat(acc, d):
         """
-        msgs = []
-        return(msgs)
+        Helper function for reducing list of non-nested dicts to list of values
+        """
+        acc.extend(list(d.values()))
+        return(acc)
 
     def evaluate_chrom_readability(self):
         """
-        All coverage files expected to be readable
+        All coverage files expected to be readable.
+        Return list of messages to the contrary.
         """
-        msgs = []
-        return(msgs)
+        messages = []
+        all_paths = reduce(S3CoverageProvider.simple_dict_flat, self.catalog.values(), [])
+
+        for path in all_paths:
+            try:
+                self.client.head_object(Bucket=self.bucket, Key=path)
+            except ClientError as err:
+                err_msg = err.response['Error']['Message']
+                messages.append(f'{self.bucket}/{path}: {err_msg}')
+        return(messages)
 
     def evaluate_catalog(self):
         warn_msgs = []
@@ -72,9 +98,31 @@ def evaluate_catalog(self):
     def discover_files(self):
         """
         Find and organize the chrN.bin_X.YZ.tar.gz coverage files into a dictionary organized by
-        bin then chromosome.
+        bin then chromosome. Form of bin-chr-file index will be:
+        {
+          'bin_0.25': {'chr1': 's3://bucket/prefix/chr1.bin_0.25.tsv.gz',
+                      ...,
+                      'chr22': 's3://bucket/prefix/chr22.bin_0.25.tsv.gz'},
+          ....
+          'bin_full': {...}
+        }
         """
+        split_url = urlparse(self.src)
+        bucket = split_url.netloc
+        prefix = split_url.path.lstrip('/')
+
+        obj_resp = self.client.list_objects_v2(Bucket=bucket, Prefix=prefix)
+        all_keys = [item['Key'] for item in obj_resp['Contents']]
+
+        chr_patt = re.compile(r'.*chr([0-9X]{1,2})')
         result = {}
+        for bin_name in self._bins:
+            result[bin_name] = {}
+            bin_keys = [item for item in all_keys if item.startswith(f'{prefix}/{bin_name}')]
+
+            for key in bin_keys:
+                chr = chr_patt.match(key).group(1)
+                result[bin_name][chr] = key
         return(result)
 
     def lookup_coverage_path(self, cov_bin, chrom):

tests/core/conftest.py

@@ -37,6 +37,10 @@
 def expected_bins():
     return(['bin_0.25', 'bin_0.50', 'bin_0.75', 'bin_1.00', 'full'])
 
+@pytest.fixture(scope="session")
+def incomplete_bins(expected_bins):
+    return(expected_bins[1:])
+
 
 @pytest.fixture(scope="session")
 def expected_chroms():
@@ -75,18 +79,46 @@ def s3(aws_credentials):
 
 
 @pytest.fixture(scope="session")
-def sham_cov_bucket(s3, expected_bins, expected_chroms):
+def sham_cov_url(s3, expected_bins, expected_chroms):
+    """
+    Create sham bucket with all bins and chroms stubbed out.
+    Return url to of coverage prefix containing bins.
+    """
     bucket_name = 'test-'.join([random.choice(string.ascii_lowercase) for i in range(8)])
+    prefix = 'coverage'
     s3.create_bucket(Bucket=bucket_name,
                      CreateBucketConfiguration={'LocationConstraint': 'atlantis'})
 
     for cbin in expected_bins:
         for chrom in expected_chroms:
             s3.put_object(Bucket=bucket_name,
-                          Key=f'coverage/{cbin}/chr{chrom}.{cbin}.tsv.gz',
+                          Key=f'{prefix}/{cbin}/chr{chrom}.{cbin}.tsv.gz',
+                          Body='sham coverage content')
+            s3.put_object(Bucket=bucket_name,
+                          Key=f'{prefix}/{cbin}/chr{chrom}.{cbin}.tsv.gz.tbi',
+                          Body='sham index content')
+
+    return(f's3://{bucket_name}/{prefix}')
+
+
+@pytest.fixture(scope="session")
+def sham_incomplete_cov_url(s3, incomplete_bins, expected_chroms):
+    """
+    Create sham bucket missing an entire bin of chroms stubs.
+    Return url to of coverage prefix containing bins.
+    """
+    bucket_name = 'test-'.join([random.choice(string.ascii_lowercase) for i in range(8)])
+    prefix = 'coverage'
+    s3.create_bucket(Bucket=bucket_name,
+                     CreateBucketConfiguration={'LocationConstraint': 'atlantis'})
+
+    for cbin in incomplete_bins:
+        for chrom in expected_chroms:
+            s3.put_object(Bucket=bucket_name,
+                          Key=f'{prefix}/{cbin}/chr{chrom}.{cbin}.tsv.gz',
                           Body='sham coverage content')
             s3.put_object(Bucket=bucket_name,
-                          Key=f'coverage/{cbin}/chr{chrom}.{cbin}.tsv.gz.tbi',
+                          Key=f'{prefix}/{cbin}/chr{chrom}.{cbin}.tsv.gz.tbi',
                           Body='sham index content')
 
-    return(bucket_name)
+    return(f's3://{bucket_name}/{prefix}')

tests/core/test_s3_coverage.py

@@ -4,8 +4,8 @@
 from botocore.exceptions import ClientError
 
 
-def test_smokes(sham_cov_bucket):
-    cp = S3CoverageProvider(f's3://{sham_cov_bucket}')
+def test_smokes(sham_cov_url):
+    cp = S3CoverageProvider(sham_cov_url)
     assert(isinstance(cp, CoverageProvider))
 
 
@@ -31,3 +31,48 @@ def test_missing_source_error(mocker):
                                                       MaxKeys=10)
     # Verify informative error thrown
     assert(err.match(fake_msg))
+
+
+def test_catalog_discovery_full(s3, sham_cov_url, expected_bins, expected_chroms):
+    cp = S3CoverageProvider(sham_cov_url)
+
+    # All bins are represented
+    assert(set(cp.catalog.keys()) == set(expected_bins))
+
+    # All chroms are represented in each bin
+    for catalog_bin in cp.catalog.values():
+        assert(set(catalog_bin.keys()) == set(expected_chroms))
+
+
+def test_catalog_readability(sham_cov_url):
+    cp = S3CoverageProvider(sham_cov_url)
+    messages = cp.evaluate_chrom_readability()
+    assert(len(messages) == 0)
+
+
+def test_catalog_readability_denied(mocker, sham_cov_url):
+    cp = S3CoverageProvider(sham_cov_url)
+
+    mock_method = mocker.patch.object(cp.client, 'head_object')
+    fake_err = ClientError({'Error': {'Code': '403', 'Message': 'Forbidden'}}, 'head_object')
+    mock_method.side_effect = fake_err
+
+    messages = cp.evaluate_chrom_readability()
+
+    assert(len(messages) == mock_method.call_count)
+
+
+def test_evaluate_chrom_representation(sham_cov_url):
+    cp = S3CoverageProvider(sham_cov_url)
+    messages = cp.evaluate_chrom_representation()
+    assert(len(messages) == 0)
+
+
+def test_evaluate_chrom_representation_incomplete(
+    sham_incomplete_cov_url, incomplete_bins, expected_bins
+):
+    cp = S3CoverageProvider(sham_incomplete_cov_url)
+    messages = cp.evaluate_chrom_representation()
+
+    num_messages_expected = len(expected_bins) - len(incomplete_bins)
+    assert(len(messages) == num_messages_expected)