Add url_regex matching for DRS filesources and add integration test

lib/galaxy/files/sources/drs.py

@@ -1,5 +1,10 @@
 import logging
-from typing import Optional
+import re
+from typing import (
+    cast,
+    Dict,
+    Optional,
+)
 
 from typing_extensions import Unpack
 
@@ -13,6 +18,12 @@ from . import (
 log = logging.getLogger(__name__)
 
 
+class DRSFilesSourceProperties(FilesSourceProperties, total=False):
+    url_regex: str
+    force_http: bool
+    http_headers: Dict[str, str]
+
+
 class DRSFilesSource(BaseFilesSource):
     plugin_type = "drs"
 
@@ -24,28 +35,35 @@ class DRSFilesSource(BaseFilesSource):
             writable=False,
         )
         kwds.update(kwd)
-        props = self._parse_common_config_opts(kwds)
+        props: DRSFilesSourceProperties = cast(DRSFilesSourceProperties, self._parse_common_config_opts(kwds))
+        self._url_regex_str = props.pop("url_regex", r"^drs://")
+        assert self._url_regex_str
+        self._url_regex = re.compile(self._url_regex_str)
+        self._force_http = props.pop("force_http", False)
         self._props = props
 
     def _realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
         headers = props.pop("http_headers", {}) or {}
-        fetch_drs_to_file(source_path, native_path, user_context, headers=headers)
+        fetch_drs_to_file(source_path, native_path, user_context, headers=headers, force_http=self._force_http)
 
     def _write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         raise NotImplementedError()
 
     def score_url_match(self, url: str):
-        if url.startswith("drs://"):
-            return len("drs://")
+        match = self._url_regex.match(url)
+        if match:
+            return match.span()[1]
         else:
             return 0
 
-    def _serialization_props(self, user_context=None):
+    def _serialization_props(self, user_context=None) -> DRSFilesSourceProperties:
         effective_props = {}
         for key, val in self._props.items():
             effective_props[key] = self._evaluate_prop(val, user_context=user_context)
-        return effective_props
+        effective_props["url_regex"] = self._url_regex_str
+        effective_props["force_http"] = self._force_http
+        return cast(DRSFilesSourceProperties, effective_props)
 
 
 __all__ = ("DRSFilesSource",)

test/integration/file_sources_conf_remote_files.yml

@@ -0,0 +1,5 @@
+- type: drs
+  id: drshttp
+  url_regex: "^drs://(localhost|127\\.0\\.0\\.1)"
+  force_http: true
+  doc: Test DRS with forced http for local urls

test/integration/test_remote_files.py

@@ -3,6 +3,8 @@ import os
 import shutil
 from tempfile import mkdtemp
 from typing import ClassVar
+from unittest import SkipTest
+from urllib.parse import urlparse
 
 from galaxy.exceptions import error_codes
 from galaxy_test.base.api_asserts import (
@@ -16,7 +18,7 @@ from galaxy_test.base.populators import (
 from galaxy_test.driver import integration_util
 
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
-FILE_SOURCES_JOB_CONF = os.path.join(SCRIPT_DIRECTORY, "file_sources_conf.yml")
+FILE_SOURCES_JOB_CONF = os.path.join(SCRIPT_DIRECTORY, "file_sources_conf_remote_files.yml")
 VCF_GZ_PATH = os.path.join(SCRIPT_DIRECTORY, os.path.pardir, os.path.pardir, "test-data", "test.vcf.gz")
 
 USERNAME = "user--bx--psu--edu"
@@ -50,6 +52,7 @@ class ConfiguresRemoteFilesIntegrationTestCase(integration_util.IntegrationTestC
         config["ftp_upload_purge"] = True
         config["metadata_strategy"] = "extended"
         config["tool_evaluation_strategy"] = "remote"
+        config["file_sources_config_file"] = FILE_SOURCES_JOB_CONF
 
     def setUp(self):
         super().setUp()
@@ -115,23 +118,42 @@ class TestRemoteFilesIntegration(ConfiguresRemoteFilesIntegrationTestCase):
 
         assert os.path.exists(os.path.join(self.library_dir, "a"))
 
-    def test_fetch_from_drs(self):
-        with self.dataset_populator.test_history() as history_id:
-            element = dict(src="url", url="drs://bpa-drs.beta.biocommons.org.au/e949351fd8a07b50e4ed0784a4736823")
-            target = {
-                "destination": {"type": "hdas"},
-                "elements": [element],
-            }
-            targets = [target]
-            payload = {
-                "history_id": history_id,
-                "targets": targets,
-            }
-            new_dataset = self.dataset_populator.fetch(payload, assert_ok=True).json()["outputs"][0]
-            content = self.dataset_populator.get_history_dataset_content(history_id, dataset=new_dataset)
-            assert content == "a\n", content
+    def test_fetch_from_drs(self, celery_worker):
+        # This test needs an additional worker because the md5 hashing task must run in parallel
+        # with the data_fetch task. Otherwise, the data_fetch task runs first, and polls the DRS
+        # endpoint, which in turn dispatches the md5 task and returns a 204 Try Later. The data_fetch
+        # task dutifully tries again but the md5 task still hasn't run, so the drs endpoint dispatches
+        # another md5 ask and so on till the DRS fetch task eventually times out. Once the data_fetch
+        # task exits, all the md5 tasks run. Multiple workers prevent this from happening.
+        CONTENT = "a\n"
+        history_id = self.dataset_populator.new_history()
 
-        assert not os.path.exists(os.path.join(ftp_dir, "a"))
+        def create_drs_object():
+            hda = self.dataset_populator.new_dataset(history_id, content=CONTENT, wait=True)
+            drs_id = hda["drs_id"]
+            components = urlparse(self.url)
+            netloc = components.netloc
+            if components.path != "/":
+                raise SkipTest("Real DRS cannot be served on Galaxy not hosted at root.")
+            drs_uri = f"drs://{netloc}/{drs_id}"
+            return drs_uri
+
+        # Upload a dataset to Galaxy, which will be available over DRS
+        drs_url = create_drs_object()
+        # Download the created DRS object to check drs url handling
+        element = dict(src="url", url=drs_url)
+        target = {
+            "destination": {"type": "hdas"},
+            "elements": [element],
+        }
+        targets = [target]
+        payload = {
+            "history_id": history_id,
+            "targets": targets,
+        }
+        new_dataset = self.dataset_populator.fetch(payload, assert_ok=True).json()["outputs"][0]
+        content = self.dataset_populator.get_history_dataset_content(history_id, dataset=new_dataset)
+        assert content == CONTENT, content
 
     def test_fetch_from_ftp(self):
         ftp_dir = self.user_ftp_dir

test/integration/test_webdav.py

@@ -12,7 +12,7 @@ from galaxy_test.base.populators import DatasetPopulator
 from galaxy_test.driver import integration_util
 
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
-FILE_SOURCES_JOB_CONF = os.path.join(SCRIPT_DIRECTORY, "file_sources_conf.yml")
+FILE_SOURCES_JOB_CONF = os.path.join(SCRIPT_DIRECTORY, "file_sources_conf_webdav.yml")
 
 skip_if_no_webdav = pytest.mark.skipif(not os.environ.get("GALAXY_TEST_WEBDAV"), reason="GALAXY_TEST_WEBDAV not set")