Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy_test.api.test_folder_contents

from typing import (
    Any,
    List,
    Optional,
    Tuple,
)

from galaxy_test.base.decorators import requires_new_library
from galaxy_test.base.populators import (
    DatasetCollectionPopulator,
    DatasetPopulator,
    LibraryPopulator,
)
from ._framework import ApiTestCase


[docs]class TestFolderContentsApi(ApiTestCase): dataset_populator: DatasetPopulator
[docs] def setUp(self): super().setUp() self.dataset_populator = DatasetPopulator(self.galaxy_interactor) self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor) self.library_populator = LibraryPopulator(self.galaxy_interactor) self.library = self.library_populator.new_private_library("FolderContentsTestsLibrary") self.root_folder_id = self._create_folder_in_library("Test Folder Contents")
[docs] @requires_new_library def test_create_hda_with_ldda_message(self, history_id): hda_id = self._create_hda(history_id) ldda_message = "Test message" data = { "from_hda_id": hda_id, "ldda_message": ldda_message, } ldda = self._create_content_in_folder_with_payload(self.root_folder_id, data) self._assert_has_keys(ldda, "name", "id")
[docs] @requires_new_library def test_create_hdca_with_ldda_message(self, history_id): contents = ["dataset01", "dataset02"] hdca_id = self._create_hdca_with_contents(history_id, contents) ldda_message = "Test message" data = { "from_hdca_id": hdca_id, "ldda_message": ldda_message, } lddas = self._create_content_in_folder_with_payload(self.root_folder_id, data) assert len(contents) == len(lddas)
[docs] @requires_new_library def test_index(self, history_id): folder_id = self._create_folder_in_library("Test Folder Contents Index") self._create_subfolder_in(folder_id) self._create_dataset_in_folder(history_id, folder_id) response = self._get(f"folders/{folder_id}/contents") self._assert_index_count_is_correct(response, expected_contents_count=2)
[docs] @requires_new_library def test_index_include_deleted(self, history_id): folder_name = "Test Folder Contents Index include deleted" folder_id = self._create_folder_in_library(folder_name) sub_folder_id = self._create_subfolder_in(folder_id) ldda_id, _ = self._create_dataset_in_folder(history_id, folder_id) self._delete_library_dataset(ldda_id) self._delete_subfolder(sub_folder_id) response = self._get(f"folders/{folder_id}/contents") self._assert_index_count_is_correct(response, expected_contents_count=0) include_deleted = True response = self._get(f"folders/{folder_id}/contents?include_deleted={include_deleted}") index_response = self._assert_index_count_is_correct(response, expected_contents_count=2) for content in index_response["folder_contents"]: assert content["deleted"] is True
[docs] @requires_new_library def test_index_pagination(self, history_id): folder_name = "Test Folder Contents Pagination" folder_id = self._create_folder_in_library(folder_name) num_subfolders = 5 for index in range(num_subfolders): self._create_subfolder_in(folder_id, name=f"Folder_{index}") num_datasets = 5 for _ in range(num_datasets): self._create_dataset_in_folder(history_id, folder_id) total_items = num_datasets + num_subfolders response = self._get(f"folders/{folder_id}/contents") index_response = self._assert_index_count_is_correct(response, expected_contents_count=total_items) original_contents = index_response["folder_contents"] limit = 7 response = self._get(f"folders/{folder_id}/contents?limit={limit}") index_response = self._assert_index_count_is_correct( response, expected_contents_count=limit, expected_total_count=total_items ) limit = 20 response = self._get(f"folders/{folder_id}/contents?limit={limit}") index_response = self._assert_index_count_is_correct(response, expected_contents_count=total_items) offset = 3 response = self._get(f"folders/{folder_id}/contents?offset={offset}") index_response = self._assert_index_count_is_correct( response, expected_contents_count=total_items - offset, expected_total_count=total_items ) offset = 20 response = self._get(f"folders/{folder_id}/contents?offset={offset}") index_response = self._assert_index_count_is_correct( response, expected_contents_count=0, expected_total_count=total_items ) limit = 4 offset = 4 response = self._get(f"folders/{folder_id}/contents?limit={limit}&offset={offset}") index_response = self._assert_index_count_is_correct( response, expected_contents_count=limit, expected_total_count=total_items ) contents = index_response["folder_contents"] expected_query_result = original_contents[offset : offset + limit] for index in range(limit): assert contents[index]["id"] == expected_query_result[index]["id"] limit = 20 offset = 6 response = self._get(f"folders/{folder_id}/contents?limit={limit}&offset={offset}") actual_limit = limit if limit < total_items else total_items - offset index_response = self._assert_index_count_is_correct( response, expected_contents_count=actual_limit, expected_total_count=total_items ) contents = index_response["folder_contents"] expected_query_result = original_contents[offset : offset + actual_limit] for index in range(actual_limit): assert contents[index]["id"] == expected_query_result[index]["id"]
[docs] @requires_new_library def test_index_search_text(self, history_id): folder_name = "Test Folder Contents Index search text" folder_id = self._create_folder_in_library(folder_name) dataset_names = ["AB", "BX", "abx"] for name in dataset_names: self._create_dataset_in_folder(history_id, folder_id, name) subfolder_names = ["Folder_a", "Folder_X"] for name in subfolder_names: self._create_subfolder_in(folder_id, name) all_names = dataset_names + subfolder_names search_terms = ["A", "B", "X"] for search_text in search_terms: matching_names = [name for name in all_names if search_text.casefold() in name.casefold()] response = self._get(f"folders/{folder_id}/contents?search_text={search_text}") index_response = self._assert_index_count_is_correct(response, expected_contents_count=len(matching_names)) contents = index_response["folder_contents"] for content in contents: assert search_text.casefold() in content["name"].casefold()
[docs] @requires_new_library def test_index_permissions(self, history_id): folder_name = "Test Folder Contents Index permissions" folder_id = self._create_folder_in_library(folder_name) _, hda_id = self._create_dataset_in_folder(history_id, folder_id) self._make_dataset_private(hda_id) # Owner can access response = self._get(f"folders/{folder_id}/contents") self._assert_index_count_is_correct(response, expected_contents_count=1) # Admins can access response = self._get(f"folders/{folder_id}/contents", admin=True) self._assert_index_count_is_correct(response, expected_contents_count=1) # Other users can't access with self._different_user(): # Without access to the parent library the user gets a 404... should it be a 403 instead? response = self._get(f"folders/{folder_id}/contents") self._assert_status_code_is(response, 404) # Grant library access to this user different_user_role_id = self.dataset_populator.user_private_role_id() self._allow_library_access_to_user_role(different_user_role_id) # Runs as admin # The user can access the library folder but not the private dataset in it response = self._get(f"folders/{folder_id}/contents") self._assert_index_count_is_correct(response, expected_contents_count=0) # Grant access to this user self._allow_dataset_access(hda_id) response = self._get(f"folders/{folder_id}/contents") self._assert_index_count_is_correct(response, expected_contents_count=1)
[docs] @requires_new_library def test_index_permissions_include_deleted(self, history_id) -> None: folder_name = "Test Folder Contents Index permissions include deleted" folder_id = self._create_folder_in_library(folder_name) num_subfolders = 5 subfolder_ids: List[str] = [] deleted_subfolder_ids: List[str] = [] for index in range(num_subfolders): ldda_id = self._create_subfolder_in(folder_id, name=f"Folder_{index}") subfolder_ids.append(ldda_id) for index, subfolder_id in enumerate(subfolder_ids): if index % 2 == 0: self._delete_subfolder(subfolder_id) deleted_subfolder_ids.append(subfolder_id) num_datasets = 5 ldda_ids: List[str] = [] deleted_ldda_ids: List[str] = [] for _ in range(num_datasets): ldda_id, _ = self._create_dataset_in_folder(history_id, folder_id) ldda_ids.append(ldda_id) for index, ldda_id in enumerate(ldda_ids): if index % 2 == 0: self._delete_library_dataset(ldda_id) deleted_ldda_ids.append(ldda_id) num_total_contents = num_subfolders + num_datasets num_non_deleted = num_total_contents - len(deleted_subfolder_ids) - len(deleted_ldda_ids) # Verify deleted contents are not listed include_deleted = False response = self._get(f"folders/{folder_id}/contents?include_deleted={include_deleted}") self._assert_index_count_is_correct(response, expected_contents_count=num_non_deleted) include_deleted = True # Admins can see everything... response = self._get(f"folders/{folder_id}/contents?include_deleted={include_deleted}", admin=True) self._assert_index_count_is_correct(response, expected_contents_count=num_total_contents) # Owner can see everything too response = self._get(f"folders/{folder_id}/contents?include_deleted={include_deleted}") self._assert_index_count_is_correct(response, expected_contents_count=num_total_contents) # Users with access but no modify permission can't see deleted with self._different_user(): different_user_role_id = self.dataset_populator.user_private_role_id() self._allow_library_access_to_user_role(different_user_role_id) # Runs as admin response = self._get(f"folders/{folder_id}/contents?include_deleted={include_deleted}") self._assert_index_count_is_correct(response, expected_contents_count=num_non_deleted)
[docs] @requires_new_library def test_index_order_by(self, history_id): folder_name = "Test Folder Contents Index Order By" folder_id = self._create_folder_in_library(folder_name) subfolder_names = ["Folder_A", "Folder_B", "Folder_C"] subfolder_descriptions = ["Description Z", "Description Y", "Description X"] for index, name in enumerate(subfolder_names): self._create_subfolder_in(folder_id, name, subfolder_descriptions[index]) dataset_names = ["a", "b", "c"] ldda_messages = ["Message Z", "Message Y", "Message X"] dataset_sizes = [50, 100, 10] file_types = ["txt", "csv", "json"] for index, name in enumerate(dataset_names): self._create_dataset_in_folder( history_id, folder_id, name, content=f"{'0' * dataset_sizes[index]}", ldda_message=ldda_messages[index], file_type=file_types[index], ) # Wait for datasets to finish upload self.dataset_populator.wait_for_history(history_id) # Folders always have priority (they show-up before any dataset regardless of the sorting) and they # can only be sorted by name, description and update_time, the other sorting attributes are ignored sort_desc = False order_by = "name" expected_order_by_name = ["Folder_A", "Folder_B", "Folder_C", "a", "b", "c"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "description" expected_order_by_name = ["Folder_C", "Folder_B", "Folder_A", "c", "b", "a"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "type" expected_order_by_name = ["Folder_A", "Folder_B", "Folder_C", "b", "c", "a"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "size" expected_order_by_name = ["Folder_A", "Folder_B", "Folder_C", "c", "a", "b"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "update_time" expected_order_by_name = ["Folder_A", "Folder_B", "Folder_C", "a", "b", "c"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) sort_desc = True order_by = "name" expected_order_by_name = ["Folder_C", "Folder_B", "Folder_A", "c", "b", "a"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "description" expected_order_by_name = ["Folder_A", "Folder_B", "Folder_C", "a", "b", "c"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "type" expected_order_by_name = ["Folder_A", "Folder_B", "Folder_C", "a", "c", "b"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "size" expected_order_by_name = ["Folder_A", "Folder_B", "Folder_C", "b", "a", "c"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name) order_by = "update_time" expected_order_by_name = ["Folder_C", "Folder_B", "Folder_A", "c", "b", "a"] self._assert_folder_order_by_is_expected(folder_id, order_by, sort_desc, expected_order_by_name)
def _assert_folder_order_by_is_expected( self, folder_id: str, order_by: str, sort_desc: str, expected_order_by_name: List[str] ): response = self._get(f"folders/{folder_id}/contents?order_by={order_by}&sort_desc={sort_desc}") index_response = self._assert_index_count_is_correct( response, expected_contents_count=len(expected_order_by_name) ) for index, item in enumerate(index_response["folder_contents"]): assert item["name"] == expected_order_by_name[index] def _assert_index_count_is_correct( self, raw_response, expected_contents_count: int, expected_total_count: Optional[int] = None ) -> dict: self._assert_status_code_is(raw_response, 200) if expected_total_count is None: expected_total_count = expected_contents_count index_response = raw_response.json() metadata = index_response["metadata"] contents = index_response["folder_contents"] assert metadata["total_rows"] == expected_total_count, "Expected total rows doesn't match" assert len(contents) == expected_contents_count, "Expected number of contents doesn't match" return index_response def _create_folder_in_library(self, name: str) -> str: root_folder_id = self.library["root_folder_id"] return self._create_subfolder_in(root_folder_id, name) def _create_subfolder_in( self, folder_id: str, name: Optional[str] = None, description: Optional[str] = None ) -> str: data = { "name": name or "Test Folder", "description": description or f"The description of {name}", } create_response = self._post(f"folders/{folder_id}", data=data, json=True) self._assert_status_code_is(create_response, 200) folder = create_response.json() return folder["id"] def _create_dataset_in_folder( self, history_id: str, folder_id: str, name: Optional[str] = None, content: Optional[str] = None, ldda_message: Optional[str] = None, **kwds, ) -> Tuple[str, str]: """Returns a tuple with the LDDA ID and the underlying HDA ID""" hda_id = self._create_hda(history_id, name, content, **kwds) data = { "from_hda_id": hda_id, "ldda_message": ldda_message or "Test msg", } ldda = self._create_content_in_folder_with_payload(folder_id, data) return ldda["id"], hda_id def _create_content_in_folder_with_payload(self, folder_id: str, payload) -> Any: create_response = self._post(f"folders/{folder_id}/contents", data=payload, json=True) self._assert_status_code_is(create_response, 200) return create_response.json() def _create_hda(self, history_id: str, name: Optional[str] = None, content: Optional[str] = None, **kwds) -> str: hda = self.dataset_populator.new_dataset(history_id, name=name, content=content, **kwds) hda_id = hda["id"] return hda_id def _create_hdca_with_contents(self, history_id: str, contents: List[str]) -> str: hdca = self.dataset_collection_populator.create_list_in_history( history_id, contents=contents, direct_upload=True, wait=True ).json()["outputs"][0] hdca_id = hdca["id"] return hdca_id def _delete_library_dataset(self, ldda_id: str) -> None: delete_response = self._delete(f"libraries/datasets/{ldda_id}") self._assert_status_code_is(delete_response, 200) def _delete_subfolder(self, folder_id: str) -> None: delete_response = self._delete(f"folders/{folder_id}") self._assert_status_code_is(delete_response, 200) def _allow_library_access_to_user_role(self, role_id: str): library_id = self.library["id"] action = "set_permissions" data = { "access_ids[]": role_id, } response = self._post(f"libraries/{library_id}/permissions?action={action}", data=data, admin=True, json=True) self._assert_status_code_is(response, 200) def _allow_dataset_access(self, dataset_id: str): payload = {"action": "remove_restrictions"} update_response = self._put(f"datasets/{dataset_id}/permissions", payload, admin=True, json=True) self._assert_status_code_is_ok(update_response) def _make_dataset_private(self, dataset_id: str): payload = {"action": "make_private"} update_response = self._put(f"datasets/{dataset_id}/permissions", payload, json=True) self._assert_status_code_is_ok(update_response)