Add failing test for export to FTP using tasks

test/integration/test_history_import_export.py

@@ -1,3 +1,4 @@
+import os
 import tarfile
 from uuid import uuid4
 
@@ -41,6 +42,14 @@ class TestImportExportHistoryViaTasksIntegration(
     def handle_galaxy_config_kwds(cls, config):
         PosixFileSourceSetup.handle_galaxy_config_kwds(config, cls)
         UsesCeleryTasks.handle_galaxy_config_kwds(config)
+        cls.setup_ftp_config(config)
+
+    @classmethod
+    def setup_ftp_config(cls, config):
+        ftp_dir = cls.temp_config_dir("ftp")
+        os.makedirs(ftp_dir)
+        config["ftp_upload_dir"] = ftp_dir
+        config["ftp_upload_site"] = "ftp://cow.com"
 
     def setUp(self):
         super().setUp()
@@ -93,6 +102,23 @@ class TestImportExportHistoryViaTasksIntegration(
                 bai_metadata = imported_bam_details["meta_files"][0]
                 assert bai_metadata["file_type"] == "bam_index"
 
+    def test_import_export_ftp(self):
+        history_name = f"for_export_ftp_async_{uuid4()}"
+        history_id = self.dataset_populator.setup_history_for_export_testing(history_name)
+
+        model_store_format = "rocrate.zip"
+        target_uri = f"gxftp://history.{model_store_format}"
+
+        self.dataset_populator.export_history_to_uri_async(history_id, target_uri, model_store_format)
+        self.dataset_populator.import_history_from_uri_async(target_uri, model_store_format)
+
+        last_history = self._get("histories?limit=1").json()
+        assert len(last_history) == 1
+        imported_history = last_history[0]
+        imported_history_id = imported_history["id"]
+        assert imported_history_id != history_id
+        assert imported_history["name"] == history_name
+
 
 class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, UsesCeleryTasks):
     dataset_populator: DatasetPopulator