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_histories
import time
from typing import ClassVar
from unittest import SkipTest
from uuid import uuid4
from requests import put
from galaxy.model.unittest_utils.store_fixtures import (
history_model_store_dict,
TEST_HISTORY_NAME,
)
from galaxy_test.api.sharable import SharingApiTests
from galaxy_test.base.api_asserts import assert_has_keys
from galaxy_test.base.decorators import (
requires_admin,
requires_new_user,
)
from galaxy_test.base.populators import (
DatasetCollectionPopulator,
DatasetPopulator,
skip_without_tool,
)
from ._framework import ApiTestCase
[docs]class BaseHistories:
def _show(self, history_id):
return self._get(f"histories/{history_id}").json()
def _update(self, history_id, data):
update_url = self._api_url(f"histories/{history_id}", use_key=True)
put_response = put(update_url, json=data)
return put_response
def _create_history(self, name):
post_data = dict(name=name)
create_response = self._post("histories", data=post_data).json()
self._assert_has_keys(create_response, "name", "id")
assert create_response["name"] == name
return create_response
def _assert_history_length(self, history_id, n):
contents_response = self._get(f"histories/{history_id}/contents")
self._assert_status_code_is(contents_response, 200)
contents = contents_response.json()
assert len(contents) == n, contents
[docs]class TestHistoriesApi(ApiTestCase, BaseHistories):
[docs] def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
[docs] def test_create_history(self):
# Create a history.
create_response = self._create_history("TestHistory1")
created_id = create_response["id"]
# Make sure new history appears in index of user's histories.
index_response = self._get("histories").json()
indexed_history = [h for h in index_response if h["id"] == created_id][0]
assert indexed_history["name"] == "TestHistory1"
[docs] def test_create_history_json(self):
name = "TestHistoryJson"
post_data = dict(name=name)
create_response = self._post("histories", data=post_data, json=True).json()
self._assert_has_keys(create_response, "name", "id")
assert create_response["name"] == name
return create_response
[docs] def test_show_history(self):
history_id = self._create_history("TestHistoryForShow")["id"]
show_response = self._show(history_id)
self._assert_has_key(
show_response, "id", "name", "annotation", "size", "contents_url", "state", "state_details", "state_ids"
)
state_details = show_response["state_details"]
state_ids = show_response["state_ids"]
states = [
"discarded",
"empty",
"error",
"failed_metadata",
"new",
"ok",
"paused",
"queued",
"running",
"setting_metadata",
"upload",
]
assert isinstance(state_details, dict)
assert isinstance(state_ids, dict)
self._assert_has_keys(state_details, *states)
self._assert_has_keys(state_ids, *states)
[docs] def test_show_history_returns_expected_urls(self):
# This test can be dropped when the URL attributes become deprecated
history_id = self._create_history("TestHistoryForUrls")["id"]
show_response = self._show(history_id)
self._assert_has_key(show_response, "id", "url", "contents_url")
assert show_response["url"] == f"/api/histories/{history_id}"
assert show_response["contents_url"] == f"/api/histories/{history_id}/contents"
[docs] def test_show_most_recently_used(self):
history_id = self._create_history("TestHistoryRecent")["id"]
show_response = self._get("histories/most_recently_used").json()
assert show_response["id"] == history_id
[docs] def test_index_order(self):
slightly_older_history_id = self._create_history("TestHistorySlightlyOlder")["id"]
newer_history_id = self._create_history("TestHistoryNewer")["id"]
index_response = self._get("histories").json()
assert index_response[0]["id"] == newer_history_id
assert index_response[1]["id"] == slightly_older_history_id
[docs] def test_index_query(self):
expected_history_name = f"TestHistoryThatMatchQuery_{uuid4()}"
expected_history_id = self._create_history(expected_history_name)["id"]
self._create_history("TestHistoryThatDoesNotMatchQuery")
# Filter by name
query = f"?q=name&qv={expected_history_name}"
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 1
assert index_response[0]["name"] == expected_history_name
# Filter by name and deleted
query = f"?q=name&qv={expected_history_name}&q=deleted&qv=True"
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 0 # Not deleted yet
# Delete the history
self._delete(f"histories/{expected_history_id}")
# Now it should match the query
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 1
assert index_response[0]["name"] == expected_history_name
[docs] def test_index_case_insensitive_contains_query(self):
# Create the histories with a different user to ensure the test
# is not conflicted with the current user's histories.
with self._different_user(f"user_{uuid4()}@bx.psu.edu"):
unique_id = uuid4()
expected_history_name = f"Test History That Match Query_{unique_id}"
self._create_history(expected_history_name)
self._create_history(expected_history_name.upper())
self._create_history(expected_history_name.lower())
self._create_history(f"Another history_{uuid4()}")
name_contains = "history"
query = f"?q=name-contains&qv={name_contains}"
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 4
name_contains = "history that match query"
query = f"?q=name-contains&qv={name_contains}"
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 3
name_contains = "ANOTHER"
query = f"?q=name-contains&qv={name_contains}"
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 1
name_contains = "test"
query = f"?q=name-contains&qv={name_contains}"
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 3
name_contains = unique_id
query = f"?q=name-contains&qv={name_contains}"
index_response = self._get(f"histories{query}").json()
assert len(index_response) == 3
[docs] def test_delete(self):
# Setup a history and ensure it is in the index
history_id = self._create_history("TestHistoryForDelete")["id"]
index_response = self._get("histories").json()
assert index_response[0]["id"] == history_id
show_response = self._show(history_id)
assert not show_response["deleted"]
# Delete the history
self._delete(f"histories/{history_id}")
# Check can view it - but it is deleted
show_response = self._show(history_id)
assert show_response["deleted"]
# Verify it is dropped from history index
index_response = self._get("histories").json()
assert len(index_response) == 0 or index_response[0]["id"] != history_id
# Add deleted filter to index to view it
index_response = self._get("histories", {"deleted": "true"}).json()
assert index_response[0]["id"] == history_id
[docs] def test_purge(self):
history_id = self._create_history("TestHistoryForPurge")["id"]
data = {"purge": True}
self._delete(f"histories/{history_id}", data=data, json=True)
show_response = self._show(history_id)
assert show_response["deleted"]
assert show_response["purged"]
[docs] def test_undelete(self):
history_id = self._create_history("TestHistoryForDeleteAndUndelete")["id"]
self._delete(f"histories/{history_id}")
self._post(f"histories/deleted/{history_id}/undelete")
show_response = self._show(history_id)
assert not show_response["deleted"]
[docs] def test_update(self):
history_id = self._create_history("TestHistoryForUpdating")["id"]
self._update(history_id, {"name": "New Name"})
show_response = self._show(history_id)
assert show_response["name"] == "New Name"
unicode_name = "桜ゲノム"
self._update(history_id, {"name": unicode_name})
show_response = self._show(history_id)
assert show_response["name"] == unicode_name, show_response
quoted_name = "'MooCow'"
self._update(history_id, {"name": quoted_name})
show_response = self._show(history_id)
assert show_response["name"] == quoted_name
self._update(history_id, {"deleted": True})
show_response = self._show(history_id)
assert show_response["deleted"], show_response
self._update(history_id, {"deleted": False})
show_response = self._show(history_id)
assert not show_response["deleted"]
self._update(history_id, {"published": True})
show_response = self._show(history_id)
assert show_response["published"]
self._update(history_id, {"genome_build": "hg18"})
show_response = self._show(history_id)
assert show_response["genome_build"] == "hg18"
self._update(history_id, {"annotation": "The annotation is cool"})
show_response = self._show(history_id)
assert show_response["annotation"] == "The annotation is cool"
self._update(history_id, {"annotation": unicode_name})
show_response = self._show(history_id)
assert show_response["annotation"] == unicode_name, show_response
self._update(history_id, {"annotation": quoted_name})
show_response = self._show(history_id)
assert show_response["annotation"] == quoted_name
[docs] def test_update_invalid_attribute(self):
history_id = self._create_history("TestHistoryForInvalidUpdating")["id"]
put_response = self._update(history_id, {"invalidkey": "moo"})
assert "invalidkey" not in put_response.json()
[docs] def test_update_invalid_types(self):
history_id = self._create_history("TestHistoryForUpdatingInvalidTypes")["id"]
for str_key in ["name", "annotation"]:
assert self._update(history_id, {str_key: False}).status_code == 400
for bool_key in ["deleted", "importable", "published"]:
assert self._update(history_id, {bool_key: "a string"}).status_code == 400
assert self._update(history_id, {"tags": "a simple string"}).status_code == 400
assert self._update(history_id, {"tags": [True]}).status_code == 400
[docs] def test_invalid_keys(self):
invalid_history_id = "1234123412341234"
assert self._get(f"histories/{invalid_history_id}").status_code == 400
assert self._update(invalid_history_id, {"name": "new name"}).status_code == 400
assert self._delete(f"histories/{invalid_history_id}").status_code == 400
assert self._post(f"histories/deleted/{invalid_history_id}/undelete").status_code == 400
[docs] def test_create_anonymous_fails(self):
post_data = dict(name="CannotCreate")
create_response = self._post("histories", data=post_data, anon=True)
self._assert_status_code_is(create_response, 403)
[docs] @requires_admin
def test_create_without_session_fails(self):
post_data = dict(name="SessionNeeded")
# Using admin=True will boostrap an Admin user without session
create_response = self._post("histories", data=post_data, admin=True, json=True)
self._assert_status_code_is(create_response, 400)
[docs] def test_create_tag(self):
post_data = dict(name="TestHistoryForTag")
history_id = self._post("histories", data=post_data, json=True).json()["id"]
tag_data = dict(value="awesometagvalue")
tag_url = f"histories/{history_id}/tags/awesometagname"
tag_create_response = self._post(tag_url, data=tag_data, json=True)
self._assert_status_code_is(tag_create_response, 200)
[docs] def test_copy_history(self):
history_id = self.dataset_populator.new_history()
fetch_response = self.dataset_collection_populator.create_list_in_history(
history_id, contents=["Hello", "World"], direct_upload=True
)
dataset_collection = self.dataset_collection_populator.wait_for_fetched_collection(fetch_response.json())
copied_history_response = self.dataset_populator.copy_history(history_id)
copied_history_response.raise_for_status()
copied_history = copied_history_response.json()
copied_collection = self.dataset_populator.get_history_collection_details(
history_id=copied_history["id"], history_content_type="dataset_collection"
)
assert dataset_collection["name"] == copied_collection["name"]
assert dataset_collection["id"] != copied_collection["id"]
assert len(dataset_collection["elements"]) == len(copied_collection["elements"]) == 2
source_element = dataset_collection["elements"][0]
copied_element = copied_collection["elements"][0]
assert source_element["element_identifier"] == copied_element["element_identifier"] == "data0"
assert source_element["id"] != copied_element["id"]
source_hda = source_element["object"]
copied_hda = copied_element["object"]
assert source_hda["name"] == copied_hda["name"] == "data0"
assert source_hda["id"] != copied_hda["id"]
assert source_hda["history_id"] != copied_hda["history_id"]
assert source_hda["hid"] == copied_hda["hid"] == 2
# TODO: (CE) test_create_from_copy
[docs] def test_import_from_model_store_dict(self):
response = self.dataset_populator.create_from_store(store_dict=history_model_store_dict())
assert_has_keys(response, "name", "id")
assert response["name"] == TEST_HISTORY_NAME
self._assert_history_length(response["id"], 1)
[docs] def test_anonymous_can_import_published(self):
history_name = f"for_importing_by_anonymous_{uuid4()}"
history_id = self.dataset_populator.new_history(name=history_name)
self.dataset_collection_populator.create_list_of_pairs_in_history(history_id)
self.dataset_populator.make_public(history_id)
with self._different_user(anon=True):
imported_history_name = f"imported_by_anonymous_{uuid4()}"
import_data = {
"archive_type": "url",
"history_id": history_id,
"name": imported_history_name,
}
self.dataset_populator.import_history(import_data)
[docs] def test_immutable_history_update_fails(self):
history_id = self._create_history("TestHistoryForImmutability")["id"]
# we can update the name as usual
self._update(history_id, {"name": "Immutable Name"})
show_response = self._show(history_id)
assert show_response["name"] == "Immutable Name"
# once we purge the history, it becomes immutable
self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
# we cannot update the name anymore
response = self._update(history_id, {"name": "New Name"})
self._assert_status_code_is(response, 403)
assert response.json()["err_msg"] == "History is immutable"
show_response = self._show(history_id)
assert show_response["name"] == "Immutable Name"
[docs] def test_immutable_history_cannot_add_datasets(self):
history_id = self._create_history("TestHistoryForAddImmutability")["id"]
# we add a dataset
self.dataset_populator.new_dataset(history_id, content="TestContents")
# once we purge the history, it becomes immutable
self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
# we cannot add another dataset
with self.assertRaisesRegex(AssertionError, "History is immutable"):
self.dataset_populator.new_dataset(history_id, content="TestContents")
[docs] def test_cannot_modify_tags_on_immutable_history(self):
history_id = self._create_history("TestHistoryForTagImmutability")["id"]
hda = self.dataset_populator.new_dataset(history_id, content="TestContents")
# we add a tag
self._update(history_id, {"tags": ["FirstTag"]})
# once we purge the history, it becomes immutable
self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
# we cannot add another tag
response = self._update(history_id, {"tags": ["SecondTag"]})
self._assert_status_code_is(response, 403)
assert response.json()["err_msg"] == "History is immutable"
# we cannot remove the tag
response = self._update(history_id, {"tags": []})
self._assert_status_code_is(response, 403)
assert response.json()["err_msg"] == "History is immutable"
# we cannot add a tag to the dataset
response = self.dataset_populator.tag_dataset(history_id, hda["id"], ["DatasetTag"], raise_on_error=False)
assert response["err_msg"] == "History is immutable"
[docs] def test_histories_count(self):
# Create a new user so we can test the count without other existing histories
with self._different_user("user_for_count@test.com"):
first_history_id = self._create_history("TestHistoryForCount 1")["id"]
self._assert_expected_histories_count(expected_count=1)
second_history_id = self._create_history("TestHistoryForCount 2")["id"]
self._assert_expected_histories_count(expected_count=2)
third_history_id = self._create_history("TestHistoryForCount 3")["id"]
self._assert_expected_histories_count(expected_count=3)
# Delete the second history
self.dataset_populator.delete_history(second_history_id)
self._assert_expected_histories_count(expected_count=2)
# Archive the first history
self.dataset_populator.archive_history(first_history_id)
self._assert_expected_histories_count(expected_count=1)
# Only the third history should be active
active_histories = self._get("histories").json()
assert len(active_histories) == 1
assert active_histories[0]["id"] == third_history_id
def _assert_expected_histories_count(self, expected_count):
response = self._get("histories/count")
self._assert_status_code_is(response, 200)
assert response.json() == expected_count
[docs]class ImportExportTests(BaseHistories):
task_based: ClassVar[bool]
def _set_up_populators(self):
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
[docs] def test_import_export(self):
history_name = f"for_export_default_{uuid4()}"
history_id = self.dataset_populator.setup_history_for_export_testing(history_name)
imported_history_id = self._reimport_history(history_id, history_name, wait_on_history_length=2)
def upload_job_check(job):
assert job["tool_id"] == "__DATA_FETCH__"
def check_discarded(hda):
assert hda["deleted"]
assert hda["state"] == "discarded", hda
assert hda["purged"] is True
self._check_imported_dataset(history_id=imported_history_id, hid=1, job_checker=upload_job_check)
self._check_imported_dataset(
history_id=imported_history_id,
hid=2,
has_job=False,
hda_checker=check_discarded,
job_checker=upload_job_check,
)
imported_content = self.dataset_populator.get_history_dataset_content(
history_id=imported_history_id,
hid=1,
)
assert imported_content == "1 2 3\n"
[docs] def test_import_1901_histories(self):
f = open(self.test_data_resolver.get_filename("exports/1901_two_datasets.tgz"), "rb")
import_data = dict(archive_source="", archive_file=f)
self._import_history_and_wait(import_data, "API Test History", wait_on_history_length=2)
[docs] def test_import_export_include_deleted(self):
history_name = f"for_export_include_deleted_{uuid4()}"
history_id = self.dataset_populator.new_history(name=history_name)
self.dataset_populator.new_dataset(history_id, content="1 2 3")
deleted_hda = self.dataset_populator.new_dataset(history_id, content="1 2 3", wait=True)
self.dataset_populator.delete_dataset(history_id, deleted_hda["id"])
imported_history_id = self._reimport_history(
history_id, history_name, wait_on_history_length=2, export_kwds={"include_deleted": True}
)
self._assert_history_length(imported_history_id, 2)
def upload_job_check(job):
assert job["tool_id"] == "__DATA_FETCH__"
def check_deleted_not_purged(hda):
assert hda["state"] == "ok", hda
assert hda["deleted"] is True, hda
assert hda["purged"] is False, hda
self._check_imported_dataset(history_id=imported_history_id, hid=1, job_checker=upload_job_check)
self._check_imported_dataset(
history_id=imported_history_id, hid=2, hda_checker=check_deleted_not_purged, job_checker=upload_job_check
)
imported_content = self.dataset_populator.get_history_dataset_content(
history_id=imported_history_id,
hid=1,
)
assert imported_content == "1 2 3\n"
[docs] @skip_without_tool("job_properties")
def test_import_export_failed_job(self):
history_name = f"for_export_include_failed_job_{uuid4()}"
history_id = self.dataset_populator.new_history(name=history_name)
self.dataset_populator.run_tool_raw("job_properties", inputs={"failbool": True}, history_id=history_id)
self.dataset_populator.wait_for_history(history_id, assert_ok=False)
imported_history_id = self._reimport_history(
history_id, history_name, assert_ok=False, wait_on_history_length=4, export_kwds={"include_deleted": True}
)
self._assert_history_length(imported_history_id, 4)
def check_failed(hda_or_job):
print(hda_or_job)
assert hda_or_job["state"] == "error", hda_or_job
self.dataset_populator._summarize_history(imported_history_id)
self._check_imported_dataset(
history_id=imported_history_id, hid=1, assert_ok=False, hda_checker=check_failed, job_checker=check_failed
)
[docs] def test_import_metadata_regeneration(self):
if self.task_based:
raise SkipTest("skipping test_import_metadata_regeneration for task based...")
history_name = f"for_import_metadata_regeneration_{uuid4()}"
history_id = self.dataset_populator.new_history(name=history_name)
self.dataset_populator.new_bam_dataset(history_id, self.test_data_resolver)
imported_history_id = self._reimport_history(history_id, history_name)
self._assert_history_length(imported_history_id, 1)
self._check_imported_dataset(history_id=imported_history_id, hid=1)
import_bam_metadata = self.dataset_populator.get_history_dataset_details(
history_id=imported_history_id,
hid=1,
)
# The cleanup() method of the __IMPORT_HISTORY__ job (which is executed
# after the job has entered its final state):
# - creates a new dataset with 'ok' state and adds it to the history
# - starts a __SET_METADATA__ job to regenerate the dataset metadata, if
# needed
# We need to wait a bit for the creation of the __SET_METADATA__ job.
time.sleep(1)
self.dataset_populator.wait_for_history_jobs(imported_history_id, assert_ok=True)
bai_metadata = import_bam_metadata["meta_files"][0]
assert bai_metadata["file_type"] == "bam_index"
assert "api/" in bai_metadata["download_url"], bai_metadata["download_url"]
api_url = bai_metadata["download_url"].split("api/", 1)[1]
bai_response = self._get(api_url)
self._assert_status_code_is(bai_response, 200)
assert len(bai_response.content) > 4
[docs] def test_import_export_collection(self):
history_name = f"for_export_with_collections_{uuid4()}"
history_id = self.dataset_populator.new_history(name=history_name)
self.dataset_collection_populator.create_list_in_history(
history_id, contents=["Hello", "World"], direct_upload=True, wait=True
)
imported_history_id = self._reimport_history(history_id, history_name, wait_on_history_length=3)
self._assert_history_length(imported_history_id, 3)
def check_elements(elements):
assert len(elements) == 2
element0 = elements[0]["object"]
element1 = elements[1]["object"]
for element in [element0, element1]:
assert not element["visible"]
assert not element["deleted"]
assert element["state"] == "ok"
assert element0["hid"] == 2
assert element1["hid"] == 3
self._check_imported_collection(
imported_history_id, hid=1, collection_type="list", elements_checker=check_elements
)
[docs] def test_import_export_nested_collection(self):
history_name = f"for_export_with_nested_collections_{uuid4()}"
history_id = self.dataset_populator.new_history(name=history_name)
fetch_response = self.dataset_collection_populator.create_list_of_pairs_in_history(history_id, wait=True).json()
dataset_collection = self.dataset_collection_populator.wait_for_fetched_collection(fetch_response)
imported_history_id = self._reimport_history(history_id, history_name, wait_on_history_length=3)
self._assert_history_length(imported_history_id, 3)
def check_elements(elements):
assert len(elements) == 1
element0 = elements[0]["object"]
self._assert_has_keys(element0, "elements", "collection_type")
child_elements = element0["elements"]
assert len(child_elements) == 2
assert element0["collection_type"] == "paired"
self._check_imported_collection(
imported_history_id,
hid=dataset_collection["hid"],
collection_type="list:paired",
elements_checker=check_elements,
)
def _reimport_history(
self, history_id, history_name, wait_on_history_length=None, assert_ok=True, export_kwds=None
):
# Ensure the history is ready to go...
export_kwds = export_kwds or {}
self.dataset_populator.wait_for_history(history_id, assert_ok=assert_ok)
return self.dataset_populator.reimport_history(
history_id,
history_name,
wait_on_history_length=wait_on_history_length,
export_kwds=export_kwds,
task_based=self.task_based,
)
def _import_history_and_wait(self, import_data, history_name, wait_on_history_length=None):
imported_history_id = self.dataset_populator.import_history_and_wait_for_name(import_data, history_name)
if wait_on_history_length:
self.dataset_populator.wait_on_history_length(imported_history_id, wait_on_history_length)
return imported_history_id
def _check_imported_dataset(
self, history_id, hid, assert_ok=True, has_job=True, hda_checker=None, job_checker=None
):
imported_dataset_metadata = self.dataset_populator.get_history_dataset_details(
history_id=history_id,
hid=hid,
assert_ok=assert_ok,
)
assert imported_dataset_metadata["history_content_type"] == "dataset"
assert imported_dataset_metadata["history_id"] == history_id
if hda_checker is not None:
hda_checker(imported_dataset_metadata)
assert "creating_job" in imported_dataset_metadata
job_id = imported_dataset_metadata["creating_job"]
if has_job:
assert job_id
job_details = self.dataset_populator.get_job_details(job_id, full=True)
assert job_details.status_code == 200, job_details.content
job = job_details.json()
assert "history_id" in job, job
assert job["history_id"] == history_id, job
if job_checker is not None:
job_checker(job)
def _check_imported_collection(self, history_id, hid, collection_type=None, elements_checker=None):
imported_collection_metadata = self.dataset_populator.get_history_collection_details(
history_id=history_id,
hid=hid,
)
assert imported_collection_metadata["history_content_type"] == "dataset_collection"
assert imported_collection_metadata["history_id"] == history_id
assert "collection_type" in imported_collection_metadata
assert "elements" in imported_collection_metadata
if collection_type is not None:
assert imported_collection_metadata["collection_type"] == collection_type, imported_collection_metadata
if elements_checker is not None:
elements_checker(imported_collection_metadata["elements"])
[docs]class TestSharingHistory(ApiTestCase, BaseHistories, SharingApiTests):
"""Tests specific for the particularities of sharing Histories."""
api_name = "histories"
[docs] def create(self, name: str) -> str:
response_json = self._create_history(name)
history_id = response_json["id"]
# History to share cannot be empty
populator = DatasetPopulator(self.galaxy_interactor)
populator.new_dataset(history_id)
return history_id
[docs] def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
[docs] @requires_new_user
def test_sharing_with_private_datasets(self):
history_id = self.dataset_populator.new_history()
hda = self.dataset_populator.new_dataset(history_id)
hda_id = hda["id"]
self.dataset_populator.make_private(history_id, hda_id)
with self._different_user():
target_user_id = self.dataset_populator.user_id()
payload = {"user_ids": [target_user_id]}
sharing_response = self._share_history_with_payload(history_id, payload)
# If no share_option is provided, the extra field will contain the
# datasets that need to be accessible before sharing
assert sharing_response["extra"]
assert sharing_response["extra"]["can_share"] is False
assert sharing_response["extra"]["can_change"][0]["id"] == hda_id
assert not sharing_response["users_shared_with"]
# Now we provide the share_option
payload = {"user_ids": [target_user_id], "share_option": "make_accessible_to_shared"}
sharing_response = self._share_history_with_payload(history_id, payload)
assert sharing_response["users_shared_with"]
assert sharing_response["users_shared_with"][0]["id"] == target_user_id
[docs] @requires_admin
@requires_new_user
def test_sharing_without_manage_permissions(self):
history_id = self.dataset_populator.new_history()
hda = self.dataset_populator.new_dataset(history_id)
hda_id = hda["id"]
owner_role_id = self.dataset_populator.user_private_role_id()
with self._different_user():
target_user_id = self.dataset_populator.user_id()
with self._different_user("alice@test.com"):
alice_role_id = self.dataset_populator.user_private_role_id()
# We have one dataset that we cannot manage
payload = {"access": [owner_role_id], "manage": [alice_role_id]}
update_response = self._update_permissions(history_id, hda_id, payload)
self._assert_status_code_is(update_response, 200)
# We will get an error if none of the datasets can be made accessible
payload = {"user_ids": [target_user_id]}
sharing_response = self._share_history_with_payload(history_id, payload)
assert sharing_response["extra"]
assert sharing_response["extra"]["can_share"] is False
assert sharing_response["extra"]["cannot_change"][0]["id"] == hda_id
assert sharing_response["errors"]
assert not sharing_response["users_shared_with"]
# Trying to change the permissions when sharing should fail
# because we don't have manage permissions
payload = {"user_ids": [target_user_id], "share_option": "make_public"}
sharing_response = self._share_history_with_payload(history_id, payload)
assert sharing_response["extra"]
assert sharing_response["extra"]["can_share"] is False
assert sharing_response["errors"]
assert not sharing_response["users_shared_with"]
# we can share if we don't try to make any permission changes
payload = {"user_ids": [target_user_id], "share_option": "no_changes"}
sharing_response = self._share_history_with_payload(history_id, payload)
assert not sharing_response["errors"]
assert sharing_response["users_shared_with"]
assert sharing_response["users_shared_with"][0]["id"] == target_user_id
[docs] @requires_new_user
def test_sharing_empty_not_allowed(self):
history_id = self.dataset_populator.new_history()
with self._different_user():
target_user_id = self.dataset_populator.user_id()
payload = {"user_ids": [target_user_id]}
sharing_response = self._share_history_with_payload(history_id, payload)
assert sharing_response["extra"]["can_share"] is False
assert sharing_response["errors"]
assert "empty" in sharing_response["errors"][0]
[docs] @requires_new_user
def test_sharing_with_duplicated_users(self):
history_id = self.create("HistoryToShareWithDuplicatedUser")
with self._different_user():
target_user_id = self.dataset_populator.user_id()
# Ignore repeated users in the same request
payload = {"user_ids": [target_user_id, target_user_id]}
sharing_response = self._share_history_with_payload(history_id, payload)
assert sharing_response["users_shared_with"]
assert len(sharing_response["users_shared_with"]) == 1
assert sharing_response["users_shared_with"][0]["id"] == target_user_id
[docs] @requires_new_user
def test_sharing_private_history_makes_datasets_public(self):
history_id = self.dataset_populator.new_history()
hda = self.dataset_populator.new_dataset(history_id)
hda_id = hda["id"]
self.dataset_populator.make_private(history_id, hda_id)
# Other users cannot access the dataset
with self._different_user():
show_response = self._get(f"datasets/{hda_id}")
self._assert_status_code_is(show_response, 403)
sharing_response = self._set_resource_sharing(history_id, "publish")
assert sharing_response["published"] is True
# Other users can access the dataset after publishing
with self._different_user():
show_response = self._get(f"datasets/{hda_id}")
self._assert_status_code_is(show_response, 200)
hda = show_response.json()
assert hda["id"] == hda_id
def _share_history_with_payload(self, history_id, payload):
sharing_response = self._put(f"histories/{history_id}/share_with_users", data=payload, json=True)
self._assert_status_code_is(sharing_response, 200)
return sharing_response.json()
def _update_permissions(self, history_id: str, dataset_id: str, payload):
url = f"histories/{history_id}/contents/{dataset_id}/permissions"
update_url = self._api_url(url, **{"use_admin_key": True})
update_response = put(update_url, json=payload)
return update_response
[docs]class TestArchivingHistoriesWithoutExportRecord(ApiTestCase, BaseHistories):
[docs] def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
[docs] def test_archive(self):
history_id = self.dataset_populator.new_history()
history_details = self._show(history_id)
assert history_details["archived"] is False
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 200)
assert archive_response.json()["archived"] is True
history_details = self._show(history_id)
assert history_details["archived"] is True
[docs] def test_other_users_cannot_archive_history(self):
history_id = self.dataset_populator.new_history()
with self._different_user():
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 403)
[docs] def test_restore(self):
history_id = self.dataset_populator.new_history()
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 200)
assert archive_response.json()["archived"] is True
restore_response = self.dataset_populator.restore_archived_history(history_id)
self._assert_status_code_is(restore_response, 200)
assert restore_response.json()["archived"] is False
[docs] def test_other_users_cannot_restore_history(self):
history_id = self.dataset_populator.new_history()
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 200)
assert archive_response.json()["archived"] is True
with self._different_user():
restore_response = self.dataset_populator.restore_archived_history(history_id)
self._assert_status_code_is(restore_response, 403)
[docs] def test_archived_histories_index(self):
with self._different_user("archived_histories_index_user@bx.psu.edu"):
history_id = self.dataset_populator.new_history()
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 200)
assert archive_response.json()["archived"] is True
archived_histories = self.dataset_populator.get_archived_histories()
assert len(archived_histories) == 1
assert archived_histories[0]["id"] == history_id
[docs] def test_archived_histories_filtering_and_sorting(self):
with self._different_user("archived_histories_filtering_user@bx.psu.edu"):
num_histories = 2
history_ids = []
for i in range(num_histories):
history_id = self.dataset_populator.new_history(name=f"History {i}")
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 200)
assert archive_response.json()["archived"] is True
history_ids.append(history_id)
# Filter by name
archived_histories = self.dataset_populator.get_archived_histories(query="q=name-contains&qv=history")
assert len(archived_histories) == num_histories
archived_histories = self.dataset_populator.get_archived_histories(query="q=name-contains&qv=History 1")
assert len(archived_histories) == 1
# Order by name
archived_histories = self.dataset_populator.get_archived_histories(query="order=name-dsc")
assert len(archived_histories) == num_histories
assert archived_histories[0]["name"] == "History 1"
assert archived_histories[1]["name"] == "History 0"
archived_histories = self.dataset_populator.get_archived_histories(query="order=name-asc")
assert len(archived_histories) == num_histories
assert archived_histories[0]["name"] == "History 0"
assert archived_histories[1]["name"] == "History 1"
[docs] def test_archiving_an_archived_history_conflicts(self):
history_id = self.dataset_populator.new_history()
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 200)
assert archive_response.json()["archived"] is True
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 409)
[docs] def test_archived_histories_are_not_listed_by_default(self):
history_id = self.dataset_populator.new_history()
archive_response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is(archive_response, 200)
assert archive_response.json()["archived"] is True
histories = self.dataset_populator.get_histories()
for history in histories:
assert history["id"] != history_id