Merge branch 'release_22.01' into dev

lib/galaxy/celery/__init__.py

@@ -126,6 +126,7 @@ def get_cleanup_short_term_storage_interval():
     celery_app_kwd["backend"] = backend
 
 celery_app = Celery("galaxy", **celery_app_kwd)
+celery_app.set_default()
 
 # setup cron like tasks...
 beat_schedule: Dict[str, Dict[str, Any]] = {}

lib/galaxy/config/sample/galaxy.yml.sample

@@ -25,7 +25,7 @@ gravity:
   # galaxy_root:
 
   # Set to a directory that should contain log files for the processes controlled by Gravity.
-  # If not specified defaults to ``<state_dir>/logs``.
+  # If not specified defaults to ``<state_dir>/log``.
   # log_dir:
 
   # Set to Galaxy's virtualenv directory.
@@ -77,6 +77,13 @@ gravity:
     # Valid options are: DEBUG, INFO, WARNING, ERROR
     # loglevel: DEBUG
 
+    # Queues to join
+    # queues: celery,galaxy.internal,galaxy.external
+
+    # Pool implementation
+    # Valid options are: prefork, eventlet, gevent, solo, processes, threads
+    # pool: threads
+
     # Extra arguments to pass to Celery command line.
     # extra_args:
 

lib/galaxy/datatypes/data.py

