From 58e928f02a476986182801225aa1445a93648a1e Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Tue, 17 Mar 2026 12:53:23 +0100 Subject: [PATCH 1/3] Fix duplicate collection elements in metadata outputs_populated When re-populating dynamic output collections, existing elements were not cleared before adding new ones, causing duplicate entries to accumulate in metadata/outputs_populated/collections_attrs.txt and then propagate to the database. Two fixes: - In collect_dynamic_outputs: clear elements and reset element_count before re-populating a dynamic collection - In _import_collection_instances: delete existing DB elements before materializing new ones from import attrs (allow_edit path) --- lib/galaxy/job_execution/output_collect.py | 3 +++ lib/galaxy/model/store/__init__.py | 5 +++++ 2 files changed, 8 insertions(+) diff --git a/lib/galaxy/job_execution/output_collect.py b/lib/galaxy/job_execution/output_collect.py index bea31b54daf..703cdaae339 100644 --- a/lib/galaxy/job_execution/output_collect.py +++ b/lib/galaxy/job_execution/output_collect.py @@ -189,6 +189,9 @@ def collect_dynamic_outputs( # We are adding dynamic collections, which may be precreated, but their actually state is still new! collection.populated_state = collection.populated_states.NEW + # Clear any existing elements to avoid duplicates when re-populating + collection.elements.clear() + collection.element_count = None try: collection_builder = builder.BoundCollectionBuilder(collection) diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py index 65d9ae780d3..22d314d4745 100644 --- a/lib/galaxy/model/store/__init__.py +++ b/lib/galaxy/model/store/__init__.py @@ -899,6 +899,11 @@ class ModelImportStore(metaclass=abc.ABCMeta): for attribute in attributes: if attribute in collection_attrs: setattr(dc, attribute, collection_attrs.get(attribute)) + # Clear existing elements to avoid duplicates when re-importing + if "elements" in collection_attrs: + for element in list(dc.elements): + self.sa_session.delete(element) + dc.elements.clear() materialize_elements(dc) else: # create collection From e51262fd00324f93ae525a8e3d5e063d25ad66b1 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Tue, 17 Mar 2026 12:53:24 +0100 Subject: [PATCH 2/3] Fix mapped dynamic collection outputs missing with extended metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Commit 4172b9b5 ("Fix AttributeError serializing implicit output collections during job prep") excluded all output_dataset_collections from io_dicts when exclude_implicit_outputs=True. This broke mapped dynamic collection outputs because their per-job DCs were no longer serialized to outputs_new, so set_metadata.py could not discover and populate them. Selectively include output_dataset_collections where the name is not in out_data — this excludes shared DCs for mapped dataset outputs (which have N precreated elements with uninitialized sentinels) while including per-job DCs for mapped collection outputs. Add integration tests verifying no duplicate collection elements in both non-mapped and mapped dynamic collection outputs with extended metadata. --- lib/galaxy/model/__init__.py | 17 +- ...st_extended_metadata_duplicate_elements.py | 210 ++++++++++++++++++ test/unit/data/model/test_model.py | 28 ++- 3 files changed, 246 insertions(+), 9 deletions(-) create mode 100644 test/integration/test_extended_metadata_duplicate_elements.py diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 5f506daee13..3d53d6a401a 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -1839,8 +1839,21 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable): if obj.name not in out_data: out_collections[obj.name] = obj.dataset_collection_instance # else this is a mapped over output - if not exclude_implicit_outputs: - out_collections.update([(obj.name, obj.dataset_collection) for obj in self.output_dataset_collections]) + if exclude_implicit_outputs: + # Include implicit output dataset collections only when they represent + # a tool's collection output (name not in out_data). Exclude shared DCs + # for mapped dataset outputs (name in out_data) which have N precreated + # elements where only the current job's element is initialized. + for implicit_obj in self.output_dataset_collections: + if implicit_obj.name not in out_data: + out_collections[implicit_obj.name] = implicit_obj.dataset_collection + else: + out_collections.update( + [ + (implicit_obj.name, implicit_obj.dataset_collection) + for implicit_obj in self.output_dataset_collections + ] + ) return IoDicts(inp_data, out_data, out_collections) # TODO: Add accessors for members defined in SQL Alchemy for the Job table and diff --git a/test/integration/test_extended_metadata_duplicate_elements.py b/test/integration/test_extended_metadata_duplicate_elements.py new file mode 100644 index 00000000000..82ad0981bc4 --- /dev/null +++ b/test/integration/test_extended_metadata_duplicate_elements.py @@ -0,0 +1,210 @@ +"""Integration test verifying that extended metadata does not create duplicate collection elements.""" + +from sqlalchemy import ( + func, + select, +) + +from galaxy.model import ( + DatasetCollectionElement, + HistoryDatasetCollectionAssociation, +) +from galaxy_test.base.populators import ( + DatasetCollectionPopulator, + DatasetPopulator, +) +from galaxy_test.driver.integration_util import IntegrationTestCase + + +class TestExtendedMetadataDuplicateElements(IntegrationTestCase): + dataset_populator: DatasetPopulator + framework_tool_and_types = True + + @classmethod + def handle_galaxy_config_kwds(cls, config): + super().handle_galaxy_config_kwds(config) + config["metadata_strategy"] = "extended" + config["retry_metadata_internally"] = False + + def setUp(self): + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor) + + def test_no_duplicate_elements_in_dynamic_list_output(self, history_id): + """Run a tool with dynamic collection output and verify no duplicate elements.""" + response = self.dataset_populator.run_tool( + "collection_creates_dynamic_list_of_pairs", + {"foo": "bar"}, + history_id, + ) + job_api_id = response["jobs"][0]["id"] + self.dataset_populator.wait_for_job(job_api_id, assert_ok=True) + + output_collections = response["output_collections"] + hdca_details = self.dataset_populator.get_history_collection_details( + history_id, content_id=output_collections[0]["id"] + ) + + sa_session = self._app.model.session + hdca_id = self._app.security.decode_id(hdca_details["id"]) + hdca = sa_session.get(HistoryDatasetCollectionAssociation, hdca_id) + assert hdca is not None + dc = hdca.collection + + # Verify exactly 3 outer elements (samp1, samp2, samp3) - no duplicates + outer_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == dc.id) + ) + assert outer_count == 3, f"Expected 3 outer elements but found {outer_count}. Duplicate elements detected!" + + # Verify each inner pair has exactly 2 elements (forward, reverse) + for element in dc.elements: + assert element.child_collection is not None + inner_count = sa_session.scalar( + select(func.count()).where( + DatasetCollectionElement.dataset_collection_id == element.child_collection.id + ) + ) + assert inner_count == 2, ( + f"Expected 2 inner elements for '{element.element_identifier}' " + f"but found {inner_count}. Duplicate elements detected!" + ) + + def test_no_duplicate_elements_in_mapped_dynamic_collection(self, history_id): + """Map a tool with dynamic collection output over a list and verify no duplicate elements.""" + # Create a list of 2 tabular datasets + fetch_response = self.dataset_collection_populator.create_list_in_history( + history_id, + contents=["101\t1\n101\t2\n105\t3\n", "201\t10\n201\t20\n205\t30\n"], + ext="tabular", + wait=True, + ) + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + hdca_id = fetch_response.json()["output_collections"][0]["id"] + + # Map collection_split_on_column over the list + inputs = { + "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}, + } + response = self.dataset_populator.run_tool( + "collection_split_on_column", + inputs, + history_id, + ) + + # Wait for all mapping jobs to complete + for job in response["jobs"]: + self.dataset_populator.wait_for_job(job["id"], assert_ok=True) + + # The implicit output should be a list:list collection + implicit_collections = response["implicit_collections"] + assert len(implicit_collections) == 1 + implicit_hdca_details = self.dataset_populator.get_history_collection_details( + history_id, content_id=implicit_collections[0]["id"] + ) + + sa_session = self._app.model.session + implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"]) + implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id) + assert implicit_hdca is not None + outer_dc = implicit_hdca.collection + + # The outer collection should have exactly 2 elements (one per input) + outer_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id) + ) + assert ( + outer_count == 2 + ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!" + + # Each inner collection should have no duplicate elements + for element in outer_dc.elements: + inner_dc = element.child_collection + assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None" + inner_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ) + # Each input has 2 unique first-column values, so 2 split files + assert ( + inner_count is not None and inner_count > 0 + ), f"Inner collection for '{element.element_identifier}' has no elements" + # Check for duplicates: element_identifiers should be unique + inner_elements = sa_session.scalars( + select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ).all() + identifiers = [e.element_identifier for e in inner_elements] + assert len(identifiers) == len(set(identifiers)), ( + f"Duplicate element identifiers found in inner collection " + f"for '{element.element_identifier}': {identifiers}" + ) + + def test_no_duplicate_elements_in_mapped_static_collection(self, history_id): + """Map a tool with a static collection output over a list and verify no duplicate elements. + + This covers the case where a tool declares a static collection output + (e.g. paired with forward/reverse) and is mapped over a list input, + producing a list:paired output. The per-job DC contains fully + initialized elements and must be included in io_dicts for metadata + serialization. + """ + hdca = self.dataset_collection_populator.create_list_in_history( + history_id, + contents=["line1\nline2\nline3\nline4\n", "lineA\nlineB\nlineC\nlineD\n"], + ext="txt", + wait=True, + ) + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + hdca_id = hdca.json()["output_collections"][0]["id"] + + inputs = { + "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}, + } + response = self.dataset_populator.run_tool( + "collection_creates_pair", + inputs, + history_id, + ) + + for job in response["jobs"]: + self.dataset_populator.wait_for_job(job["id"], assert_ok=True) + + implicit_collections = response["implicit_collections"] + assert len(implicit_collections) == 1 + implicit_hdca_details = self.dataset_populator.get_history_collection_details( + history_id, content_id=implicit_collections[0]["id"] + ) + + sa_session = self._app.model.session + implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"]) + implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id) + assert implicit_hdca is not None + outer_dc = implicit_hdca.collection + + # The outer collection should have exactly 2 elements (one per input) + outer_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id) + ) + assert ( + outer_count == 2 + ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!" + + # Each inner pair should have exactly 2 elements (forward, reverse) with no duplicates + for element in outer_dc.elements: + inner_dc = element.child_collection + assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None" + inner_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ) + assert inner_count == 2, ( + f"Expected 2 inner elements for '{element.element_identifier}' " + f"but found {inner_count}. Duplicate elements detected!" + ) + inner_elements = sa_session.scalars( + select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ).all() + identifiers = [e.element_identifier for e in inner_elements] + assert set(identifiers) == { + "forward", + "reverse", + }, f"Expected forward/reverse but got {identifiers} for '{element.element_identifier}'" diff --git a/test/unit/data/model/test_model.py b/test/unit/data/model/test_model.py index 4ec6c459af8..36ebbbffa56 100644 --- a/test/unit/data/model/test_model.py +++ b/test/unit/data/model/test_model.py @@ -23,22 +23,36 @@ def test_permitted_actions(): def test_io_dicts_excludes_implicit_output_collections(): """Regression test for https://github.com/galaxyproject/galaxy/issues/22015 - When a tool with an explicit collection output is mapped over a list, - each job gets a JobToImplicitOutputDatasetCollectionAssociation pointing - to a DatasetCollection with precreated (unpopulated) elements. - io_dicts(exclude_implicit_outputs=True) must exclude these to avoid - crashes during metadata serialization on unpopulated elements. + When a tool with a dataset output is mapped over a list, each job gets + both a JobToOutputDatasetAssociation and a + JobToImplicitOutputDatasetCollectionAssociation with the same name. + The implicit DC has precreated (unpopulated) elements; only the current + job's element is initialized. io_dicts(exclude_implicit_outputs=True) + must exclude these shared DCs (name in out_data) to avoid crashes + during metadata serialization, but must include implicit DCs for + collection outputs (name not in out_data) so set_metadata.py can + discover and populate them. """ job = model.Job() dc = model.DatasetCollection(collection_type="paired") assoc = model.JobToImplicitOutputDatasetCollectionAssociation(name="paired_output", dataset_collection=dc) job.output_dataset_collections.append(assoc) - # With exclude_implicit_outputs=True, output_dataset_collections must be excluded + # When the name is NOT in out_data (collection output), the implicit DC + # should be included even with exclude_implicit_outputs=True + io = job.io_dicts(exclude_implicit_outputs=True) + assert "paired_output" in io.out_collections + assert io.out_collections["paired_output"] is dc + + # Now simulate a mapped dataset output: same name in both out_data and + # output_dataset_collections. The shared DC must be excluded. + hda = model.HistoryDatasetAssociation() + out_assoc = model.JobToOutputDatasetAssociation(name="paired_output", dataset=hda) + job.output_datasets.append(out_assoc) + io = job.io_dicts(exclude_implicit_outputs=True) assert "paired_output" not in io.out_collections # With exclude_implicit_outputs=False (default), they should be included io = job.io_dicts(exclude_implicit_outputs=False) assert "paired_output" in io.out_collections - assert io.out_collections["paired_output"] is dc From 9e2d5f5a94aeb95d4e85f238bc75712bb0f25e57 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Tue, 17 Mar 2026 16:36:04 +0100 Subject: [PATCH 3/3] Move tests into TestExtendedMetadataIntegration --- test/integration/test_extended_metadata.py | 189 ++++++++++++++++ ...st_extended_metadata_duplicate_elements.py | 210 ------------------ 2 files changed, 189 insertions(+), 210 deletions(-) delete mode 100644 test/integration/test_extended_metadata_duplicate_elements.py diff --git a/test/integration/test_extended_metadata.py b/test/integration/test_extended_metadata.py index b6a5d7ae192..91896daa18f 100644 --- a/test/integration/test_extended_metadata.py +++ b/test/integration/test_extended_metadata.py @@ -1,6 +1,16 @@ """Integration tests for the Pulsar embedded runner.""" +from sqlalchemy import ( + func, + select, +) + +from galaxy.model import ( + DatasetCollectionElement, + HistoryDatasetCollectionAssociation, +) from galaxy_test.base.populators import ( + DatasetCollectionPopulator, DatasetPopulator, LibraryPopulator, ) @@ -56,6 +66,7 @@ class TestExtendedMetadataIntegration(integration_util.IntegrationTestCase): 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) @classmethod @@ -100,6 +111,184 @@ class TestExtendedMetadataIntegration(integration_util.IntegrationTestCase): assert dataset["file_ext"] == "bed", dataset assert dataset["created_from_basename"] == "4.bed" + def test_no_duplicate_elements_in_dynamic_list_output(self, history_id): + """Run a tool with dynamic collection output and verify no duplicate elements.""" + response = self.dataset_populator.run_tool( + "collection_creates_dynamic_list_of_pairs", + {"foo": "bar"}, + history_id, + ) + job_api_id = response["jobs"][0]["id"] + self.dataset_populator.wait_for_job(job_api_id, assert_ok=True) + + output_collections = response["output_collections"] + hdca_details = self.dataset_populator.get_history_collection_details( + history_id, content_id=output_collections[0]["id"] + ) + + sa_session = self._app.model.session + hdca_id = self._app.security.decode_id(hdca_details["id"]) + hdca = sa_session.get(HistoryDatasetCollectionAssociation, hdca_id) + assert hdca is not None + dc = hdca.collection + + # Verify exactly 3 outer elements (samp1, samp2, samp3) - no duplicates + outer_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == dc.id) + ) + assert outer_count == 3, f"Expected 3 outer elements but found {outer_count}. Duplicate elements detected!" + + # Verify each inner pair has exactly 2 elements (forward, reverse) + for element in dc.elements: + assert element.child_collection is not None + inner_count = sa_session.scalar( + select(func.count()).where( + DatasetCollectionElement.dataset_collection_id == element.child_collection.id + ) + ) + assert inner_count == 2, ( + f"Expected 2 inner elements for '{element.element_identifier}' " + f"but found {inner_count}. Duplicate elements detected!" + ) + + def test_no_duplicate_elements_in_mapped_dynamic_collection(self, history_id): + """Map a tool with dynamic collection output over a list and verify no duplicate elements.""" + # Create a list of 2 tabular datasets + fetch_response = self.dataset_collection_populator.create_list_in_history( + history_id, + contents=["101\t1\n101\t2\n105\t3\n", "201\t10\n201\t20\n205\t30\n"], + ext="tabular", + wait=True, + ) + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + hdca_id = fetch_response.json()["output_collections"][0]["id"] + + # Map collection_split_on_column over the list + inputs = { + "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}, + } + response = self.dataset_populator.run_tool( + "collection_split_on_column", + inputs, + history_id, + ) + + # Wait for all mapping jobs to complete + for job in response["jobs"]: + self.dataset_populator.wait_for_job(job["id"], assert_ok=True) + + # The implicit output should be a list:list collection + implicit_collections = response["implicit_collections"] + assert len(implicit_collections) == 1 + implicit_hdca_details = self.dataset_populator.get_history_collection_details( + history_id, content_id=implicit_collections[0]["id"] + ) + + sa_session = self._app.model.session + implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"]) + implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id) + assert implicit_hdca is not None + outer_dc = implicit_hdca.collection + + # The outer collection should have exactly 2 elements (one per input) + outer_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id) + ) + assert ( + outer_count == 2 + ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!" + + # Each inner collection should have no duplicate elements + for element in outer_dc.elements: + inner_dc = element.child_collection + assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None" + inner_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ) + # Each input has 2 unique first-column values, so 2 split files + assert ( + inner_count is not None and inner_count > 0 + ), f"Inner collection for '{element.element_identifier}' has no elements" + # Check for duplicates: element_identifiers should be unique + inner_elements = sa_session.scalars( + select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ).all() + identifiers = [e.element_identifier for e in inner_elements] + assert len(identifiers) == len(set(identifiers)), ( + f"Duplicate element identifiers found in inner collection " + f"for '{element.element_identifier}': {identifiers}" + ) + + def test_no_duplicate_elements_in_mapped_static_collection(self, history_id): + """Map a tool with a static collection output over a list and verify no duplicate elements. + + This covers the case where a tool declares a static collection output + (e.g. paired with forward/reverse) and is mapped over a list input, + producing a list:paired output. The per-job DC contains fully + initialized elements and must be included in io_dicts for metadata + serialization. + """ + hdca = self.dataset_collection_populator.create_list_in_history( + history_id, + contents=["line1\nline2\nline3\nline4\n", "lineA\nlineB\nlineC\nlineD\n"], + ext="txt", + wait=True, + ) + self.dataset_populator.wait_for_history(history_id, assert_ok=True) + hdca_id = hdca.json()["output_collections"][0]["id"] + + inputs = { + "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}, + } + response = self.dataset_populator.run_tool( + "collection_creates_pair", + inputs, + history_id, + ) + + for job in response["jobs"]: + self.dataset_populator.wait_for_job(job["id"], assert_ok=True) + + implicit_collections = response["implicit_collections"] + assert len(implicit_collections) == 1 + implicit_hdca_details = self.dataset_populator.get_history_collection_details( + history_id, content_id=implicit_collections[0]["id"] + ) + + sa_session = self._app.model.session + implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"]) + implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id) + assert implicit_hdca is not None + outer_dc = implicit_hdca.collection + + # The outer collection should have exactly 2 elements (one per input) + outer_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id) + ) + assert ( + outer_count == 2 + ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!" + + # Each inner pair should have exactly 2 elements (forward, reverse) with no duplicates + for element in outer_dc.elements: + inner_dc = element.child_collection + assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None" + inner_count = sa_session.scalar( + select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ) + assert inner_count == 2, ( + f"Expected 2 inner elements for '{element.element_identifier}' " + f"but found {inner_count}. Duplicate elements detected!" + ) + inner_elements = sa_session.scalars( + select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) + ).all() + identifiers = [e.element_identifier for e in inner_elements] + assert set(identifiers) == { + "forward", + "reverse", + }, f"Expected forward/reverse but got {identifiers} for '{element.element_identifier}'" + def test_purge_while_job_running(self): # pass extra_sleep, since templating the command line will fail if the output # is deleted before remote_tool_eval runs. diff --git a/test/integration/test_extended_metadata_duplicate_elements.py b/test/integration/test_extended_metadata_duplicate_elements.py deleted file mode 100644 index 82ad0981bc4..00000000000 --- a/test/integration/test_extended_metadata_duplicate_elements.py +++ /dev/null @@ -1,210 +0,0 @@ -"""Integration test verifying that extended metadata does not create duplicate collection elements.""" - -from sqlalchemy import ( - func, - select, -) - -from galaxy.model import ( - DatasetCollectionElement, - HistoryDatasetCollectionAssociation, -) -from galaxy_test.base.populators import ( - DatasetCollectionPopulator, - DatasetPopulator, -) -from galaxy_test.driver.integration_util import IntegrationTestCase - - -class TestExtendedMetadataDuplicateElements(IntegrationTestCase): - dataset_populator: DatasetPopulator - framework_tool_and_types = True - - @classmethod - def handle_galaxy_config_kwds(cls, config): - super().handle_galaxy_config_kwds(config) - config["metadata_strategy"] = "extended" - config["retry_metadata_internally"] = False - - def setUp(self): - super().setUp() - self.dataset_populator = DatasetPopulator(self.galaxy_interactor) - self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor) - - def test_no_duplicate_elements_in_dynamic_list_output(self, history_id): - """Run a tool with dynamic collection output and verify no duplicate elements.""" - response = self.dataset_populator.run_tool( - "collection_creates_dynamic_list_of_pairs", - {"foo": "bar"}, - history_id, - ) - job_api_id = response["jobs"][0]["id"] - self.dataset_populator.wait_for_job(job_api_id, assert_ok=True) - - output_collections = response["output_collections"] - hdca_details = self.dataset_populator.get_history_collection_details( - history_id, content_id=output_collections[0]["id"] - ) - - sa_session = self._app.model.session - hdca_id = self._app.security.decode_id(hdca_details["id"]) - hdca = sa_session.get(HistoryDatasetCollectionAssociation, hdca_id) - assert hdca is not None - dc = hdca.collection - - # Verify exactly 3 outer elements (samp1, samp2, samp3) - no duplicates - outer_count = sa_session.scalar( - select(func.count()).where(DatasetCollectionElement.dataset_collection_id == dc.id) - ) - assert outer_count == 3, f"Expected 3 outer elements but found {outer_count}. Duplicate elements detected!" - - # Verify each inner pair has exactly 2 elements (forward, reverse) - for element in dc.elements: - assert element.child_collection is not None - inner_count = sa_session.scalar( - select(func.count()).where( - DatasetCollectionElement.dataset_collection_id == element.child_collection.id - ) - ) - assert inner_count == 2, ( - f"Expected 2 inner elements for '{element.element_identifier}' " - f"but found {inner_count}. Duplicate elements detected!" - ) - - def test_no_duplicate_elements_in_mapped_dynamic_collection(self, history_id): - """Map a tool with dynamic collection output over a list and verify no duplicate elements.""" - # Create a list of 2 tabular datasets - fetch_response = self.dataset_collection_populator.create_list_in_history( - history_id, - contents=["101\t1\n101\t2\n105\t3\n", "201\t10\n201\t20\n205\t30\n"], - ext="tabular", - wait=True, - ) - self.dataset_populator.wait_for_history(history_id, assert_ok=True) - hdca_id = fetch_response.json()["output_collections"][0]["id"] - - # Map collection_split_on_column over the list - inputs = { - "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}, - } - response = self.dataset_populator.run_tool( - "collection_split_on_column", - inputs, - history_id, - ) - - # Wait for all mapping jobs to complete - for job in response["jobs"]: - self.dataset_populator.wait_for_job(job["id"], assert_ok=True) - - # The implicit output should be a list:list collection - implicit_collections = response["implicit_collections"] - assert len(implicit_collections) == 1 - implicit_hdca_details = self.dataset_populator.get_history_collection_details( - history_id, content_id=implicit_collections[0]["id"] - ) - - sa_session = self._app.model.session - implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"]) - implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id) - assert implicit_hdca is not None - outer_dc = implicit_hdca.collection - - # The outer collection should have exactly 2 elements (one per input) - outer_count = sa_session.scalar( - select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id) - ) - assert ( - outer_count == 2 - ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!" - - # Each inner collection should have no duplicate elements - for element in outer_dc.elements: - inner_dc = element.child_collection - assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None" - inner_count = sa_session.scalar( - select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) - ) - # Each input has 2 unique first-column values, so 2 split files - assert ( - inner_count is not None and inner_count > 0 - ), f"Inner collection for '{element.element_identifier}' has no elements" - # Check for duplicates: element_identifiers should be unique - inner_elements = sa_session.scalars( - select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) - ).all() - identifiers = [e.element_identifier for e in inner_elements] - assert len(identifiers) == len(set(identifiers)), ( - f"Duplicate element identifiers found in inner collection " - f"for '{element.element_identifier}': {identifiers}" - ) - - def test_no_duplicate_elements_in_mapped_static_collection(self, history_id): - """Map a tool with a static collection output over a list and verify no duplicate elements. - - This covers the case where a tool declares a static collection output - (e.g. paired with forward/reverse) and is mapped over a list input, - producing a list:paired output. The per-job DC contains fully - initialized elements and must be included in io_dicts for metadata - serialization. - """ - hdca = self.dataset_collection_populator.create_list_in_history( - history_id, - contents=["line1\nline2\nline3\nline4\n", "lineA\nlineB\nlineC\nlineD\n"], - ext="txt", - wait=True, - ) - self.dataset_populator.wait_for_history(history_id, assert_ok=True) - hdca_id = hdca.json()["output_collections"][0]["id"] - - inputs = { - "input1": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}, - } - response = self.dataset_populator.run_tool( - "collection_creates_pair", - inputs, - history_id, - ) - - for job in response["jobs"]: - self.dataset_populator.wait_for_job(job["id"], assert_ok=True) - - implicit_collections = response["implicit_collections"] - assert len(implicit_collections) == 1 - implicit_hdca_details = self.dataset_populator.get_history_collection_details( - history_id, content_id=implicit_collections[0]["id"] - ) - - sa_session = self._app.model.session - implicit_hdca_id = self._app.security.decode_id(implicit_hdca_details["id"]) - implicit_hdca = sa_session.get(HistoryDatasetCollectionAssociation, implicit_hdca_id) - assert implicit_hdca is not None - outer_dc = implicit_hdca.collection - - # The outer collection should have exactly 2 elements (one per input) - outer_count = sa_session.scalar( - select(func.count()).where(DatasetCollectionElement.dataset_collection_id == outer_dc.id) - ) - assert ( - outer_count == 2 - ), f"Expected 2 outer elements but found {outer_count}. Duplicate elements detected in mapped output!" - - # Each inner pair should have exactly 2 elements (forward, reverse) with no duplicates - for element in outer_dc.elements: - inner_dc = element.child_collection - assert inner_dc is not None, f"Inner collection for '{element.element_identifier}' is None" - inner_count = sa_session.scalar( - select(func.count()).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) - ) - assert inner_count == 2, ( - f"Expected 2 inner elements for '{element.element_identifier}' " - f"but found {inner_count}. Duplicate elements detected!" - ) - inner_elements = sa_session.scalars( - select(DatasetCollectionElement).where(DatasetCollectionElement.dataset_collection_id == inner_dc.id) - ).all() - identifiers = [e.element_identifier for e in inner_elements] - assert set(identifiers) == { - "forward", - "reverse", - }, f"Expected forward/reverse but got {identifiers} for '{element.element_identifier}'"