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_libraries
import json
import unittest
from galaxy_test.base.populators import (
DatasetCollectionPopulator,
DatasetPopulator,
LibraryPopulator,
TestsDatasets,
)
from ._framework import ApiTestCase
FILE_URL = 'https://raw.githubusercontent.com/galaxyproject/galaxy/dev/test-data/4.bed'
FILE_MD5 = "37b59762b59fff860460522d271bc111"
[docs]class LibrariesApiTestCase(ApiTestCase, TestsDatasets):
[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)
[docs] def test_create(self):
data = dict(name="CreateTestLibrary")
create_response = self._post("libraries", data=data, admin=True)
self._assert_status_code_is(create_response, 200)
library = create_response.json()
self._assert_has_keys(library, "name")
assert library["name"] == "CreateTestLibrary"
[docs] def test_delete(self):
library = self.library_populator.new_library("DeleteTestLibrary")
create_response = self._delete("libraries/%s" % library["id"], admin=True)
self._assert_status_code_is(create_response, 200)
library = create_response.json()
self._assert_has_keys(library, "deleted")
assert library["deleted"] is True
# Test undeleting
data = dict(undelete=True)
create_response = self._delete("libraries/%s" % library["id"], data=data, admin=True)
library = create_response.json()
self._assert_status_code_is(create_response, 200)
assert library["deleted"] is False
[docs] def test_nonadmin(self):
# Anons can't create libs
data = dict(name="CreateTestLibrary")
create_response = self._post("libraries", data=data, admin=False, anon=True)
self._assert_status_code_is(create_response, 403)
# Anons can't delete libs
library = self.library_populator.new_library("AnonDeleteTestLibrary")
create_response = self._delete("libraries/%s" % library["id"], admin=False, anon=True)
self._assert_status_code_is(create_response, 403)
# Anons can't update libs
data = dict(name="ChangedName", description="ChangedDescription", synopsis='ChangedSynopsis')
create_response = self._patch("libraries/%s" % library["id"], data=data, admin=False, anon=True)
self._assert_status_code_is(create_response, 403)
[docs] def test_update(self):
library = self.library_populator.new_library("UpdateTestLibrary")
data = dict(name='ChangedName', description='ChangedDescription', synopsis='ChangedSynopsis')
create_response = self._patch("libraries/%s" % library["id"], data=data, admin=True)
self._assert_status_code_is(create_response, 200)
library = create_response.json()
self._assert_has_keys(library, 'name', 'description', 'synopsis')
assert library['name'] == 'ChangedName'
assert library['description'] == 'ChangedDescription'
assert library['synopsis'] == 'ChangedSynopsis'
[docs] def test_create_private_library_permissions(self):
library = self.library_populator.new_library("PermissionTestLibrary")
library_id = library["id"]
role_id = self.library_populator.user_private_role_id()
self.library_populator.set_permissions(library_id, role_id)
create_response = self._create_folder(library)
self._assert_status_code_is(create_response, 200)
[docs] def test_create_dataset_denied(self):
library = self.library_populator.new_private_library("ForCreateDatasets")
folder_response = self._create_folder(library)
self._assert_status_code_is(folder_response, 200)
folder_id = folder_response.json()[0]['id']
history_id = self.dataset_populator.new_history()
hda_id = self.dataset_populator.new_dataset(history_id, content="1 2 3")['id']
with self._different_user():
payload = {'from_hda_id': hda_id}
create_response = self._post("folders/%s/contents" % folder_id, payload)
self._assert_status_code_is(create_response, 403)
[docs] def test_show_private_dataset_permissions(self):
library, library_dataset = self.library_populator.new_library_dataset_in_private_library("ForCreateDatasets", wait=True)
with self._different_user():
response = self.library_populator.show_ldda(library["id"], library_dataset["id"])
# TODO: this should really be 403 and a proper JSON exception.
self._assert_status_code_is(response, 400)
[docs] def test_create_dataset(self):
library, library_dataset = self.library_populator.new_library_dataset_in_private_library("ForCreateDatasets", wait=True)
self._assert_has_keys(library_dataset, "peek", "data_type")
assert library_dataset["peek"].find("create_test") >= 0
assert library_dataset["file_ext"] == "txt", library_dataset["file_ext"]
[docs] def test_fetch_upload_to_folder(self):
history_id, library, destination = self._setup_fetch_to_folder("flat_zip")
items = [{"src": "files", "dbkey": "hg19", "info": "my cool bed", "created_from_basename": "4.bed"}]
targets = [{
"destination": destination,
"items": items
}]
payload = {
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": json.dumps(targets),
"__files": {"files_0|file_data": open(self.test_data_resolver.get_filename("4.bed"))},
}
self.dataset_populator.fetch(payload)
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/4.bed")
assert dataset["file_size"] == 61, dataset
assert dataset["genome_build"] == "hg19", dataset
assert dataset["misc_info"] == "my cool bed", dataset
assert dataset["file_ext"] == "bed", dataset
assert dataset["created_from_basename"] == "4.bed"
[docs] def test_fetch_zip_to_folder(self):
history_id, library, destination = self._setup_fetch_to_folder("flat_zip")
bed_test_data_path = self.test_data_resolver.get_filename("4.bed.zip")
targets = [{
"destination": destination,
"items_from": "archive", "src": "files",
}]
payload = {
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": json.dumps(targets),
"__files": {"files_0|file_data": open(bed_test_data_path, 'rb')}
}
self.dataset_populator.fetch(payload)
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/4.bed")
assert dataset["file_size"] == 61, dataset
[docs] def test_fetch_single_url_to_folder(self):
library, response = self._fetch_single_url_to_folder()
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/4.bed")
assert dataset["file_size"] == 61, dataset
[docs] def test_fetch_single_url_with_invalid_datatype(self):
_, response = self._fetch_single_url_to_folder('xxx', assert_ok=False)
self._assert_status_code_is(response, 400)
assert response.json()['err_msg'] == "Requested extension 'xxx' unknown, cannot upload dataset."
def _fetch_single_url_to_folder(self, file_type='auto', assert_ok=True):
history_id, library, destination = self._setup_fetch_to_folder("single_url")
items = [{
"src": "url",
"url": FILE_URL,
"MD5": FILE_MD5,
"ext": file_type,
}]
targets = [{
"destination": destination,
"items": items,
}]
payload = {
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": json.dumps(targets),
"validate_hashes": True
}
return library, self.dataset_populator.fetch(payload, assert_ok=assert_ok)
[docs] def test_legacy_upload_unknown_datatype(self):
library = self.library_populator.new_private_library("ForLegacyUpload")
folder_response = self._create_folder(library)
self._assert_status_code_is(folder_response, 200)
folder_id = folder_response.json()[0]['id']
payload = {
'folder_id': folder_id,
'create_type': 'file',
'file_type': 'xxx',
'upload_option': 'upload_file',
'files_0|url_paste': FILE_URL,
}
create_response = self._post("libraries/%s/contents" % library['id'], payload)
self._assert_status_code_is(create_response, 400)
assert create_response.json() == "Requested extension 'xxx' unknown, cannot upload dataset."
[docs] def test_fetch_failed_validation(self):
# Exception handling is really rough here - we should be creating a dataset in error instead
# of just failing the job like this.
history_id, library, destination = self._setup_fetch_to_folder("single_url")
items = [{
"src": "url",
"url": "https://raw.githubusercontent.com/galaxyproject/galaxy/dev/test-data/4.bed",
"MD5": "37b59762b59fff860460522d271bc112",
"name": "4.bed",
}]
targets = [{
"destination": destination,
"items": items,
}]
payload = {
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": json.dumps(targets),
"validate_hashes": True
}
tool_response = self.dataset_populator.fetch(payload, assert_ok=False)
job = self.dataset_populator.check_run(tool_response)
self.dataset_populator.wait_for_job(job["id"])
job = tool_response.json()["jobs"][0]
details = self.dataset_populator.get_job_details(job["id"]).json()
assert details["state"] == "ok", details
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/4.bed")
assert dataset["state"] == "error", dataset
[docs] def test_fetch_url_archive_to_folder(self):
history_id, library, destination = self._setup_fetch_to_folder("single_url")
targets = [{
"destination": destination,
"items_from": "archive",
"src": "url",
"url": "https://raw.githubusercontent.com/galaxyproject/galaxy/dev/test-data/4.bed.zip",
}]
payload = {
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": json.dumps(targets),
}
self.dataset_populator.fetch(payload)
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/4.bed")
assert dataset["file_size"] == 61, dataset
[docs] @unittest.skip("reference URLs changed, checksums now invalid.")
def test_fetch_bagit_archive_to_folder(self):
history_id, library, destination = self._setup_fetch_to_folder("bagit_archive")
example_bag_path = self.test_data_resolver.get_filename("example-bag.zip")
targets = [{
"destination": destination,
"items_from": "bagit_archive", "src": "files",
}]
payload = {
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": json.dumps(targets),
"__files": {"files_0|file_data": open(example_bag_path)},
}
self.dataset_populator.fetch(payload)
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/README.txt")
assert dataset["file_size"] == 66, dataset
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/bdbag-profile.json")
assert dataset["file_size"] == 723, dataset
def _setup_fetch_to_folder(self, test_name):
return self.library_populator.setup_fetch_to_folder(test_name)
[docs] def test_create_dataset_in_folder(self):
library = self.library_populator.new_private_library("ForCreateDatasets")
folder_response = self._create_folder(library)
self._assert_status_code_is(folder_response, 200)
folder_id = folder_response.json()[0]['id']
history_id = self.dataset_populator.new_history()
hda_id = self.dataset_populator.new_dataset(history_id, content="1 2 3")['id']
payload = {'from_hda_id': hda_id}
create_response = self._post("folders/%s/contents" % folder_id, payload)
self._assert_status_code_is(create_response, 200)
self._assert_has_keys(create_response.json(), "name", "id")
[docs] def test_create_dataset_in_subfolder(self):
library = self.library_populator.new_private_library("ForCreateDatasets")
folder_response = self._create_folder(library)
self._assert_status_code_is(folder_response, 200)
folder_id = folder_response.json()[0]['id']
subfolder_response = self._create_subfolder(folder_id)
self._assert_status_code_is(folder_response, 200)
print(subfolder_response.json())
subfolder_id = subfolder_response.json()['id']
history_id = self.dataset_populator.new_history()
hda_id = self.dataset_populator.new_dataset(history_id, content="1 2 3 sub")['id']
payload = {'from_hda_id': hda_id}
create_response = self._post("folders/%s/contents" % subfolder_id, payload)
self._assert_status_code_is(create_response, 200)
self._assert_has_keys(create_response.json(), "name", "id")
dataset_update_time = create_response.json()['update_time']
container_fetch_response = self.galaxy_interactor.get("folders/%s/contents" % folder_id)
container_update_time = container_fetch_response.json()['folder_contents'][0]['update_time']
assert dataset_update_time == container_update_time, container_fetch_response
def _patch_library_dataset(self, library_dataset_id, data):
create_response = self._patch("libraries/datasets/{}".format(library_dataset_id), data=data)
self._assert_status_code_is(create_response, 200)
ld = create_response.json()
library_id = ld['parent_library_id']
return self.library_populator.wait_on_library_dataset(library={'id': library_id}, dataset=ld)
[docs] def test_update_dataset_in_folder(self):
ld = self._create_dataset_in_folder_in_library("ForUpdateDataset", content=">seq\nATGC", wait=True).json()
data = {'name': 'updated_name', 'file_ext': 'fasta', 'misc_info': 'updated_info', 'genome_build': 'updated_genome_build'}
ld_updated = self._patch_library_dataset(ld['id'], data)
self._assert_has_keys(ld_updated, "name", "file_ext", "misc_info", "genome_build")
[docs] def test_update_dataset_tags(self):
ld = self._create_dataset_in_folder_in_library("ForTagtestDataset", wait=True).json()
data = {"tags": ["#Lancelot", "name:Holy Grail", "blue"]}
ld_updated = self._patch_library_dataset(ld['id'], data)
self._assert_has_keys(ld_updated, "tags")
assert ld_updated["tags"] == "name:Lancelot, name:HolyGrail, blue"
[docs] def test_invalid_update_dataset_in_folder(self):
ld = self._create_dataset_in_folder_in_library("ForInvalidUpdateDataset", wait=True).json()
data = {'file_ext': 'nonexisting_type'}
create_response = self._patch("libraries/datasets/{}".format(ld["id"]), data=data)
self._assert_status_code_is(create_response, 400)
assert 'This Galaxy does not recognize the datatype of:' in create_response.json()['err_msg']
[docs] def test_detect_datatype_of_dataset_in_folder(self):
ld = self._create_dataset_in_folder_in_library("ForDetectDataset", wait=True).json()
data = {'file_ext': 'data'}
ld_updated = self._patch_library_dataset(ld['id'], data)
self._assert_has_keys(ld_updated, "file_ext")
assert ld_updated["file_ext"] == "data"
data = {'file_ext': 'auto'}
ld_updated = self._patch_library_dataset(ld['id'], data)
self._assert_has_keys(ld_updated, "file_ext")
assert ld_updated["file_ext"] == "txt"
[docs] def test_ldda_collection_import_to_history_hide_source(self):
self._import_to_history(visible=False)
[docs] def test_import_paired_collection(self):
ld = self._create_dataset_in_folder_in_library("ForHistoryImport").json()
history_id = self.dataset_populator.new_history()
url = "histories/%s/contents" % history_id
collection_name = 'Paired-end data (from library)'
payload = {
'name': collection_name,
'collection_type': 'list:paired',
"type": "dataset_collection",
'element_identifiers': json.dumps([
{
'src': 'new_collection',
'name': 'pair1',
'collection_type': 'paired',
'element_identifiers': [{'name': 'forward', 'src': 'ldda', 'id': ld['id']},
{'name': 'reverse', 'src': 'ldda', 'id': ld['id']}]
}
])
}
new_collection = self._post(url, payload).json()
assert new_collection['name'] == collection_name
pair = new_collection['elements'][0]
assert pair['element_identifier'] == 'pair1'
assert pair['object']['elements'][0]['object']['history_id'] == history_id
def _import_to_history(self, visible=True):
ld = self._create_dataset_in_folder_in_library("ForHistoryImport").json()
history_id = self.dataset_populator.new_history()
url = "histories/%s/contents" % history_id
collection_name = 'new_collection_name'
element_identifer = 'new_element_identifier'
payload = {
"collection_type": "list",
"history_content_type": "dataset_collection",
"model_class": "HistoryDatasetCollectionAssociation",
"history_id": history_id,
"name": collection_name,
"hide_source_items": not visible,
"element_identifiers": json.dumps([{
"id": ld['id'],
"name": element_identifer,
"src": "ldda"}]),
"type": "dataset_collection",
"elements": []
}
new_collection = self._post(url, payload).json()
assert new_collection['name'] == collection_name
assert new_collection['element_count'] == 1
element = new_collection['elements'][0]
assert element['element_identifier'] == element_identifer
assert element['object']['visible'] == visible
[docs] def test_create_datasets_in_library_from_collection(self):
library = self.library_populator.new_private_library("ForCreateDatasetsFromCollection")
folder_response = self._create_folder(library)
self._assert_status_code_is(folder_response, 200)
folder_id = folder_response.json()[0]['id']
history_id = self.dataset_populator.new_history()
hdca_id = self.dataset_collection_populator.create_list_in_history(history_id, contents=["xxx", "yyy"], direct_upload=True).json()["outputs"][0]["id"]
payload = {'from_hdca_id': hdca_id, 'create_type': 'file', 'folder_id': folder_id}
create_response = self._post("libraries/%s/contents" % library['id'], payload)
self._assert_status_code_is(create_response, 200)
[docs] def test_create_datasets_in_folder_from_collection(self):
library = self.library_populator.new_private_library("ForCreateDatasetsFromCollection")
history_id = self.dataset_populator.new_history()
hdca_id = self.dataset_collection_populator.create_list_in_history(history_id, contents=["xxx", "yyy"], direct_upload=True).json()["outputs"][0]["id"]
folder_response = self._create_folder(library)
self._assert_status_code_is(folder_response, 200)
folder_id = folder_response.json()[0]['id']
payload = {'from_hdca_id': hdca_id}
create_response = self._post("folders/%s/contents" % folder_id, payload)
self._assert_status_code_is(create_response, 200)
assert len(create_response.json()) == 2
# Also test that anything different from a flat dataset collection list
# is refused
hdca_pair_id = self.dataset_collection_populator.create_list_of_pairs_in_history(history_id).json()["outputs"][0]['id']
payload = {'from_hdca_id': hdca_pair_id}
create_response = self._post("folders/%s/contents" % folder_id, payload)
self._assert_status_code_is(create_response, 501)
assert create_response.json()['err_msg'] == 'Cannot add nested collections to library. Please flatten your collection first.'
def _create_folder(self, library):
create_data = dict(
folder_id=library["root_folder_id"],
create_type="folder",
name="New Folder",
)
return self._post("libraries/%s/contents" % library["id"], data=create_data)
def _create_subfolder(self, containing_folder_id):
create_data = dict(
description="new subfolder desc",
name="New Subfolder",
)
return self._post("folders/%s" % containing_folder_id, data=create_data)
def _create_dataset_in_folder_in_library(self, library_name, content="1 2 3", wait=False):
library = self.library_populator.new_private_library(library_name)
folder_response = self._create_folder(library)
self._assert_status_code_is(folder_response, 200)
folder_id = folder_response.json()[0]['id']
history_id = self.dataset_populator.new_history()
hda_id = self.dataset_populator.new_dataset(history_id, content=content, wait=wait)['id']
payload = {'from_hda_id': hda_id, 'create_type': 'file', 'folder_id': folder_id}
ld = self._post("libraries/%s/contents" % folder_id, payload)
return ld