@@ -360,7 +360,7 @@ def _archive_composite_dataset(self, trans, data, headers: Headers, do_action="z
                     continue
         if not error:
             headers.update(archive.get_headers())
-            return archive.response(), headers
+            return archive, headers
         return trans.show_error_message(msg), headers
 
     def __archive_extra_files_path(self, extra_files_path):

lib/galaxy/dependencies/dev-requirements.txt

@@ -70,7 +70,7 @@ fs==2.4.15
 funcsigs==1.0.2
 future==0.18.2; (python_version >= "2.6" and python_full_version < "3.0.0") or (python_full_version >= "3.3.0")
 galaxy-sequence-utils==1.1.5
-gravity==0.11.0; python_version >= "3.6"
+gravity==0.12.0; python_version >= "3.6"
 greenlet==1.1.2; python_version >= "3" and python_full_version < "3.0.0" and (platform_machine == "aarch64" or platform_machine == "ppc64le" or platform_machine == "x86_64" or platform_machine == "amd64" or platform_machine == "AMD64" or platform_machine == "win32" or platform_machine == "WIN32") and (python_version >= "2.7" and python_full_version < "3.0.0" or python_full_version >= "3.6.0") and (python_version >= "3.6" and python_full_version < "3.0.0" or python_full_version >= "3.6.0" and python_version >= "3.6") and (python_version >= "3.6" and python_full_version < "3.0.0" and python_version < "4.0" or python_version >= "3.6" and python_version < "4.0" and python_full_version >= "3.6.0") or python_version >= "3" and (platform_machine == "aarch64" or platform_machine == "ppc64le" or platform_machine == "x86_64" or platform_machine == "amd64" or platform_machine == "AMD64" or platform_machine == "win32" or platform_machine == "WIN32") and (python_version >= "2.7" and python_full_version < "3.0.0" or python_full_version >= "3.6.0") and (python_version >= "3.6" and python_full_version < "3.0.0" or python_full_version >= "3.6.0" and python_version >= "3.6") and (python_version >= "3.6" and python_full_version < "3.0.0" and python_version < "4.0" or python_version >= "3.6" and python_version < "4.0" and python_full_version >= "3.6.0") and python_full_version >= "3.5.0"
 gunicorn==20.1.0; python_version >= "3.5"
 gxformat2==0.15.0

lib/galaxy/dependencies/pinned-requirements.txt

@@ -58,7 +58,7 @@ fs==2.4.15
 funcsigs==1.0.2
 future==0.18.2; (python_version >= "2.6" and python_full_version < "3.0.0") or (python_full_version >= "3.3.0")
 galaxy-sequence-utils==1.1.5
-gravity==0.11.0; python_version >= "3.6"
+gravity==0.12.0; python_version >= "3.6"
 greenlet==1.1.2; python_version >= "3" and python_full_version < "3.0.0" and (platform_machine == "aarch64" or platform_machine == "ppc64le" or platform_machine == "x86_64" or platform_machine == "amd64" or platform_machine == "AMD64" or platform_machine == "win32" or platform_machine == "WIN32") and (python_version >= "2.7" and python_full_version < "3.0.0" or python_full_version >= "3.6.0") and (python_version >= "3.6" and python_full_version < "3.0.0" or python_full_version >= "3.6.0" and python_version >= "3.6") and (python_version >= "3.6" and python_full_version < "3.0.0" and python_version < "4.0" or python_version >= "3.6" and python_version < "4.0" and python_full_version >= "3.6.0") or python_version >= "3" and (platform_machine == "aarch64" or platform_machine == "ppc64le" or platform_machine == "x86_64" or platform_machine == "amd64" or platform_machine == "AMD64" or platform_machine == "win32" or platform_machine == "WIN32") and (python_version >= "2.7" and python_full_version < "3.0.0" or python_full_version >= "3.6.0") and (python_version >= "3.6" and python_full_version < "3.0.0" or python_full_version >= "3.6.0" and python_version >= "3.6") and (python_version >= "3.6" and python_full_version < "3.0.0" and python_version < "4.0" or python_version >= "3.6" and python_version < "4.0" and python_full_version >= "3.6.0") and python_full_version >= "3.5.0"
 gunicorn==20.1.0; python_version >= "3.5"
 gxformat2==0.15.0

lib/galaxy/jobs/__init__.py

@@ -16,6 +16,7 @@
 from typing import (
     Any,
     Dict,
+    Iterable,
     List,
     TYPE_CHECKING,
 )
@@ -795,7 +796,7 @@ def get_destination(self, id_or_tag):
             id_or_tag = self.default_destination_id
         return copy.deepcopy(self._get_single_item(self.destinations[id_or_tag]))
 
-    def get_destinations(self, id_or_tag):
+    def get_destinations(self, id_or_tag) -> Iterable[JobDestination]:
         """Given a destination ID or tag, return all JobDestinations matching the provided ID or tag
 
         :param id_or_tag: A destination ID or tag.
@@ -806,7 +807,7 @@ def get_destinations(self, id_or_tag):
         Destinations are not deepcopied, so they should not be passed to
         anything which might modify them.
         """
-        return self.destinations.get(id_or_tag, None)
+        return self.destinations.get(id_or_tag, [])
 
     def get_job_runner_plugins(self, handler_id):
         """Load all configured job runner plugins

lib/galaxy/jobs/command_factory.py

@@ -286,7 +286,7 @@ def capture_stdout_stderr(self, stdout_file, stderr_file):
         if TRAP_KILL_CONTAINER in self.commands:
             # We need to replace the container kill trap with one that removes the named pipes and kills the container
             self.commands = self.commands.replace(TRAP_KILL_CONTAINER, "")
-            trap_command = """trap 'rm "$__out" "$__err"; _on_exit' EXIT"""
+            trap_command = """trap 'rm "$__out" "$__err" 2> /dev/null || true; _on_exit' EXIT"""
         self.prepend_command(
             f"""__out="${{TMPDIR:-.}}/out.$$" __err="${{TMPDIR:-.}}/err.$$"
 mkfifo "$__out" "$__err"

lib/galaxy/model/__init__.py

@@ -8325,7 +8325,7 @@ def store(cls, server_url, association):
             assoc = cls.sa_session.query(cls).filter_by(server_url=server_url, handle=association.handle)[0]
         except IndexError:
             assoc = cls(server_url=server_url, handle=association.handle)
-        assoc.secret = base64.encodestring(association.secret).decode()
+        assoc.secret = base64.encodebytes(association.secret).decode()
         assoc.issued = association.issued
         assoc.lifetime = association.lifetime
         assoc.assoc_type = association.assoc_type

lib/galaxy/structured_app.py

@@ -37,6 +37,7 @@
 from galaxy.workflow.trs_proxy import TrsProxy
 
 if TYPE_CHECKING:
+    from galaxy.jobs import JobConfiguration
     from galaxy.tools.data import ToolDataTableManager
 
 
@@ -94,7 +95,7 @@ class MinimalManagerApp(MinimalApp):
     role_manager: Any  # 'galaxy.managers.roles.RoleManager'
     installed_repository_manager: Any  # 'galaxy.tool_shed.galaxy_install.installed_repository_manager.InstalledRepositoryManager'
     user_manager: Any
-    job_config: Any  # 'galaxy.jobs.JobConfiguration'
+    job_config: "JobConfiguration"
     job_manager: Any  # galaxy.jobs.manager.JobManager
 
     @property
@@ -149,7 +150,7 @@ class StructuredApp(MinimalManagerApp):
     installed_repository_manager: Any  # 'galaxy.tool_shed.galaxy_install.installed_repository_manager.InstalledRepositoryManager'
     workflow_scheduling_manager: Any  # 'galaxy.workflow.scheduling_manager.WorkflowSchedulingManager'
     interactivetool_manager: Any
-    job_config: Any  # 'galaxy.jobs.JobConfiguration'
+    job_config: "JobConfiguration"
     job_manager: Any  # galaxy.jobs.manager.JobManager
     user_manager: Any
     api_keys_manager: Any  # 'galaxy.managers.api_keys.ApiKeyManager'

lib/galaxy/util/zipstream.py

@@ -21,10 +21,7 @@ def response(self):
         if self.upstream_mod_zip:
             yield "\n".join(self.files).encode()
         else:
-            yield iter(self.archive)
-
-    def get_iterator(self):
-        return iter(self.archive)
+            yield from iter(self.archive)
 
     def get_headers(self):
         headers = {}

lib/galaxy/webapps/galaxy/api/datasets.py

@@ -35,6 +35,7 @@
     DatasetSourceType,
     UpdateDatasetPermissionsPayload,
 )
