Check access level of bucket, modify unit test

astrocut/asdf_cutouts.py

@@ -4,6 +4,7 @@
 import copy
 import pathlib
 from typing import Union, Tuple
+import requests
 
 import asdf
 import astropy
@@ -25,19 +26,15 @@ def _get_cloud_http(s3_uri: Union[str, S3Path]) -> str:
     s3_uri : string | S3Path
         the S3 URI of the cloud resource
     """
-    # create file system and get URL of file
-    try:
-        fs = s3fs.S3FileSystem(anon=True)
-        with fs.open(s3_uri, 'rb') as f:
-            return f.url()
-    except PermissionError:
-        # work-around for Roman Science Platform when acccessing private resources on the cloud
-        fs = s3fs.S3FileSystem()
-        with fs.open(s3_uri, 'rb') as f:
-            return f.url()
 
+    # check if public or private by sending an HTTP request
+    s3_path = S3Path.from_uri(s3_uri) if isinstance(s3_uri, str) else s3_uri
+    url = f'https://{s3_path.bucket}.s3.amazonaws.com/{s3_path.key}'
+    resp = requests.head(url)
+    is_anon = False if resp.status_code == 403 else True
 
-    # open resource and get URL
+    # create file system and get URL of file
+    fs = s3fs.S3FileSystem(anon=is_anon)
     with fs.open(s3_uri, 'rb') as f:
         return f.url()
 

astrocut/tests/test_asdf_cut.py

@@ -315,27 +315,37 @@ def test_slice_gwcs(fakedata):
     assert (gwcsobj.footprint(bounding_box=tuple(reversed(cutout.bbox_original))) == sliced.footprint()).all()
 
 
+@patch('requests.head')
 @patch('s3fs.S3FileSystem')
-def test_get_cloud_http(mock_s3fs):
+def test_get_cloud_http(mock_s3fs, mock_requests):
     """ test we can get HTTP URI of cloud resource """
+    # mock HTTP response
+    mock_resp = MagicMock()
+    mock_resp.status_code = 200  # public bucket
+    mock_requests.return_value = mock_resp
+
     # mock s3 file system operations
     HTTP_URI = "http_test"
+    mock_fs = mock_s3fs.return_value
     mock_file = MagicMock()
-    mock_fs = MagicMock()
     mock_file.url.return_value = HTTP_URI
     mock_fs.open.return_value.__enter__.return_value = mock_file
-    mock_s3fs.return_value = mock_fs
 
     # test function with string input
     s3_uri = "s3://test_bucket/test_file.asdf"
     http_uri = _get_cloud_http(s3_uri)
     assert http_uri == HTTP_URI
-    mock_s3fs.assert_called_once_with(anon=True)
+    mock_s3fs.assert_called_with(anon=True)
     mock_fs.open.assert_called_once_with(s3_uri, 'rb')
     mock_file.url.assert_called_once()
 
     # test function with S3Path input
-    s3_uri_path = S3Path("test_bucket/test_file_2.asdf")
+    s3_uri_path = S3Path("/test_bucket/test_file_2.asdf")
     http_uri_path = _get_cloud_http(s3_uri_path)
     assert http_uri_path == HTTP_URI
     mock_fs.open.assert_called_with(s3_uri_path, 'rb')
+
+    # test function with private bucket
+    mock_resp.status_code = 403
+    http_uri = _get_cloud_http(s3_uri)
+    mock_s3fs.assert_called_with(anon=False)