[BDC-981] DRS 1.5 integration tests

.secrets.baseline

@@ -298,7 +298,7 @@
         "filename": "gen3-integration-tests/tests/test_drs_endpoint.py",
         "hashed_secret": "62bd0c4d3a6b445b13212d23500a7f0916757c3e",
         "is_verified": false,
-        "line_number": 20,
+        "line_number": 23,
         "is_secret": false
       }
     ],
@@ -384,5 +384,5 @@
       }
     ]
   },
-  "generated_at": "2025-12-18T18:30:59Z"
+  "generated_at": "2026-03-23T23:34:47Z"
 }

gen3-integration-tests/conftest.py

@@ -95,6 +95,29 @@ def get_fence_clients():
     setup.get_rotated_client_id_secret()
 
 
+@pytest.fixture(scope="session")
+def drs_version():
+    """
+    Fetch DRS version from service-info for conditional test gating.
+
+    Returns the DRS spec version from the service-info endpoint.
+    Falls back to 1.2.
+    """
+    from services.drs import Drs
+
+    drs = Drs()
+    try:
+        resp = drs.get_service_info()
+        if resp.status_code == 200:
+            svc = resp.json()
+            return svc.get("type", {}).get("version", "1.2")
+    except Exception as e:
+        logger.error(
+            f"Could not fetch DRS service-info: {e}. Defaulting to DRS 1.2. DRS 1.5 tests will be skipped."
+        )
+    return "1.2"
+
+
 def pytest_configure(config):
     # Compute hostname and namespace
     pytest.hostname = os.getenv("HOSTNAME")

gen3-integration-tests/pyproject.toml

