More upload configuration options testing - this time for libraries.

test/api/test_configuration.py

@@ -29,7 +29,7 @@ class ConfigurationApiTestCase(api.ApiTestCase):
 
     def setUp(self):
         super(ConfigurationApiTestCase, self).setUp()
-        self.library_populator = LibraryPopulator(self)
+        self.library_populator = LibraryPopulator(self.galaxy_interactor)
 
     def test_normal_user_configuration(self):
         config = self._get_configuration()

test/api/test_history_contents.py

@@ -22,7 +22,7 @@ def setUp(self):
         self.history_id = self._new_history()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
         self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
-        self.library_populator = LibraryPopulator(self)
+        self.library_populator = LibraryPopulator(self.galaxy_interactor)
 
     def test_index_hda_summary(self):
         hda1 = self._new_dataset(self.history_id)

test/api/test_libraries.py

@@ -13,7 +13,7 @@ def setUp(self):
         super(LibrariesApiTestCase, self).setUp()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
         self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
-        self.library_populator = LibraryPopulator(self)
+        self.library_populator = LibraryPopulator(self.galaxy_interactor)
 
     def test_create(self):
         data = dict(name="CreateTestLibrary")

test/api/test_tools.py

@@ -259,7 +259,7 @@ def test_multi_select_optional(self):
     @skip_without_tool("library_data")
     def test_library_data_param(self):
         with self.dataset_populator.test_history() as history_id:
-            ld = LibraryPopulator(self).new_library_dataset("lda_test_library")
+            ld = LibraryPopulator(self.galaxy_interactor).new_library_dataset("lda_test_library")
             inputs = {
                 "library_dataset": ld["ldda_id"],
                 "library_dataset_multiple": [ld["ldda_id"], ld["ldda_id"]]

test/base/populators.py

@@ -461,9 +461,8 @@ def import_workflow(self, workflow, **kwds):
 
 class LibraryPopulator(object):
 
-    def __init__(self, api_test_case):
-        self.api_test_case = api_test_case
-        self.galaxy_interactor = api_test_case.galaxy_interactor
+    def __init__(self, galaxy_interactor):
+        self.galaxy_interactor = galaxy_interactor
 
     def new_private_library(self, name):
         library = self.new_library(name)
@@ -500,44 +499,57 @@ def user_email(self):
 
     def user_private_role_id(self):
         user_email = self.user_email()
-        roles_response = self.api_test_case.galaxy_interactor.get("roles", admin=True)
+        roles_response = self.galaxy_interactor.get("roles", admin=True)
         users_roles = [r for r in roles_response.json() if r["name"] == user_email]
         assert len(users_roles) == 1
         return users_roles[0]["id"]
 
     def create_dataset_request(self, library, **kwds):
+        upload_option = kwds.get("upload_option", "upload_file")
         create_data = {
             "folder_id": kwds.get("folder_id", library["root_folder_id"]),
             "create_type": "file",
             "files_0|NAME": kwds.get("name", "NewFile"),
-            "upload_option": kwds.get("upload_option", "upload_file"),
+            "upload_option": upload_option,
             "file_type": kwds.get("file_type", "auto"),
             "db_key": kwds.get("db_key", "?"),
         }
-        files = {
-            "files_0|file_data": kwds.get("file", StringIO(kwds.get("contents", "TestData"))),
-        }
+
+        if upload_option == "upload_file":
+            files = {
+                "files_0|file_data": kwds.get("file", StringIO(kwds.get("contents", "TestData"))),
+            }
+        elif upload_option == "upload_paths":
+            create_data["filesystem_paths"] = kwds["paths"]
+            files = {}
+        elif upload_option == "upload_directory":
+            create_data["server_dir"] = kwds["server_dir"]
+            files = {}
+
         return create_data, files
 
     def new_library_dataset(self, name, **create_dataset_kwds):
         library = self.new_private_library(name)
         payload, files = self.create_dataset_request(library, **create_dataset_kwds)
-        url_rel = "libraries/%s/contents" % (library["id"])
-        dataset = self.api_test_case.galaxy_interactor.post(url_rel, payload, files=files).json()[0]
+        dataset = self.raw_library_contents_create(library["id"], payload, files=files).json()[0]
 
         def show():
-            return self.api_test_case.galaxy_interactor.get("libraries/%s/contents/%s" % (library["id"], dataset["id"]))
+            return self.galaxy_interactor.get("libraries/%s/contents/%s" % (library["id"], dataset["id"]))
 
         wait_on_state(show, timeout=DEFAULT_TIMEOUT)
         return show().json()
 
+    def raw_library_contents_create(self, library_id, payload, files={}):
+        url_rel = "libraries/%s/contents" % library_id
+        return self.galaxy_interactor.post(url_rel, payload, files=files)
+
     def show_ldda(self, library_id, library_dataset_id):
-        return self.api_test_case.galaxy_interactor.get("libraries/%s/contents/%s" % (library_id, library_dataset_id))
+        return self.galaxy_interactor.get("libraries/%s/contents/%s" % (library_id, library_dataset_id))
 
     def new_library_dataset_in_private_library(self, library_name="private_dataset", wait=True):
         library = self.new_private_library(library_name)
         payload, files = self.create_dataset_request(library, file_type="txt", contents="create_test")
-        create_response = self.api_test_case.galaxy_interactor.post("libraries/%s/contents" % library["id"], payload, files=files)
+        create_response = self.galaxy_interactor.post("libraries/%s/contents" % library["id"], payload, files=files)
         api_asserts.assert_status_code_is(create_response, 200)
         library_datasets = create_response.json()
         assert len(library_datasets) == 1

test/integration/test_upload_configuration_options.py

@@ -34,6 +34,7 @@
 )
 from base.populators import (
     DatasetPopulator,
+    LibraryPopulator,
     skip_without_datatype,
 )
 
