Warning
This document is for an in-development version 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
from urllib.parse import urljoin
from uuid import uuid4
import pytest
from requests import (
get,
post,
put,
Session,
)
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.api_mixins import (
BaseHistories,
ImportExportTests,
)
from galaxy_test.base.decorators import (
requires_admin,
requires_new_library,
requires_new_user,
)
from galaxy_test.base.populators import (
DatasetCollectionPopulator,
DatasetPopulator,
LibraryPopulator,
)
from ._framework import ApiTestCase
[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
[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_respects_view(self):
history_id = self._create_history(f"TestHistoryForShowView_{uuid4()}")["id"]
# By default the view is "detailed"
show_response = self._get(f"histories/{history_id}").json()
assert "state" in show_response
# Change the view to summary
show_response = self._get(f"histories/{history_id}", {"view": "summary"}).json()
assert "state" not in show_response
# Expect only specific keys
expected_keys = ["name"]
unexpected_keys = ["id", "deleted", "state"]
show_response = self._get(f"histories/{history_id}", {"keys": ",".join(expected_keys)}).json()
assert len(show_response) == len(expected_keys)
for key in expected_keys:
assert key in show_response
for key in unexpected_keys:
assert key not in show_response
[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_show_most_recently_used_without_session(self):
# A request with neither an API key nor a galaxysession cookie has no user and no current
# history, so there is no most recently used history to serialize.
# Regression test for https://github.com/galaxyproject/galaxy/issues/23439.
response = get(urljoin(self.url, "api/histories/most_recently_used"))
self._assert_status_code_is(response, 404)
[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_views(self):
# Make sure there is at least one history
self._create_history(f"TestHistoryForViews_{uuid4()}")["id"]
# By default the view is summary
index_response = self._get("histories").json()
for history in index_response:
assert "state" not in history
# Change the view to detailed
index_response = self._get("histories?view=detailed").json()
for history in index_response:
assert "state" in history
# Expect only specific keys
expected_keys = ["nice_size", "contents_active", "contents_states"]
unexpected_keys = ["id", "deleted", "state"]
index_response = self._get(f"histories?keys={','.join(expected_keys)}").json()
for history in index_response:
assert len(history) == len(expected_keys)
for key in expected_keys:
assert key in history
for key in unexpected_keys:
assert key not in history
# Expect combination of view and keys
view = "summary"
expected_keys = ["create_time", "count"]
data = dict(view=view, keys=",".join(expected_keys))
index_response = self._get("histories", data=data).json()
for history in index_response:
for key in expected_keys:
assert key in history
self._assert_has_keys(history, "id", "name", "url", "update_time", "deleted", "purged", "tags")
[docs]
def test_index_search_mode_views(self):
# Make sure there is at least one history
expected_name_contains = "SearchMode"
self._create_history(f"TestHistory{expected_name_contains}_{uuid4()}")["id"]
# By default the view is summary
data = dict(search=expected_name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
for history in index_response:
assert "state" not in history
# Change the view to detailed
data = dict(search=expected_name_contains, show_published=False)
index_response = self._get("histories?view=detailed", data=data).json()
for history in index_response:
assert "state" in history
# Expect only specific keys
expected_keys = ["nice_size", "contents_active", "contents_states"]
unexpected_keys = ["id", "deleted", "state"]
data = dict(search=expected_name_contains, show_published=False, keys=",".join(expected_keys))
index_response = self._get("histories", data=data).json()
for history in index_response:
assert len(history) == len(expected_keys)
for key in expected_keys:
assert key in history
for key in unexpected_keys:
assert key not in history
# Expect combination of view and keys
view = "summary"
expected_keys = ["create_time", "count"]
data = dict(search=expected_name_contains, show_published=False, view=view, keys=",".join(expected_keys))
index_response = self._get("histories", data=data).json()
for history in index_response:
for key in expected_keys:
assert key in history
self._assert_has_keys(history, "id", "name", "url", "update_time", "deleted", "purged", "tags")
[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_index_sort_by_name_is_case_insensitive(self):
# Codepoint ordering would sort every capitalised name ahead of every lowercase one.
with self._different_user(f"user_{uuid4()}@bx.psu.edu"):
unique_id = uuid4().hex
for name in [f"Zeta_{unique_id}", f"alpha_{unique_id}", f"Beta_{unique_id}"]:
self._create_history(name)
data = dict(search=unique_id, show_published=False, sort_by="name", sort_desc=False)
names = [history["name"] for history in self._get("histories", data=data).json()]
assert names == [f"alpha_{unique_id}", f"Beta_{unique_id}", f"Zeta_{unique_id}"]
data["sort_desc"] = True
names = [history["name"] for history in self._get("histories", data=data).json()]
assert names == [f"Zeta_{unique_id}", f"Beta_{unique_id}", f"alpha_{unique_id}"]
[docs]
def test_index_advanced_filter(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())
history_0 = self._create_history(expected_history_name.lower())["id"]
history_1 = self._create_history(f"Another history_{uuid4()}")["id"]
self._delete(f"histories/{history_1}")
name_contains = "history"
data = dict(search=name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 3
name_contains = "history that match query"
data = dict(search=name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 3
name_contains = "ANOTHER"
data = dict(search=name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 0
data = dict(search="is:deleted", show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 1
self._update(history_0, {"published": True})
data = dict(search=f"query_{unique_id} is:published")
index_response = self._get("histories", data=data).json()
assert len(index_response) == 1
archived_history_id = self._create_history(f"Archived history_{uuid4()}")["id"]
name_contains = "history"
data = dict(search=name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 4
# Archived histories should not be included by default
self.dataset_populator.archive_history(archived_history_id)
data = dict(search=name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 3
self._create_history_then_publish_and_archive_it(f"Public Archived history_{uuid4()}")
data = dict(search=name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 3
name_contains = "Archived"
data = dict(search=name_contains, show_published=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 0
# Archived public histories should be included when filtering by show_published and show_archived
data = dict(search="is:published", show_archived=True)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 2
# Searching all published histories will NOT include the archived if show_archived is not set
data = dict(search="is:published")
index_response = self._get("histories", data=data).json()
assert len(index_response) == 1
# Searching all published histories will include our own archived when show_own is false
# as long as they are published
data = dict(search="is:published", show_own=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 2
# Publish a history and archive it by a different user
with self._different_user(f"other_user_{uuid4()}@bx.psu.edu"):
self._create_history_then_publish_and_archive_it(f"Public Archived history_{uuid4()}")
# Searching all published histories will include archived from other users and our own
# as long as they are published
data = dict(search="is:published", show_own=False)
index_response = self._get("histories", data=data).json()
assert len(index_response) == 3
def _create_history_then_publish_and_archive_it(self, name):
history_id = self._create_history(name)["id"]
response = self._update(history_id, {"published": True})
self._assert_status_code_is_ok(response)
response = self.dataset_populator.archive_history(history_id)
self._assert_status_code_is_ok(response)
return history_id
[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())
history = self._show(history_id)
assert "update_time" in history
original_update_time = history["update_time"]
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
history = self._show(history_id)
new_update_time = history["update_time"]
assert original_update_time == new_update_time
[docs]
def test_copy_history_does_not_duplicate_tags(self):
history_id = self.dataset_populator.new_history()
# Create a standalone dataset and tag it
new_hda = self.dataset_populator.new_dataset(history_id, content="tagged dataset")
hda_id = new_hda["id"]
self.dataset_populator.tag_dataset(history_id, hda_id, tags=["hda_tag"])
# Create a collection and tag it
fetch_response = self.dataset_collection_populator.create_list_in_history(
history_id, contents=["Hello", "World"], direct_upload=True
)
collection = self.dataset_collection_populator.wait_for_fetched_collection(fetch_response.json())
hdca_id = collection["id"]
self._put(
f"histories/{history_id}/contents/dataset_collections/{hdca_id}",
data={"tags": ["hdca_tag"]},
json=True,
).raise_for_status()
# Also tag a dataset within the collection
element_hda_id = collection["elements"][0]["object"]["id"]
self.dataset_populator.tag_dataset(history_id, element_hda_id, tags=["element_tag"])
# Copy the history
copied_history_response = self.dataset_populator.copy_history(history_id)
copied_history_response.raise_for_status()
copied_history = copied_history_response.json()
copied_history_id = copied_history["id"]
# Verify standalone HDA tags are not duplicated
copied_contents = self._get(f"histories/{copied_history_id}/contents").json()
copied_hdas = [c for c in copied_contents if c["history_content_type"] == "dataset" and c["visible"]]
assert len(copied_hdas) == 1
copied_hda_details = self.dataset_populator.get_history_dataset_details(
history_id=copied_history_id, dataset_id=copied_hdas[0]["id"]
)
assert copied_hda_details["tags"] == ["hda_tag"], f"Expected ['hda_tag'] but got {copied_hda_details['tags']}"
# Verify HDCA tags are not duplicated
copied_collection = self.dataset_populator.get_history_collection_details(
history_id=copied_history_id, history_content_type="dataset_collection"
)
assert copied_collection["tags"] == ["hdca_tag"], f"Expected ['hdca_tag'] but got {copied_collection['tags']}"
[docs]
def test_copy_datasets_to_history_does_not_duplicate_tags(self):
source_history_id = self.dataset_populator.new_history()
target_history_id = self.dataset_populator.new_history()
# Create a tagged HDA
new_hda = self.dataset_populator.new_dataset(source_history_id, content="tagged dataset")
hda_id = new_hda["id"]
self.dataset_populator.tag_dataset(source_history_id, hda_id, tags=["hda_tag"])
# Create a tagged HDCA
fetch_response = self.dataset_collection_populator.create_list_in_history(
source_history_id, contents=["Hello", "World"], direct_upload=True
)
collection = self.dataset_collection_populator.wait_for_fetched_collection(fetch_response.json())
hdca_id = collection["id"]
self._put(
f"histories/{source_history_id}/contents/dataset_collections/{hdca_id}",
data={"tags": ["hdca_tag"]},
json=True,
).raise_for_status()
# Copy both to target history via copy_contents endpoint
payload = {
"source_content": [
{"id": hda_id, "type": "dataset"},
{"id": hdca_id, "type": "dataset_collection"},
],
"target_history_ids": [target_history_id],
}
self._post(
f"histories/{source_history_id}/copy_contents",
data=payload,
json=True,
).raise_for_status()
# Verify copied HDA tags are not duplicated
target_contents = self._get(f"histories/{target_history_id}/contents").json()
copied_hdas = [c for c in target_contents if c["history_content_type"] == "dataset" and c["visible"]]
assert len(copied_hdas) == 1
copied_hda_details = self.dataset_populator.get_history_dataset_details(
history_id=target_history_id, dataset_id=copied_hdas[0]["id"]
)
assert copied_hda_details["tags"] == ["hda_tag"], f"Expected ['hda_tag'] but got {copied_hda_details['tags']}"
# Verify copied HDCA tags are not duplicated
copied_hdcas = [c for c in target_contents if c["history_content_type"] == "dataset_collection"]
assert len(copied_hdcas) == 1
copied_collection = self.dataset_populator.get_history_collection_details(
history_id=target_history_id, history_content_type="dataset_collection"
)
assert copied_collection["tags"] == ["hdca_tag"], f"Expected ['hdca_tag'] but got {copied_collection['tags']}"
# 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_anonymous_without_session_cannot_copy_published(self):
history_id = self.dataset_populator.new_history(name=f"for_copying_without_session_{uuid4()}")
self.dataset_populator.make_public(history_id)
# A request with neither an API key nor a galaxysession cookie has no session to own the copy,
# so copying a published history is rejected rather than doing the copy and orphaning it.
# Regression test for https://github.com/galaxyproject/galaxy/issues/23148.
copy_response = post(
urljoin(self.url, "api/histories"),
json={"history_id": history_id, "name": f"copied_without_session_{uuid4()}"},
)
self._assert_status_code_is(copy_response, 403)
[docs]
def test_publish_non_alphanumeric(self):
history_name = "تاریخچه"
history_id = self.dataset_populator.new_history(name=history_name)
response = self.dataset_populator.make_public(history_id)
assert history_name in response["username_and_slug"]
[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.dataset_populator.purge_history(history_id)
# 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.dataset_populator.purge_history(history_id)
# 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.dataset_populator.purge_history(history_id)
# 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 TestImportExportHistory(ApiTestCase, ImportExportTests):
task_based = False
[docs]
def test_history_exports_job_id_resolves(self):
history_id = self.dataset_populator.new_history()
self.dataset_populator.new_dataset(history_id, content="1 2 3", wait=True)
self.dataset_populator.prepare_export(history_id, {})
exports_response = self._get(f"histories/{history_id}/exports")
exports_response.raise_for_status()
exports = exports_response.json()
assert exports, "Expected at least one export record"
job_id = exports[0]["job_id"]
job_response = self._get(f"jobs/{job_id}")
job_response.raise_for_status()
assert job_response.json()["id"] == job_id
[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, and its datasets must be accessible to the users it is shared with
populator = DatasetPopulator(self.galaxy_interactor)
hda = populator.new_dataset(history_id, wait=True)
populator.make_dataset_public(history_id, hda["id"])
return history_id
[docs]
def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.library_populator = LibraryPopulator(self.galaxy_interactor)
def _shared_with_me_ids(self) -> list[str]:
response = self._get("histories/shared_with_me")
self._assert_status_code_is(response, 200)
return [h["id"] for h in response.json()]
[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, wait=True)
hda_id = hda["id"]
self.dataset_populator.make_dataset_public(history_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()
# Let the upload job finish first; it still updates the dataset while it runs.
hda = self.dataset_populator.new_dataset(history_id, wait=True)
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
[docs]
@requires_new_library
@requires_new_user
def test_make_private_imported_history(self):
source_history_id = self.dataset_populator.new_history()
source_hdas = [
self.dataset_populator.new_dataset(source_history_id, content=content, wait=True)
for content in ("source dataset one", "source dataset two")
]
library_dataset = self.library_populator.new_library_dataset("source library dataset")
library_copy_response = self._post(
f"histories/{source_history_id}/contents",
data={"content": library_dataset["id"], "source": "library", "type": "dataset"},
json=True,
)
library_copy_response.raise_for_status()
source_library_hda = library_copy_response.json()
self.dataset_populator.make_dataset_public_raw(source_history_id, source_library_hda["id"]).raise_for_status()
self.dataset_populator.make_public(source_history_id)
with self._different_user():
importer_email = self._get("users/current").json()["email"]
copied_history_response = self.dataset_populator.copy_history(source_history_id)
copied_history_response.raise_for_status()
copied_history_id = copied_history_response.json()["id"]
imported_hdas = self.dataset_populator.get_history_contents(copied_history_id)
importer_hda = self.dataset_populator.new_dataset(copied_history_id, content="importer dataset", wait=True)
make_private_response = self._make_history_private_as_browser(importer_email, copied_history_id)
# the two imported datasets belong to the source user, the library dataset is not counted
assert make_private_response["skipped_datasets"] == len(source_hdas)
for imported_hda, source_hda in zip(imported_hdas, source_hdas):
imported_details = self._get(f"datasets/{imported_hda['id']}")
self._assert_status_code_is(imported_details, 200)
assert imported_details.json()["dataset_id"] == source_hda["dataset_id"]
self._assert_status_code_is(self._get(f"datasets/{importer_hda['id']}"), 200)
with self._different_user("unrelated-user@test.galaxyproject.org"):
for hda in imported_hdas:
self._assert_status_code_is(self._get(f"datasets/{hda['id']}"), 200)
self._assert_status_code_is(self._get(f"datasets/{importer_hda['id']}"), 403)
source_history = self._show(source_history_id)
assert source_history["published"] is True
with self._different_user("unrelated-user@test.galaxyproject.org"):
for hda in [*source_hdas, source_library_hda]:
self._assert_status_code_is(self._get(f"datasets/{hda['id']}"), 200)
def _make_history_private_as_browser(self, email: str, history_id: str) -> dict:
# history/make_private is a legacy controller route outside /api, so it
# only accepts a browser session, not an API key.
with Session() as session:
self._login_browser_session(session, email, "testpass")
make_private_response = session.post(
urljoin(self.url, "history/make_private"), data={"history_id": history_id}
)
self._assert_status_code_is(make_private_response, 200)
return make_private_response.json()
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_history("archived_histories_index_user@bx.psu.edu") as history_id:
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
[docs]
class TestHistoryGraphApi(ApiTestCase, BaseHistories):
"""API-level tests for ``GET /api/histories/{id}/graph``.
These cover the surface the endpoint owns: status codes, query
parameter validation, auth boundaries, and response shape. Builder
logic is exercised separately in ``test_HistoryGraphBuilder``.
"""
[docs]
def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
# ── response shape ──
[docs]
def test_empty_history_returns_empty_graph(self):
history_id = self.dataset_populator.new_history()
body = self.dataset_populator.get_history_graph(history_id)
self._assert_has_keys(body, "nodes", "edges", "truncated")
assert body["nodes"] == []
assert body["edges"] == []
assert body["truncated"]["item_count_capped"] is False
assert body["truncated"]["scope_type"] == "recent"
[docs]
def test_standalone_datasets_appear_as_dataset_nodes(self):
history_id = self.dataset_populator.new_history()
self.dataset_populator.new_dataset(history_id, content="a", wait=True)
self.dataset_populator.new_dataset(history_id, content="b", wait=True)
body = self.dataset_populator.get_history_graph(history_id)
assert len(body["nodes"]) == 2
assert body["edges"] == []
assert all(n["src"] == "hda" for n in body["nodes"])
[docs]
def test_limit_caps_items_and_sets_truncation_flag(self):
history_id = self.dataset_populator.new_history()
for i in range(5):
self.dataset_populator.new_dataset(history_id, content=f"row {i}", wait=True)
body = self.dataset_populator.get_history_graph(history_id, limit=3)
assert len(body["nodes"]) == 3
assert body["truncated"]["item_count_capped"] is True
[docs]
def test_seed_scope_returns_seed_centered_window(self):
history_id = self.dataset_populator.new_history()
dataset = self.dataset_populator.new_dataset(history_id, content="seed", wait=True)
body = self.dataset_populator.get_history_graph(
history_id, seed_scope_src="hda", seed_scope_id=dataset["id"], limit=5
)
assert body["truncated"]["scope_type"] == "seed_centered"
assert ("hda", dataset["id"]) in {(n["src"], n["id"]) for n in body["nodes"]}
# ── query-parameter validation (API-layer regex and bounds) ──
[docs]
@pytest.mark.parametrize(
"param,value",
[
("seed_src", "bogus"), # not a valid NodeSrc
("seed_scope_src", "tool_request"), # not allowed as a scope center
("limit", 5000), # above max
("depth", 21), # above max
],
)
def test_invalid_query_params_return_400(self, param, value):
history_id = self.dataset_populator.new_history()
response = self.dataset_populator.get_history_graph_raw(history_id, **{param: value})
self._assert_status_code_is(response, 400)
[docs]
def test_seed_src_without_seed_id_is_rejected(self):
history_id = self.dataset_populator.new_history()
response = self.dataset_populator.get_history_graph_raw(history_id, seed_src="hda")
self._assert_status_code_is(response, 400)
# ── manager-level validation (after API regex passes) ──
[docs]
def test_seed_scope_not_in_target_history_is_rejected(self):
source_history = self.dataset_populator.new_history()
dataset = self.dataset_populator.new_dataset(source_history, content="a", wait=True)
target_history = self.dataset_populator.new_history()
response = self.dataset_populator.get_history_graph_raw(
target_history, seed_scope_src="hda", seed_scope_id=dataset["id"]
)
self._assert_status_code_is(response, 404)
# ── auth ──
[docs]
def test_other_users_history_is_forbidden(self):
with self._different_user():
other_history_id = self.dataset_populator.new_history()
response = self.dataset_populator.get_history_graph_raw(other_history_id)
self._assert_status_code_is(response, 403)
[docs]
def test_nonexistent_history_is_rejected(self):
response = self.dataset_populator.get_history_graph_raw("0000000000000000")
self._assert_status_code_is(response, 400)