@@ -81,6 +81,7 @@ markers = [
   # features
   "agg_mds: tests for aggregate metadata service",
   "client_credentials: tests for fence client credentials",
+  "drs: tests for DRS (Data Repository Service) endpoints",
   "etl: tests for etl",
   "data_upload: tests for data upload",
   "gen3sdk: tests for gen3sdk",

gen3-integration-tests/services/drs.py

@@ -1,3 +1,4 @@
+import json
 from uuid import uuid4
 
 import pytest
@@ -10,40 +11,86 @@ class Drs(object):
     def __init__(self):
         self.BASE_URL = f"{pytest.root_url}"
         self.DRS_ENDPOINT = "/ga4gh/drs/v1/objects"
+        self.SERVICE_INFO_ENDPOINT = "/ga4gh/drs/v1/service-info"
 
-    def get_drs_object(self, file: dict, user="main_account"):
-        """Get Drs object"""
-        auth = Gen3Auth(refresh_token=pytest.api_keys[user], endpoint=self.BASE_URL)
+    def _auth(self, user: str = "main_account") -> Gen3Auth:
+        return Gen3Auth(refresh_token=pytest.api_keys[user], endpoint=self.BASE_URL)
+
+    @staticmethod
+    def _extract_id(file: dict) -> str | None:
         try:
-            id = file.get("did") or file.get("id")
+            return file.get("did") or file.get("id")
         except Exception:
-            # id is set to None to test the negative test scenario
-            id = None
+            return None
+
+    def get_drs_object(self, file: dict, user="main_account"):
+        """Get Drs object"""
+        auth = self._auth(user)
+        id = self._extract_id(file)
         response = auth.curl(path=f"{self.DRS_ENDPOINT}/{id}")
         return response
 
     def get_drs_signed_url(self, file, user="main_account"):
         """Get Drs signed url"""
-        auth = Gen3Auth(refresh_token=pytest.api_keys[user], endpoint=self.BASE_URL)
-        try:
-            id = file.get("did") or file.get("id")
-        except Exception:
-            # id is set to None to test the negative test scenario
-            id = None
+        auth = self._auth(user)
+        id = self._extract_id(file)
         access_id = file["urls"][0][:2]
         response = auth.curl(path=f"{self.DRS_ENDPOINT}/{id}/access/{access_id}")
         return response
 
     def get_drs_signed_url_without_header(self, file, user="main_account"):
         """Get Drs signed url without header"""
-        auth = Gen3Auth(refresh_token=pytest.api_keys[user], endpoint=self.BASE_URL)
-        try:
-            id = file.get("did") or file.get("id")
-        except Exception:
-            # id is set to None to test the negative test scenario
-            id = None
+        auth = self._auth(user)
+        id = self._extract_id(file)
         access_id = file["urls"][0][:2]
         response = auth.curl(
             path=f"{self.BASE_URL}{self.DRS_ENDPOINT}/{id}/access/{access_id}"
         )
         return response
+
+    def get_service_info(self, user: str = "main_account") -> requests.Response:
+        """Get DRS service info"""
+        auth = self._auth(user)
+        response = auth.curl(path=self.SERVICE_INFO_ENDPOINT)
+        return response
+
+    def get_drs_object_authorizations(
+        self, file: dict, user: str = "main_account"
+    ) -> requests.Response:
+        """Get authorization info for a DRS object (OPTIONS /objects/{id})"""
+        auth = self._auth(user)
+        id = self._extract_id(file)
+        url = f"{self.BASE_URL}{self.DRS_ENDPOINT}/{id}"
+        response = requests.options(url, auth=auth)
+        return response
+
+    def get_bulk_object_authorizations(
+        self, object_ids: list, user: str = "main_account"
+    ) -> requests.Response:
+        """Get bulk authorization info (OPTIONS /objects)"""
+        auth = self._auth(user)
+        url = f"{self.BASE_URL}{self.DRS_ENDPOINT}"
+        response = requests.options(
+            url, json={"bulk_object_ids": object_ids}, auth=auth
+        )
+        return response
+
+    def get_bulk_drs_objects(
+        self, object_ids: list, user: str = "main_account"
+    ) -> requests.Response:
+        """Get multiple DRS objects (POST /objects)"""
+        auth = self._auth(user)
+        body = json.dumps({"bulk_object_ids": object_ids})
+        response = auth.curl(path=self.DRS_ENDPOINT, request="POST", data=body)
+        return response
+
+    def get_bulk_signed_urls(
+        self, bulk_access_ids: list, user: str = "main_account"
+    ) -> requests.Response:
+        """Get bulk presigned URLs (POST /objects/access)"""
+        auth = self._auth(user)
+        body = json.dumps({"bulk_object_access_ids": bulk_access_ids})
+        response = auth.curl(
+            path=f"{self.DRS_ENDPOINT}/access", request="POST", data=body
+        )
+        return response

gen3-integration-tests/services/fence.py

@@ -73,6 +73,47 @@ def create_signed_url(
             return response.json()
         return response
 
+    @retry(times=3, delay=20, exceptions=(AssertionError, Gen3AuthError))
+    def create_bulk_signed_urls(
+        self, guids, user=None, expected_status=200, access_token=None
+    ):
+        """Creates presigned urls for multiple GUIDs"""
+        url = f"{self.DATA_DOWNLOAD_ENDPOINT}/bulk"
+
+        payload = {"guids": guids}
+
+        if user:
+            auth = Gen3Auth(refresh_token=pytest.api_keys[user], endpoint=self.BASE_URL)
+            response = requests.post(
+                self.BASE_URL + url,
+                json=payload,
+                auth=auth,
+            )
+        elif access_token:
+            response = requests.post(
+                self.BASE_URL + url,
+                json=payload,
+                headers={
+                    "Content-Type": "application/json",
+                    "Authorization": f"bearer {access_token}",
+                },
+            )
+        else:
+            response = requests.post(
+                self.BASE_URL + url,
+                json=payload,
+            )
+
+        logger.info("Status code : " + str(response.status_code))
+
+        assert (
+            expected_status == response.status_code
+        ), f"Expected response {expected_status}, but got {response.status_code}"
+
+        if response.status_code == 200:
+            return response.json()
+        return response
+
     def get_url_for_data_upload(self, file_name: str, user: str) -> dict:
         """Generate the url for uploading the data"""
         auth = Gen3Auth(refresh_token=pytest.api_keys[user], endpoint=self.BASE_URL)