+from galaxy.util.zipstream import ZipstreamWrapper
 from galaxy.webapps.galaxy.api.common import (
     get_filter_query_params,
     get_query_parameters_from_request_excluding,
@@ -252,6 +253,8 @@ def display(
             file_name = getattr(display_data, "name", None)
             if file_name:
                 return FileResponse(file_name, headers=headers)
+        elif isinstance(display_data, ZipstreamWrapper):
+            return StreamingResponse(display_data.response(), headers=headers)
         return StreamingResponse(display_data, headers=headers)
 
     @router.get(

lib/galaxy/webapps/galaxy/api/history_contents.py

@@ -479,9 +479,7 @@ def download_dataset_collection(
         while maintaining approximate collection structure.
         """
         archive = self.service.get_dataset_collection_archive_for_download(trans, id)
-        if archive.upstream_mod_zip:
-            return StreamingResponse(archive.response(), headers=archive.get_headers())
-        return StreamingResponse(archive.get_iterator(), headers=archive.get_headers(), media_type="application/zip")
+        return StreamingResponse(archive.response(), headers=archive.get_headers())
 
     @router.post(
         "/api/histories/{history_id}/contents/dataset_collections/{id}/prepare_download",
@@ -710,9 +708,7 @@ def archive(
         archive = self.service.archive(trans, history_id, filter_query_params, filename, dry_run)
         if isinstance(archive, HistoryContentsArchiveDryRunResult):
             return archive
-        if archive.upstream_mod_zip:
-            return StreamingResponse(archive.response(), headers=archive.get_headers())
-        return StreamingResponse(archive.get_iterator(), headers=archive.get_headers(), media_type="application/zip")
+        return StreamingResponse(archive.response(), headers=archive.get_headers())
 
     @router.get(
         "/api/histories/{history_id}/contents/{direction}/{hid}/{limit}",

lib/galaxy/webapps/galaxy/controllers/dataset.py

@@ -37,6 +37,7 @@
     smart_str,
 )
 from galaxy.util.sanitize_html import sanitize_html
+from galaxy.util.zipstream import ZipstreamWrapper
 from galaxy.web import form_builder
 from galaxy.web.framework.helpers import iff
 from galaxy.webapps.base.controller import (
@@ -214,6 +215,9 @@ def display(
         display_data, headers = data.datatype.display_data(
             trans, data, preview, filename, to_ext, offset=offset, ck_size=ck_size, **kwd
         )
+        if isinstance(display_data, ZipstreamWrapper):
+            trans.response.headers.update(headers)
+            return display_data.response()
         trans.response.headers.update(headers)
         return display_data
 

lib/galaxy_test/api/test_datasets.py

@@ -1,4 +1,6 @@
 import textwrap
+import zipfile
+from io import BytesIO
 from typing import (
     Dict,
     List,
@@ -7,6 +9,7 @@
 from galaxy_test.base.populators import (
     DatasetCollectionPopulator,
     DatasetPopulator,
+    skip_without_datatype,
     skip_without_tool,
 )
 from ._framework import ApiTestCase
@@ -361,3 +364,24 @@ def _delete_batch_with_payload(self, payload):
         self._assert_status_code_is_ok(delete_response)
         deleted_result = delete_response.json()
         return deleted_result
+
+    @skip_without_datatype("velvet")
+    def test_composite_datatype_download(self):
+        item = {
+            "src": "composite",
+            "ext": "velvet",
+            "composite": {
+                "items": [
+                    {"src": "pasted", "paste_content": "sequences content"},
+                    {"src": "pasted", "paste_content": "roadmaps content"},
+                    {"src": "pasted", "paste_content": "log content"},
+                ]
+            },
+        }
+        output = self.dataset_populator.fetch_hda(self.history_id, item, wait=True)
+        print(output)
+        response = self._get(f"histories/{self.history_id}/contents/{output['id']}/display?to_ext=zip")
+        self._assert_status_code_is(response, 200)
+        archive = zipfile.ZipFile(BytesIO(response.content))
+        namelist = archive.namelist()
+        assert len(namelist) == 4, f"Expected 3 elements in [{namelist}]"

test/unit/app/jobs/test_command_factory.py

@@ -57,7 +57,8 @@ def test_kill_trap_replaced(self):
         self.include_work_dir_outputs = False
         self.job_wrapper.command_line = f"{TRAP_KILL_CONTAINER}{MOCK_COMMAND_LINE}"
         expected_command_line = self._surround_command(MOCK_COMMAND_LINE).replace(
-            """trap 'rm "$__out" "$__err"' EXIT""", """trap 'rm "$__out" "$__err"; _on_exit' EXIT"""
+            """trap 'rm "$__out" "$__err"' EXIT""",
+            """trap 'rm "$__out" "$__err" 2> /dev/null || true; _on_exit' EXIT""",
         )
         self.__assert_command_is(expected_command_line)