Merge pull request #1196 from uc-cdis/feat/s3_bucket_regex_unit_test

unit test for s3 bucket regex

.secrets.baseline

@@ -314,14 +314,14 @@
         "filename": "tests/data/test_indexed_file.py",
         "hashed_secret": "a62f2225bf70bfaccbc7f1ef2a397836717377de",
         "is_verified": false,
-        "line_number": 411
+        "line_number": 449
       },
       {
         "type": "Secret Keyword",
         "filename": "tests/data/test_indexed_file.py",
         "hashed_secret": "c258a8d1264cc59de81f8b1975ac06732b1cf182",
         "is_verified": false,
-        "line_number": 432
+        "line_number": 470
       }
     ],
     "tests/keys/2018-05-01T21:29:02Z/jwt_private_key.pem": [
@@ -422,5 +422,5 @@
       }
     ]
   },
-  "generated_at": "2024-08-22T19:43:39Z"
+  "generated_at": "2024-11-04T09:20:13Z"
 }

tests/data/conftest.py

@@ -2,6 +2,8 @@
 Fixtures to support tests/data
 """
 import pytest
+from unittest.mock import MagicMock
+from fence.blueprints.data.indexd import S3IndexedFileLocation
 
 
 @pytest.fixture(scope="function", params=("upload", "download"))
@@ -21,3 +23,22 @@ def supported_protocol(request):
     Note that "az" is an internal mapping for a supported protocol
     """
     return request.param
+
+
+@pytest.fixture(params=["invalid_bucket*name", "validbucketname-alreadyvalid"])
+def s3_indexed_file_location(request):
+    """
+    Provides a mock s3 file location instance, parameterized with a valid and invalid bucket_name
+    """
+    mock_url = "only/needed/for/instantiation"
+    location = S3IndexedFileLocation(url=mock_url)
+
+    # Mock parsed_url attributes, made an always valid value for fallback upon failed regex validation
+    location.parsed_url = MagicMock()
+    location.parsed_url.netloc = "validbucketname-netloc"
+    location.parsed_url.path = "/test/object"
+
+    # Set bucket_name based on the parameter, can be valid or invalid
+    location.bucket_name = MagicMock(return_value=request.param)
+
+    return location

tests/data/test_indexed_file.py

@@ -3,13 +3,17 @@
 """
 import json
 from unittest import mock
-from mock import patch
+from mock import patch, MagicMock
 
 import gen3cirrus
 import pytest
 
 import fence.blueprints.data.indexd as indexd
-from fence.blueprints.data.indexd import IndexedFile, GoogleStorageIndexedFileLocation
+from fence.blueprints.data.indexd import (
+    IndexedFile,
+    GoogleStorageIndexedFileLocation,
+    S3IndexedFileLocation,
+)
 from fence.models import (
     AssumeRoleCacheGCP,
     GoogleServiceAccountKey,
@@ -359,6 +363,40 @@ def test_internal_get_signed_url(
                         )
 
 
+@patch("fence.blueprints.data.indexd.get_value")
+def test_get_signed_url_s3_bucket_name(mock_get_value, s3_indexed_file_location, app):
+    # Mock config with buckets for s3_buckets.get(bucket_name)
+    mock_get_value.side_effect = lambda config, key, error: {
+        "AWS_CREDENTIALS": {"aws_access_key_id": "mock_key"},
+        "S3_BUCKETS": {
+            "invalid_bucket*name": {"endpoint_url": "https://custom.endpoint.com"},
+            "validbucketname-alreadyvalid": {
+                "endpoint_url": "https://custom.endpoint2.com"
+            },
+        },
+    }.get(key, error)
+
+    with patch("fence.blueprints.data.indexd.flask.current_app", return_value=app):
+        # patch get_credential_to_access_bucket() ensure get_signed_url can proceed without actually accessing AWS credentials
+        with patch.object(
+            S3IndexedFileLocation, "get_credential_to_access_bucket"
+        ) as mock_get_credential:
+            mock_get_credential.return_value = {
+                "aws_access_key_id": "mock_key",
+                "aws_secret_access_key": "mock_secret",  # pragma: allowlist secret
+            }
+
+            result_url = s3_indexed_file_location.get_signed_url(
+                "download", expires_in=3600
+            )
+
+            # Check that real_bucket_name fell back to parsed_url.netloc, or otherwise used the already valid bucket
+            if s3_indexed_file_location.bucket_name() == "invalid_bucket*name":
+                assert "validbucketname-netloc" in result_url
+            else:
+                assert "validbucketname-alreadyvalid" in result_url
+
+
 @pytest.mark.parametrize("supported_action", ["download"], indirect=True)
 def test_internal_get_signed_url_no_location_match(
     app, supported_action, supported_protocol, indexd_client_accepting_record