Fix dataset collection tag duplication when copying a history

HistoryDatasetCollectionAssociation.copy() was internally calling
copy_tags_from(), but both callers (History.copy() and
DatasetCollectionManager.copy()) also called copy_tags_from() on the
returned HDCA, resulting in every tag being duplicated.

Move the tag-copying responsibility entirely into HDCA.copy() via a new
target_user parameter (falling back to source history user), and remove
the redundant copy_tags_from() calls from both callers. This is safer
than the reverse approach since any future caller of HDCA.copy() will
get correct tag copying automatically.

Fixes https://github.com/galaxyproject/galaxy/issues/21872
This commit is contained in:
mvdbeek
2026-03-04 17:07:13 +01:00
parent 0281dd7c71
commit 8f4577eebe
3 changed files with 52 additions and 5 deletions
+1 -1
View File
@@ -508,8 +508,8 @@ class DatasetCollectionManager:
flush=False,
element_destination=element_destination,
dataset_instance_attributes=dataset_instance_attributes,
target_user=trans.get_user(),
)
new_hdca.copy_tags_from(target_user=trans.get_user(), source=source_hdca)
if not copy_elements:
parent.add_dataset_collection(new_hdca)
trans.sa_session.commit()
+11 -4
View File
@@ -3747,13 +3747,18 @@ class History(Base, HasTags, Dictifiable, UsesAnnotations, HasName, Serializable
else:
hdcas = self.active_dataset_collections
for hdca in hdcas:
new_hdca = hdca.copy(flush=False, element_destination=new_history, set_hid=False, minimize_copies=True)
new_hdca = hdca.copy(
flush=False,
element_destination=new_history,
set_hid=False,
minimize_copies=True,
target_user=target_user,
)
new_history.add_dataset_collection(new_hdca, set_hid=False)
db_session.add(new_hdca)
if target_user:
new_hdca.copy_item_annotation(db_session, self.user, hdca, target_user, new_hdca)
new_hdca.copy_tags_from(target_user, hdca)
new_history.hid_counter = self.hid_counter
db_session.commit()
@@ -7850,6 +7855,7 @@ class HistoryDatasetCollectionAssociation(
flush: bool = True,
set_hid: bool = True,
minimize_copies: bool = False,
target_user: Optional[User] = None,
):
"""
Create a copy of this history dataset collection association. Copy
@@ -7878,8 +7884,9 @@ class HistoryDatasetCollectionAssociation(
hdca.collection = collection_copy
session = required_object_session(self)
session.add(hdca)
if self.history and self.history.user:
hdca.copy_tags_from(self.history.user, self)
copy_user = target_user or (self.history.user if self.history else None)
if copy_user:
hdca.copy_tags_from(copy_user, self)
if element_destination and set_hid:
element_destination.stage_addition(hdca)
element_destination.add_pending_items()
+40
View File
@@ -522,6 +522,46 @@ class TestHistoriesApi(ApiTestCase, BaseHistories):
new_update_time = history["update_time"]
assert original_update_time == new_update_time
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/{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']}"
# TODO: (CE) test_create_from_copy
def test_import_from_model_store_dict(self):
response = self.dataset_populator.create_from_store(store_dict=history_model_store_dict())