diff --git a/test/unit/files/temporary_file_sources_conf.yml b/test/unit/files/temporary_file_sources_conf.yml deleted file mode 100644 index f4b22c967deb..000000000000 --- a/test/unit/files/temporary_file_sources_conf.yml +++ /dev/null @@ -1,5 +0,0 @@ -- type: temp - id: test1 - doc: Test temporal file source - writable: true - root_path: /tmp/test1 diff --git a/test/unit/files/test_temp.py b/test/unit/files/test_temp.py index 44d7b1f6a32c..c4281e9bd042 100644 --- a/test/unit/files/test_temp.py +++ b/test/unit/files/test_temp.py @@ -1,40 +1,42 @@ -import os -import shutil import tempfile from typing import List import pytest -from galaxy.files import ConfiguredFileSources -from galaxy.files.sources import FilesSource +from galaxy.files.plugins import FileSourcePluginsConfig +from galaxy.files.sources.temp import TempFilesSource +from galaxy.files.unittest_utils import ( + setup_root, + TestConfiguredFileSources, +) from ._util import ( assert_realizes_contains, - configured_file_sources, user_context_fixture, ) -SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__)) -FILE_SOURCES_CONF = os.path.join(SCRIPT_DIRECTORY, "temporary_file_sources_conf.yml") - ROOT_URI = "temp://test1" -ROOT_PATH = "/tmp/test1" +TEMP_PLUGIN = { + "type": "temp", + "id": "test1", + "doc": "Test temporal file source", + "writable": True, +} @pytest.fixture(scope="session") -def file_sources() -> ConfiguredFileSources: - file_sources = configured_file_sources(FILE_SOURCES_CONF) +def file_sources() -> TestConfiguredFileSources: + file_sources = _configured_file_sources() return file_sources @pytest.fixture(scope="session") -def temp_file_source(file_sources: ConfiguredFileSources) -> FilesSource: +def temp_file_source(file_sources: TestConfiguredFileSources) -> TempFilesSource: file_source_pair = file_sources.get_file_source_path(ROOT_URI) file_source = file_source_pair.file_source - _populate_test_scenario(file_source) return file_source -def test_file_source(file_sources: FilesSource): +def test_file_source(file_sources: TestConfiguredFileSources): assert_realizes_contains(file_sources, f"{ROOT_URI}/a", "a") assert_realizes_contains(file_sources, f"{ROOT_URI}/b", "b") assert_realizes_contains(file_sources, f"{ROOT_URI}/c", "c") @@ -43,17 +45,17 @@ def test_file_source(file_sources: FilesSource): assert_realizes_contains(file_sources, f"{ROOT_URI}/dir1/sub1/f", "f") -def test_list(temp_file_source: FilesSource): +def test_list(temp_file_source: TempFilesSource): assert_list_names(temp_file_source, "/", recursive=False, expected_names=["a", "b", "c", "dir1"]) assert_list_names(temp_file_source, "/dir1", recursive=False, expected_names=["d", "e", "sub1"]) -def test_list_recursive(temp_file_source: FilesSource): +def test_list_recursive(temp_file_source: TempFilesSource): expected_names = ["a", "b", "c", "dir1", "d", "e", "sub1", "f"] assert_list_names(temp_file_source, "/", recursive=True, expected_names=expected_names) -def test_pagination(temp_file_source: FilesSource): +def test_pagination(temp_file_source: TempFilesSource): # Pagination is only supported for non-recursive listings. recursive = False root_lvl_entries = temp_file_source.list("/", recursive=recursive) @@ -83,12 +85,8 @@ def test_pagination(temp_file_source: FilesSource): assert result[2] == root_lvl_entries[3] -def _populate_test_scenario(file_source: FilesSource): +def _populate_test_scenario(file_source: TempFilesSource): """Create a directory structure in the file source.""" - if os.path.exists(ROOT_PATH): - shutil.rmtree(ROOT_PATH) - os.mkdir(ROOT_PATH) - user_context = user_context_fixture() _upload_to(file_source, "/a", content="a", user_context=user_context) @@ -99,14 +97,24 @@ def _populate_test_scenario(file_source: FilesSource): _upload_to(file_source, "/dir1/sub1/f", content="f", user_context=user_context) -def _upload_to(file_source: FilesSource, target_uri: str, content: str, user_context=None): +def _upload_to(file_source: TempFilesSource, target_uri: str, content: str, user_context=None): with tempfile.NamedTemporaryFile(mode="w") as f: f.write(content) f.flush() file_source.write_from(target_uri, f.name, user_context=user_context) -def assert_list_names(file_source: FilesSource, uri: str, recursive: bool, expected_names: List[str]): +def assert_list_names(file_source: TempFilesSource, uri: str, recursive: bool, expected_names: List[str]): result = file_source.list(uri, recursive=recursive) assert sorted([entry["name"] for entry in result]) == sorted(expected_names) return result + + +def _configured_file_sources() -> TestConfiguredFileSources: + tmp, root = setup_root() + file_sources_config = FileSourcePluginsConfig() + plugin = TEMP_PLUGIN + plugin["root_path"] = root + file_sources = TestConfiguredFileSources(file_sources_config, conf_dict={TEMP_PLUGIN["id"]: plugin}, test_root=root) + _populate_test_scenario(file_sources.get_file_source_path(ROOT_URI).file_source) + return file_sources