@@ -49,6 +50,7 @@ class BaseUploadContentConfigurationTestCase(integration_util.IntegrationTestCas
     def setUp(self):
         super(BaseUploadContentConfigurationTestCase, self).setUp()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.library_populator = LibraryPopulator(self.galaxy_interactor)
         self.history_id = self.dataset_populator.new_history()
 
 
@@ -85,6 +87,12 @@ def test_disallowed_for_composite_file(self):
         # the newer API decorator that handles those details.
         assert create_response.status_code >= 400
 
+    def test_disallowed_for_libraries(self):
+        library = self.library_populator.new_private_library("pathpastedisallowedlibraries")
+        payload, files = self.library_populator.create_dataset_request(library, upload_option="upload_paths", paths="%s/1.txt" % TEST_DATA_DIRECTORY)
+        response = self.library_populator.raw_library_contents_create(library["id"], payload, files=files)
+        assert response.status_code == 403, response.json()
+
 
 class AdminsCanPasteFilePathsTestCase(BaseUploadContentConfigurationTestCase):
 
@@ -99,10 +107,16 @@ def test_admin_path_paste(self):
             self.history_id, 'file://%s/random-file' % TEST_DATA_DIRECTORY,
         )
         create_response = self._post("tools", data=payload)
-        # Ideally this would be 403 but the tool API endpoint isn't using
-        # the newer API decorator that handles those details.
+        # Is admin - so this should work fine!
         assert create_response.status_code == 200
 
+    def test_admin_path_paste_libraries(self):
+        library = self.library_populator.new_private_library("pathpasteallowedlibraries")
+        payload, files = self.library_populator.create_dataset_request(library, upload_option="upload_paths", paths="%s/1.txt" % TEST_DATA_DIRECTORY)
+        response = self.library_populator.raw_library_contents_create(library["id"], payload, files=files)
+        # Was 403 for non-admin above.
+        assert response.status_code == 200
+
 
 class DefaultBinaryContentFiltersTestCase(BaseUploadContentConfigurationTestCase):
 
@@ -414,3 +428,45 @@ def _copy_to_user_ftp_file(self, test_data_path):
 
     def _write_user_ftp_file(self, path, content):
         return self._write_ftp_file(os.path.join(self.ftp_dir(), TEST_USER), content, filename=path)
+
+
+class ServerDirectoryOffByDefaultTestCase(BaseUploadContentConfigurationTestCase):
+
+    require_admin_user = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["library_import_dir"] = None
+
+    def test_server_dir_uploads_403_if_dir_not_set(self):
+        library = self.library_populator.new_private_library("serverdiroffbydefault")
+        payload, files = self.library_populator.create_dataset_request(library, upload_option="upload_directory", server_dir="foobar")
+        response = self.library_populator.raw_library_contents_create(library["id"], payload, files=files)
+        assert response.status_code == 403, response.json()
+        assert '"library_import_dir" is not set' in response.json()["err_msg"]
+
+
+class ServerDirectoryValidUsageTestCase(BaseUploadContentConfigurationTestCase):
+
+    require_admin_user = True
+
+    def test_valid_server_dir_uploads_okay(self):
+        library = self.library_populator.new_private_library("serverdirupload")
+        # upload $GALAXY_ROOT/test-data/library
+        payload, files = self.library_populator.create_dataset_request(library, upload_option="upload_directory", server_dir="library")
+        response = self.library_populator.raw_library_contents_create(library["id"], payload, files=files)
+        assert response.status_code == 200, response.json()
+
+
+class ServerDirectoryRestrictedToAdminsUsageTestCase(BaseUploadContentConfigurationTestCase):
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["user_library_import_dir"] = None
+
+    def test_library_import_dir_not_available_to_non_admins(self):
+        # same test case above works for admins
+        library = self.library_populator.new_private_library("serverdirupload")
+        payload, files = self.library_populator.create_dataset_request(library, upload_option="upload_directory", server_dir="library")
+        response = self.library_populator.raw_library_contents_create(library["id"], payload, files=files)
+        assert response.status_code == 403, response.json()