diff --git a/lib/galaxy_test/base/api.py b/lib/galaxy_test/base/api.py index 21cf16c0f2c..9dfb1a235bf 100644 --- a/lib/galaxy_test/base/api.py +++ b/lib/galaxy_test/base/api.py @@ -1,6 +1,10 @@ import os from contextlib import contextmanager -from typing import Optional +from typing import ( + Any, + Dict, + Optional, +) from urllib.parse import ( urlencode, urljoin, @@ -38,7 +42,7 @@ def celery_config(): class UsesCeleryTasks: @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config: Dict[str, Any]) -> None: config["enable_celery_tasks"] = True config["metadata_strategy"] = f'{config.get("metadata_strategy", "directory")}_celery' config.update({"celery_conf": {"broker_url": CELERY_BROKER}}) diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index 1fe5cfa7f6b..db254d95e3c 100644 --- a/lib/galaxy_test/base/populators.py +++ b/lib/galaxy_test/base/populators.py @@ -371,10 +371,10 @@ class BaseDatasetPopulator(BasePopulator): to_posix_lines=True, auto_decompress=True, **kwds, - ) -> dict: - """Create a new history dataset instance (HDA) and return its ID. + ) -> Dict[str, Any]: + """Create a new history dataset instance (HDA). - :returns: the HDA id of the new object + :returns: a dictionary describing the new HDA """ run_response = self.new_dataset_request( history_id, @@ -916,7 +916,7 @@ class BaseDatasetPopulator(BasePopulator): assert isinstance(transform, list) return {t["action"] for t in transform} - def get_history_dataset_details(self, history_id: str, **kwds) -> dict: + def get_history_dataset_details(self, history_id: str, **kwds) -> Dict[str, Any]: dataset_id = self.__history_content_id(history_id, **kwds) details_response = self.get_history_dataset_details_raw(history_id, dataset_id) details_response.raise_for_status() diff --git a/lib/galaxy_test/driver/integration_util.py b/lib/galaxy_test/driver/integration_util.py index 485518c93cc..1e52aba5493 100644 --- a/lib/galaxy_test/driver/integration_util.py +++ b/lib/galaxy_test/driver/integration_util.py @@ -9,7 +9,9 @@ from typing import ( ClassVar, Iterator, Optional, + Type, TYPE_CHECKING, + TypeVar, ) from unittest import ( skip, @@ -199,8 +201,11 @@ class IntegrationTestCase(IntegrationInstance, TestCase): """Unit TestCase with utilities for spinning up Galaxy.""" -def integration_module_instance(clazz): - def _instance(): +IntegrationInstanceObject = TypeVar("IntegrationInstanceObject", bound=IntegrationInstance) + + +def integration_module_instance(clazz: Type[IntegrationInstanceObject]): + def _instance() -> Iterator[IntegrationInstanceObject]: instance = clazz() instance.setUpClass() instance.setUp() diff --git a/test/integration/objectstore/test_objectstore_datatype_upload.py b/test/integration/objectstore/test_objectstore_datatype_upload.py index 60950039484..17712a41c9e 100644 --- a/test/integration/objectstore/test_objectstore_datatype_upload.py +++ b/test/integration/objectstore/test_objectstore_datatype_upload.py @@ -10,6 +10,7 @@ import pytest from galaxy_test.driver import integration_util from ..test_datatype_upload import ( TEST_CASES, + TestData, upload_datatype_helper, UploadTestDatatypeDataTestCase, ) @@ -195,34 +196,38 @@ idle_connection_irods_instance = integration_util.integration_module_instance(Ir @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys())) -def test_upload_datatype_dos_disk_and_disk(distributed_instance, test_data, temp_file): - assert distributed_instance.dataset_populator +def test_upload_datatype_dos_disk_and_disk( + distributed_instance: UploadTestDosDiskAndDiskTestCase, test_data: TestData, temp_file +) -> None: with distributed_instance.dataset_populator.test_history() as history_id: upload_datatype_helper(distributed_instance, test_data, temp_file, history_id) @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys())) -def test_upload_datatype_irods(irods_instance, test_data, temp_file): - assert irods_instance.dataset_populator +def test_upload_datatype_irods( + irods_instance: IrodsUploadTestDatatypeDataTestCase, test_data: TestData, temp_file +) -> None: with irods_instance.dataset_populator.test_history() as history_id: - upload_datatype_helper(irods_instance, test_data, temp_file, True, history_id) + upload_datatype_helper(irods_instance, test_data, temp_file, history_id, True) @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys())) -def test_upload_datatype_dos_irods_and_disk(distributed_and_irods_instance, test_data, temp_file): - assert distributed_and_irods_instance.dataset_populator +def test_upload_datatype_dos_irods_and_disk( + distributed_and_irods_instance: UploadTestDosIrodsAndDiskTestCase, test_data: TestData, temp_file +) -> None: with distributed_and_irods_instance.dataset_populator.test_history() as history_id: upload_datatype_helper(distributed_and_irods_instance, test_data, temp_file, history_id) @pytest.mark.parametrize("test_data", SINGLE_TEST_CASE.values(), ids=list(SINGLE_TEST_CASE.keys())) -def test_upload_datatype_irods_idle_connections(idle_connection_irods_instance, test_data, temp_file): - # Upload a file to iRods - assert idle_connection_irods_instance.dataset_populator +def test_upload_datatype_irods_idle_connections( + idle_connection_irods_instance: IrodsIdleConnectionUploadTestCase, test_data: TestData, temp_file +) -> None: with idle_connection_irods_instance.dataset_populator.test_history() as history_id: - upload_datatype_helper(idle_connection_irods_instance, test_data, temp_file, True, history_id) + upload_datatype_helper(idle_connection_irods_instance, test_data, temp_file, history_id, True) # Get Irods object store's connection pool + assert idle_connection_irods_instance._test_driver.app connection_pool = idle_connection_irods_instance._test_driver.app.object_store.session.pool # Verify the connection pool has 0 active and 1 idle connections diff --git a/test/integration/test_datatype_upload.py b/test/integration/test_datatype_upload.py index 21c3312cf09..babf8496a11 100644 --- a/test/integration/test_datatype_upload.py +++ b/test/integration/test_datatype_upload.py @@ -1,6 +1,7 @@ import collections import os import shutil +from typing import TYPE_CHECKING import pytest @@ -57,13 +58,24 @@ TEST_CASES = collect_test_data(registry) @pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys())) -def test_upload_datatype_auto(instance, test_data, temp_file, celery_session_worker, celery_session_app): - assert instance.dataset_populator +def test_upload_datatype_auto( + instance: UploadTestDatatypeDataTestCase, + test_data: TestData, + temp_file, + celery_session_worker, + celery_session_app, +) -> None: with instance.dataset_populator.test_history() as history_id: upload_datatype_helper(instance, test_data, temp_file, history_id) -def upload_datatype_helper(instance, test_data, temp_file, history_id: str, delete_cache_dir=False): +def upload_datatype_helper( + instance: UploadTestDatatypeDataTestCase, + test_data: TestData, + temp_file, + history_id: str, + delete_cache_dir: bool = False, +) -> None: is_compressed = False for is_method in (is_bz2, is_gzip, is_zip): is_compressed = is_method(test_data.path) @@ -106,6 +118,10 @@ def upload_datatype_helper(instance, test_data, temp_file, history_id: str, dele if delete_cache_dir: # Delete cache directory and then re-create it. This way we confirm # that dataset is fetched from the object store, not from the cache + if TYPE_CHECKING: + from .objectstore.test_objectstore_datatype_upload import BaseObjectstoreUploadTest + + assert isinstance(instance, BaseObjectstoreUploadTest) temp_dir = instance.get_object_store_kwargs()["temp_directory"] cache_dir = temp_dir + "/object_store_cache" shutil.rmtree(cache_dir) diff --git a/test/integration/test_upload_configuration_options.py b/test/integration/test_upload_configuration_options.py index 46a7113664b..830c6f861b3 100644 --- a/test/integration/test_upload_configuration_options.py +++ b/test/integration/test_upload_configuration_options.py @@ -28,6 +28,8 @@ from typing import ( Dict, ) +from requests import Response + from galaxy.util.unittest import TestCase from galaxy_test.base.api_util import TEST_USER from galaxy_test.base.constants import ( @@ -50,12 +52,19 @@ class BaseUploadContentConfigurationInstance(integration_util.IntegrationInstanc dataset_populator: DatasetPopulator framework_tool_and_types = True - def setUp(self): + def setUp(self) -> None: super().setUp() self.dataset_populator = DatasetPopulator(self.galaxy_interactor) self.library_populator = LibraryPopulator(self.galaxy_interactor) - def fetch_target(self, target, history_id: str, assert_ok=False, attach_test_file=False, wait=False): + def fetch_target( + self, + target: Dict[str, Any], + history_id: str, + assert_ok: bool = False, + attach_test_file: bool = False, + wait: bool = False, + ) -> Response: payload: Dict[str, Any] = { "history_id": history_id, "targets": [target], @@ -66,7 +75,7 @@ class BaseUploadContentConfigurationInstance(integration_util.IntegrationInstanc response = self.dataset_populator.fetch(payload, assert_ok=assert_ok, wait=wait) return response - def _write_file(self, dir_path, content, filename="test"): + def _write_file(self, dir_path: str, content: str, filename: str = "test") -> str: """Helper for writing ftp/server dir files.""" self._ensure_directory(dir_path) path = os.path.join(dir_path, filename) @@ -74,7 +83,7 @@ class BaseUploadContentConfigurationInstance(integration_util.IntegrationInstanc f.write(content) return path - def _ensure_directory(self, path): + def _ensure_directory(self, path: str) -> None: if not os.path.exists(path): os.makedirs(path) @@ -84,36 +93,36 @@ class BaseUploadContentConfigurationTestCase(BaseUploadContentConfigurationInsta class TestInvalidFetchRequests(BaseUploadContentConfigurationTestCase): - def test_in_place_not_allowed(self, history_id): + def test_in_place_not_allowed(self, history_id: str) -> None: elements = [{"src": "files", "in_place": False}] target = { "destination": {"type": "hdca"}, "elements": elements, "collection_type": "list", } - response = self.fetch_target(target, history_id=history_id, attach_test_file=True) + response = self.fetch_target(target, history_id, attach_test_file=True) self._assert_status_code_is(response, 400) assert "in_place" in response.json()["err_msg"] - def test_files_not_attached(self, history_id: str): + def test_files_not_attached(self, history_id: str) -> None: elements = [{"src": "files"}] target = { "destination": {"type": "hdca"}, "elements": elements, "collection_type": "list", } - response = self.fetch_target(target, history_id=history_id) + response = self.fetch_target(target, history_id) self._assert_status_code_is(response, 400) assert "Failed to find uploaded file matching target" in response.json()["err_msg"] class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase): @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["allow_path_paste"] = True - def test_disallowed_for_primary_file(self, history_id): + def test_disallowed_for_primary_file(self, history_id: str) -> None: payload = self.dataset_populator.upload_payload( history_id, f"file://{TEST_DATA_DIRECTORY}/1.RData", file_type="binary" ) @@ -123,7 +132,7 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase): assert create_response.status_code >= 400 @skip_without_datatype("velvet") - def test_disallowed_for_composite_file(self, history_id): + def test_disallowed_for_composite_file(self, history_id: str) -> None: path = os.path.join(TEST_DATA_DIRECTORY, "1.txt") assert os.path.exists(path) payload = self.dataset_populator.upload_payload( @@ -143,7 +152,7 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase): assert create_response.status_code >= 400 assert os.path.exists(path) - def test_disallowed_for_libraries(self): + def test_disallowed_for_libraries(self) -> None: path = os.path.join(TEST_DATA_DIRECTORY, "1.txt") assert os.path.exists(path) library = self.library_populator.new_private_library("pathpastedisallowedlibraries") @@ -154,7 +163,7 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase): assert response.status_code == 403, response.json() assert os.path.exists(path) - def test_disallowed_for_fetch(self, history_id: str): + def test_disallowed_for_fetch(self, history_id: str) -> None: path = os.path.join(TEST_DATA_DIRECTORY, "1.txt") assert os.path.exists(path) elements = [{"src": "path", "path": path}] @@ -163,11 +172,11 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase): "elements": elements, "collection_type": "list", } - response = self.fetch_target(target, history_id=history_id) + response = self.fetch_target(target, history_id) self._assert_status_code_is(response, 403) assert os.path.exists(path) - def test_disallowed_for_fetch_urls(self, history_id: str): + def test_disallowed_for_fetch_urls(self, history_id: str) -> None: path = os.path.join(TEST_DATA_DIRECTORY, "1.txt") assert os.path.exists(path) elements = [{"src": "url", "url": f"file://{path}"}] @@ -176,7 +185,7 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase): "elements": elements, "collection_type": "list", } - response = self.fetch_target(target, history_id=history_id) + response = self.fetch_target(target, history_id) self._assert_status_code_is(response, 403) assert os.path.exists(path) @@ -186,11 +195,11 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["allow_path_paste"] = True - def test_admin_path_paste(self, history_id): + def test_admin_path_paste(self, history_id: str) -> None: payload = self.dataset_populator.upload_payload( history_id, f"file://{TEST_DATA_DIRECTORY}/random-file", @@ -199,7 +208,7 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase): # Is admin - so this should work fine! assert create_response.status_code == 200 - def test_admin_path_paste_libraries(self): + def test_admin_path_paste_libraries(self) -> None: library = self.library_populator.new_private_library("pathpasteallowedlibraries") path = f"{TEST_DATA_DIRECTORY}/1.txt" assert os.path.exists(path) @@ -212,7 +221,7 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase): # Test regression where this was getting deleted in this mode. assert os.path.exists(path) - def test_admin_path_paste_libraries_link(self): + def test_admin_path_paste_libraries_link(self) -> None: library = self.library_populator.new_private_library("pathpasteallowedlibraries") path = f"{TEST_DATA_DIRECTORY}/1.txt" assert os.path.exists(path) @@ -226,7 +235,7 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase): # We should probably verify the linking, but this was enough for now to exhibit # https://github.com/galaxyproject/galaxy/issues/8756 - def test_admin_fetch(self, history_id: str): + def test_admin_fetch(self, history_id: str) -> None: path = os.path.join(TEST_DATA_DIRECTORY, "1.txt") elements = [{"src": "path", "path": path}] target = { @@ -234,11 +243,11 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase): "elements": elements, "collection_type": "list", } - response = self.fetch_target(target, history_id=history_id) + response = self.fetch_target(target, history_id) self._assert_status_code_is(response, 200) assert os.path.exists(path) - def test_admin_fetch_file_url(self, history_id: str): + def test_admin_fetch_file_url(self, history_id: str) -> None: path = os.path.join(TEST_DATA_DIRECTORY, "1.txt") elements = [{"src": "url", "url": f"file://{path}"}] target = { @@ -246,7 +255,7 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase): "elements": elements, "collection_type": "list", } - response = self.fetch_target(target, history_id=history_id) + response = self.fetch_target(target, history_id) self._assert_status_code_is(response, 200) assert os.path.exists(path) @@ -256,18 +265,18 @@ class TestDefaultBinaryContentFilters(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["allow_path_paste"] = True - def test_random_binary_allowed(self, history_id: str): + def test_random_binary_allowed(self, history_id: str) -> None: dataset = self.dataset_populator.new_dataset( history_id, f"file://{TEST_DATA_DIRECTORY}/random-file", file_type="auto", wait=True ) dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset) assert dataset["file_ext"] == "binary", dataset - def test_gzipped_html_content_blocked_by_default(self, history_id: str): + def test_gzipped_html_content_blocked_by_default(self, history_id: str) -> None: dataset = self.dataset_populator.new_dataset( history_id, f"file://{TEST_DATA_DIRECTORY}/bad.html.gz", file_type="auto", wait=True, assert_ok=False ) @@ -280,12 +289,12 @@ class TestDisableContentChecking(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["allow_path_paste"] = True config["check_upload_content"] = False - def test_gzipped_html_content_now_allowed(self, history_id: str): + def test_gzipped_html_content_now_allowed(self, history_id: str) -> None: dataset = self.dataset_populator.new_dataset( history_id, f"file://{TEST_DATA_DIRECTORY}/bad.html.gz", file_type="auto", wait=True ) @@ -299,11 +308,11 @@ class TestAutoDecompress(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["allow_path_paste"] = True - def test_auto_decompress_off(self, history_id: str): + def test_auto_decompress_off(self, history_id: str) -> None: dataset = self.dataset_populator.new_dataset( history_id, f"file://{TEST_DATA_DIRECTORY}/1.sam.gz", @@ -314,7 +323,7 @@ class TestAutoDecompress(BaseUploadContentConfigurationTestCase): dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset) assert dataset["file_ext"] == "binary", dataset - def test_auto_decompress_on(self, history_id: str): + def test_auto_decompress_on(self, history_id: str) -> None: dataset = self.dataset_populator.new_dataset( history_id, f"file://{TEST_DATA_DIRECTORY}/1.sam.gz", file_type="auto", wait=True ) @@ -323,7 +332,7 @@ class TestAutoDecompress(BaseUploadContentConfigurationTestCase): class TestLocalAddressWhitelisting(BaseUploadContentConfigurationTestCase): - def test_blocked_url_for_primary_file(self, history_id: str): + def test_blocked_url_for_primary_file(self, history_id: str) -> None: payload = self.dataset_populator.upload_payload(history_id, "http://localhost/", file_type="txt") create_response = self.dataset_populator.tools_post(payload) # Ideally this would be 403 but the tool API endpoint isn't using @@ -331,7 +340,7 @@ class TestLocalAddressWhitelisting(BaseUploadContentConfigurationTestCase): assert create_response.status_code >= 400 @skip_without_datatype("velvet") - def test_blocked_url_for_composite_file(self, history_id: str): + def test_blocked_url_for_composite_file(self, history_id: str) -> None: payload = self.dataset_populator.upload_payload( history_id, "sequences content", @@ -348,20 +357,20 @@ class TestLocalAddressWhitelisting(BaseUploadContentConfigurationTestCase): # the newer API decorator that handles those details. assert create_response.status_code >= 400 - def test_blocked_url_for_fetch(self, history_id: str): + def test_blocked_url_for_fetch(self, history_id: str) -> None: elements = [{"src": "url", "url": "http://localhost"}] target = { "destination": {"type": "hdca"}, "elements": elements, "collection_type": "list", } - response = self.fetch_target(target, history_id=history_id) + response = self.fetch_target(target, history_id) self._assert_status_code_is(response, 403) class BaseFtpUploadConfigurationTestCase(BaseUploadContentConfigurationTestCase): @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) ftp_dir = cls.ftp_dir() os.makedirs(ftp_dir) @@ -369,23 +378,23 @@ class BaseFtpUploadConfigurationTestCase(BaseUploadContentConfigurationTestCase) cls.handle_extra_ftp_config(config) @classmethod - def handle_extra_ftp_config(cls, config): + def handle_extra_ftp_config(cls, config: Dict[str, Any]) -> None: """Overrride to specify additional FTP configuration options.""" @classmethod - def ftp_dir(cls): + def ftp_dir(cls) -> str: return cls.temp_config_dir("ftp") - def _check_content(self, dataset, content, history_id: str, ext="txt"): + def _check_content(self, dataset: Dict[str, Any], content: str, history_id: str, ext: str = "txt") -> None: dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset) assert dataset["file_ext"] == ext, dataset content = self.dataset_populator.get_history_dataset_content(history_id, dataset=dataset) assert content == content, content - def _get_user_ftp_path(self): + def _get_user_ftp_path(self) -> str: return os.path.join(self.ftp_dir(), TEST_USER) - def _write_ftp_file(self, content, filename="test"): + def _write_ftp_file(self, content: str, filename: str = "test") -> str: dir_path = self._get_user_ftp_path() self._ensure_directory(dir_path) path = os.path.join(dir_path, filename) @@ -393,11 +402,11 @@ class BaseFtpUploadConfigurationTestCase(BaseUploadContentConfigurationTestCase) f.write(content) return path - def _ensure_directory(self, path): + def _ensure_directory(self, path: str) -> None: if not os.path.exists(path): os.makedirs(path) - def _run_purgable_upload(self, history_id: str): + def _run_purgable_upload(self, history_id: str) -> str: # Purge setting is actually used with a fairly specific set of parameters - see: # https://github.com/galaxyproject/galaxy/issues/5361 content = "hello world\n" @@ -410,12 +419,12 @@ class BaseFtpUploadConfigurationTestCase(BaseUploadContentConfigurationTestCase) dataset = self.dataset_populator.new_dataset( history_id, ftp_files="test", file_type="txt", to_posix_lines=None, wait=True ) - self._check_content(dataset, content, history_id=history_id) + self._check_content(dataset, content, history_id) return ftp_path class TestSimpleFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase): - def test_ftp_upload(self, history_id: str): + def test_ftp_upload(self, history_id: str) -> None: content = "hello world\n" ftp_path = self._write_ftp_file(content) ftp_files = self.dataset_populator.get_remote_files() @@ -427,9 +436,9 @@ class TestSimpleFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase): dataset = self.dataset_populator.new_dataset( history_id, ftp_files="test", file_type="txt", to_posix_lines=False, wait=True ) - self._check_content(dataset, content, history_id=history_id) + self._check_content(dataset, content, history_id) - def test_ftp_fetch(self, history_id: str): + def test_ftp_fetch(self, history_id: str) -> None: content = "hello world\n" ftp_path = self._write_ftp_file(content) ftp_files = self.dataset_populator.get_remote_files() @@ -443,63 +452,63 @@ class TestSimpleFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase): "collection_type": "list", "name": "cool collection", } - response = self.fetch_target(target, history_id=history_id, assert_ok=True, wait=True) + response = self.fetch_target(target, history_id, assert_ok=True, wait=True) response_object = response.json() assert "output_collections" in response_object output_collections = response_object["output_collections"] assert len(output_collections) == 1, response_object dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=2) - self._check_content(dataset, content, history_id=history_id) + self._check_content(dataset, content, history_id) class TestExplicitEmailAsIdentifierFtpUploadConfiguration(TestSimpleFtpUploadConfiguration): @classmethod - def handle_extra_ftp_config(cls, config): + def handle_extra_ftp_config(cls, config) -> None: config["ftp_upload_dir_identifier"] = "email" class TestPerUsernameFtpUploadConfiguration(TestSimpleFtpUploadConfiguration): @classmethod - def handle_extra_ftp_config(cls, config): + def handle_extra_ftp_config(cls, config) -> None: config["ftp_upload_dir_identifier"] = "username" - def _get_user_ftp_path(self): + def _get_user_ftp_path(self) -> str: username = re.sub("[^a-z-]", "--", TEST_USER.lower()) return os.path.join(self.ftp_dir(), username) class TestTemplatedFtpDirectoryUploadConfiguration(TestSimpleFtpUploadConfiguration): @classmethod - def handle_extra_ftp_config(cls, config): + def handle_extra_ftp_config(cls, config) -> None: config["ftp_upload_dir_template"] = "${ftp_upload_dir}/moo_${ftp_upload_dir_identifier}_cow" - def _get_user_ftp_path(self): + def _get_user_ftp_path(self) -> str: return os.path.join(self.ftp_dir(), f"moo_{TEST_USER}_cow") class TestDisableFtpPurgeUploadConfiguration(BaseFtpUploadConfigurationTestCase): @classmethod - def handle_extra_ftp_config(cls, config): + def handle_extra_ftp_config(cls, config) -> None: config["ftp_upload_purge"] = "False" - def test_ftp_uploads_not_purged(self, history_id: str): - ftp_path = self._run_purgable_upload(history_id=history_id) + def test_ftp_uploads_not_purged(self, history_id: str) -> None: + ftp_path = self._run_purgable_upload(history_id) # Purge is disabled, this better still be here. assert os.path.exists(ftp_path) class TestEnableFtpPurgeUploadConfiguration(BaseFtpUploadConfigurationTestCase): @classmethod - def handle_extra_ftp_config(cls, config): + def handle_extra_ftp_config(cls, config) -> None: config["ftp_upload_purge"] = "True" - def test_ftp_uploads_not_purged(self, history_id: str): - ftp_path = self._run_purgable_upload(history_id=history_id) + def test_ftp_uploads_not_purged(self, history_id: str) -> None: + ftp_path = self._run_purgable_upload(history_id) assert not os.path.exists(ftp_path) class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase): - def test_fetch_ftp_directory(self, history_id: str): + def test_fetch_ftp_directory(self, history_id: str) -> None: dir_path = self._get_user_ftp_path() self._write_file(os.path.join(dir_path, "subdir"), "content 1", filename="1") self._write_file(os.path.join(dir_path, "subdir"), "content 22", filename="2") @@ -511,14 +520,14 @@ class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase): "ftp_path": "subdir", "collection_type": "list", } - self.fetch_target(target, history_id=history_id, assert_ok=True, wait=True) + self.fetch_target(target, history_id, assert_ok=True, wait=True) hdca = self.dataset_populator.get_history_collection_details(history_id, hid=1) assert len(hdca["elements"]) == 3, hdca element0 = hdca["elements"][0] assert element0["element_identifier"] == "1", hdca assert element0["object"]["file_size"] == 9, element0 - def test_fetch_nested_elements_from(self, history_id: str): + def test_fetch_nested_elements_from(self, history_id: str) -> None: dir_path = self._get_user_ftp_path() self._write_file(os.path.join(dir_path, "subdir1"), "content 1", filename="1") self._write_file(os.path.join(dir_path, "subdir1"), "content 22", filename="2") @@ -544,7 +553,7 @@ class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase): "elements": elements, "collection_type": "list:list", } - self.fetch_target(target, history_id=history_id, assert_ok=True, wait=True) + self.fetch_target(target, history_id, assert_ok=True, wait=True) hdca = self.dataset_populator.get_history_collection_details( history_id, history_content_type="dataset_collection" ) @@ -554,7 +563,7 @@ class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase): class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase): - def test_upload_api_option_space_to_tab(self, history_id: str): + def test_upload_api_option_space_to_tab(self, history_id: str) -> None: self._write_user_ftp_file("0.txt", ONE_TO_SIX_WITH_SPACES) self._write_user_ftp_file("1.txt", ONE_TO_SIX_WITH_SPACES) self._write_user_ftp_file("2.txt", ONE_TO_SIX_WITH_SPACES) @@ -591,7 +600,7 @@ class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[2]) assert content == ONE_TO_SIX_WITH_TABS - def test_upload_api_option_to_posix_lines(self, history_id: str): + def test_upload_api_option_to_posix_lines(self, history_id: str) -> None: self._write_user_ftp_file("0.txt", ONE_TO_SIX_ON_WINDOWS) self._write_user_ftp_file("1.txt", ONE_TO_SIX_ON_WINDOWS) self._write_user_ftp_file("2.txt", ONE_TO_SIX_ON_WINDOWS) @@ -628,7 +637,7 @@ class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[2]) assert content == ONE_TO_SIX_WITH_TABS - def test_upload_option_auto_decompress_default(self, history_id: str): + def test_upload_option_auto_decompress_default(self, history_id: str) -> None: self._copy_to_user_ftp_file("1.sam.gz") payload = self.dataset_populator.upload_payload( history_id, @@ -641,7 +650,7 @@ class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=datasets[0]) assert dataset["file_ext"] == "sam", dataset - def test_upload_option_auto_decompress_off(self, history_id: str): + def test_upload_option_auto_decompress_off(self, history_id: str) -> None: self._copy_to_user_ftp_file("1.sam.gz") payload = self.dataset_populator.upload_payload( history_id, @@ -655,13 +664,13 @@ class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=datasets[0]) assert dataset["file_ext"] != "sam", dataset - def _copy_to_user_ftp_file(self, test_data_path): + def _copy_to_user_ftp_file(self, test_data_path: str) -> None: input_path = os.path.join(TEST_DATA_DIRECTORY, test_data_path) target_dir = os.path.join(self.ftp_dir(), TEST_USER) self._ensure_directory(target_dir) shutil.copyfile(input_path, os.path.join(target_dir, test_data_path)) - def _write_user_ftp_file(self, path, content): + def _write_user_ftp_file(self, path: str, content: str) -> str: return self._write_file(os.path.join(self.ftp_dir(), TEST_USER), content, filename=path) @@ -670,11 +679,11 @@ class TestServerDirectoryOffByDefault(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["library_import_dir"] = None - def test_server_dir_uploads_403_if_dir_not_set(self): + def test_server_dir_uploads_403_if_dir_not_set(self) -> None: library = self.library_populator.new_private_library("serverdiroffbydefault") payload, files = self.library_populator.create_dataset_request( library, upload_option="upload_directory", server_dir="foobar" @@ -691,13 +700,13 @@ class TestServerDirectoryValidUsage(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) server_dir = cls.server_dir() os.makedirs(server_dir) config["library_import_dir"] = server_dir - def test_valid_server_dir_uploads_okay(self): + def test_valid_server_dir_uploads_okay(self) -> None: dir_to_import = "library" full_dir_path = os.path.join(self.server_dir(), dir_to_import) os.makedirs(full_dir_path) @@ -716,7 +725,7 @@ class TestServerDirectoryValidUsage(BaseUploadContentConfigurationTestCase): assert library_dataset["file_size"] == 12, library_dataset - def test_link_data_only(self): + def test_link_data_only(self) -> None: content = "hello world\n" dir_path = os.path.join(self.server_dir(), "lib1") file_path = self._write_file(dir_path, content) @@ -733,17 +742,17 @@ class TestServerDirectoryValidUsage(BaseUploadContentConfigurationTestCase): assert ok_dataset["file_name"] == file_path, ok_dataset @classmethod - def server_dir(cls): + def server_dir(cls) -> str: return cls.temp_config_dir("server") class TestUserServerDirectoryOffByDefault(BaseUploadContentConfigurationTestCase): @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["user_library_import_dir"] = None - def test_library_import_dir_not_available_to_non_admins(self): + def test_library_import_dir_not_available_to_non_admins(self) -> None: # same test case above works for admins library = self.library_populator.new_private_library("serverdirupload") payload, files = self.library_populator.create_dataset_request( @@ -755,17 +764,17 @@ class TestUserServerDirectoryOffByDefault(BaseUploadContentConfigurationTestCase class TestUserServerDirectoryValidUsage(BaseUploadContentConfigurationTestCase): @classmethod - def user_server_dir(cls): + def user_server_dir(cls) -> str: return cls.temp_config_dir("user_library_import_dir") @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) user_server_dir = cls.user_server_dir() os.makedirs(user_server_dir) config["user_library_import_dir"] = os.path.join(user_server_dir) - def test_valid_user_server_dir_uploads_okay(self): + def test_valid_user_server_dir_uploads_okay(self) -> None: dir_to_import = "library" full_dir_path = os.path.join(self.user_server_dir(), TEST_USER, dir_to_import) os.makedirs(full_dir_path) @@ -790,11 +799,11 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["allow_path_paste"] = True - def test_fetch_path_to_folder(self): + def test_fetch_path_to_folder(self) -> None: history_id, library, destination = self.library_populator.setup_fetch_to_folder("simple_fetch") bed_test_data_path = self.test_data_resolver.get_filename("4.bed") assert os.path.exists(bed_test_data_path) @@ -809,7 +818,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase): assert dataset["file_size"] == 61, dataset assert os.path.exists(bed_test_data_path) - def test_fetch_link_data_only(self): + def test_fetch_link_data_only(self) -> None: history_id, library, destination = self.library_populator.setup_fetch_to_folder("fetch_and_link") bed_test_data_path = self.test_data_resolver.get_filename("4.bed") assert os.path.exists(bed_test_data_path) @@ -825,7 +834,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase): assert dataset["file_name"] == bed_test_data_path, dataset assert os.path.exists(bed_test_data_path) - def test_fetch_recursive_archive(self): + def test_fetch_recursive_archive(self) -> None: history_id, library, destination = self.library_populator.setup_fetch_to_folder("recursive_archive") archive_test_data_path = self.test_data_resolver.get_filename("testdir1.zip") targets = [ @@ -850,7 +859,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase): dataset = self.library_populator.get_library_contents_with_path(library["id"], "/dir1/file3") assert dataset["file_size"] == 11, dataset - def test_fetch_history_compressed_type(self, history_id: str): + def test_fetch_history_compressed_type(self, history_id: str) -> None: destination = {"type": "hdas"} archive = self.test_data_resolver.get_filename("1.fastqsanger.gz") targets = [ @@ -877,7 +886,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase): assert contents[0]["name"] == "1.fastqsanger.gz", contents[0] assert contents[0]["hid"] == 1, contents[0] - def test_fetch_recursive_archive_history(self, history_id: str): + def test_fetch_recursive_archive_history(self, history_id: str) -> None: destination = {"type": "hdas"} archive = self.test_data_resolver.get_filename("testdir1.zip") targets = [ @@ -898,7 +907,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase): contents = contents_response.json() assert len(contents) == 3 - def test_fetch_recursive_archive_to_library(self, history_id: str): + def test_fetch_recursive_archive_to_library(self, history_id: str) -> None: bed_test_data_path = self.test_data_resolver.get_filename("testdir1.zip") targets = [ { @@ -926,11 +935,11 @@ class TestDirectoryAndCompressedTypes(BaseUploadContentConfigurationTestCase): require_admin_user = True @classmethod - def handle_galaxy_config_kwds(cls, config): + def handle_galaxy_config_kwds(cls, config) -> None: super().handle_galaxy_config_kwds(config) config["allow_path_paste"] = True - def test_tar_to_directory(self, history_id: str): + def test_tar_to_directory(self, history_id: str) -> None: dataset = self.dataset_populator.new_dataset( history_id, f"file://{TEST_DATA_DIRECTORY}/testdir.tar",