Merge pull request #22679 from natefoo/issue-21448-skip-shed-tool-data-tables

Don't install data table entries for non-DM tool installs

lib/galaxy/tool_shed/galaxy_install/install_manager.py

@@ -170,6 +170,7 @@ class InstallRepositoryManager:
         session.add(tool_shed_repository)
         session.commit()
 
+        is_data_manager = "data_manager" in irmm_metadata_dict
         if "sample_files" in irmm_metadata_dict:
             sample_files = irmm_metadata_dict.get("sample_files", [])
             tool_index_sample_files = stdtm.get_tool_index_sample_files(sample_files)
@@ -186,23 +187,26 @@ class InstallRepositoryManager:
             )
             sample_files = irmm_metadata_dict.get("sample_files", [])
             tool_index_sample_files = stdtm.get_tool_index_sample_files(sample_files)
-            tool_data_path = self.app.config.tool_data_path
-            tool_util.copy_sample_files(tool_data_path, tool_index_sample_files, tool_path=tool_path)
+            # Galaxy-managed loc files for shed-installed tools live under tool_data_path/shed/
+            # so they're separated from admin-configured loc files at tool_data_path root.
+            shed_loc_dir = os.path.join(self.app.config.tool_data_path, "shed")
+            os.makedirs(shed_loc_dir, exist_ok=True)
+            tool_util.copy_sample_files(shed_loc_dir, tool_index_sample_files, tool_path=tool_path)
             sample_files_copied = [str(s) for s in tool_index_sample_files]
             repository_tools_tups = irmm.get_repository_tools_tups()
             if repository_tools_tups:
-                # Handle missing data table entries for tool parameters that are dynamically generated select lists.
-                repository_tools_tups = stdtm.handle_missing_data_table_entry(
-                    relative_install_dir, tool_path, repository_tools_tups
-                )
+                if is_data_manager:
+                    # Only Data Manager repos register data tables on install.
+                    repository_tools_tups = stdtm.handle_missing_data_table_entry(
+                        relative_install_dir, tool_path, repository_tools_tups
+                    )
                 # Handle missing index files for tool parameters that are dynamically generated select lists.
                 repository_tools_tups, sample_files_copied = tool_util.handle_missing_index_file(
                     self.app, tool_path, sample_files, repository_tools_tups, sample_files_copied
                 )
-                # Copy remaining sample files included in the repository to the ~/tool-data directory of the
-                # local Galaxy instance.
+                # Copy remaining sample files included in the repository to the shed loc dir.
                 tool_util.copy_sample_files(
-                    tool_data_path, sample_files, tool_path=tool_path, sample_files_copied=sample_files_copied
+                    shed_loc_dir, sample_files, tool_path=tool_path, sample_files_copied=sample_files_copied
                 )
                 self.tpm.add_to_tool_panel(
                     repository_name=tool_shed_repository.name,

lib/galaxy/tool_shed/tools/data_table_manager.py

@@ -8,6 +8,10 @@ from typing import (
 
 from galaxy.tool_shed.galaxy_install.client import InstallationTarget
 from galaxy.tool_shed.util import hg_util
+from galaxy.tool_util.data import (
+    DataTableColumnMismatch,
+    TabularToolDataTable,
+)
 from galaxy.util import (
     Element,
     SubElement,
@@ -25,6 +29,14 @@ log = logging.getLogger(__name__)
 RequiredAppT = Union["BasicSharedApp", InstallationTarget]
 
 
+def _parse_table_columns(table_elem: Element) -> dict[str, int]:
+    """Parse a ``<table>`` element's column spec into a name->index mapping."""
+    columns, _, _ = TabularToolDataTable.parse_column_spec_element(table_elem)
+    if "value" in columns and "name" not in columns:
+        columns["name"] = columns["value"]
+    return columns
+
+
 class BaseShedToolDataTableManager:
     def __init__(self, app: RequiredAppT):
         self.app = app
@@ -146,33 +158,108 @@ class ShedToolDataTableManager(BaseShedToolDataTableManager):
             os.makedirs(target_dir)
         return target_dir, tool_path, relative_target_dir
 
+    def _merge_loc_sample_entries(
+        self,
+        existing_table: TabularToolDataTable,
+        elem: Element,
+        loc_basename_to_source: dict,
+        tool_shed_repository: "ToolShedRepository",
+    ) -> None:
+        """Append any non-comment rows from this install's .loc.sample files to the shared loc file,
+        attributing them to the installing repository. 99% of .loc.sample files are empty/comments,
+        in which case this is a no-op.
+
+        Skipped for non-tabular table types (e.g. ``RefgenieToolDataTable``) whose
+        ``parse_file_fields`` is not designed to read a ``.loc.sample``.
+        """
+        if getattr(existing_table, "type_key", "tabular") != "tabular":
+            return
+        attribution = (
+            f"Added by {tool_shed_repository.owner}/{tool_shed_repository.name}"
+            f"@{tool_shed_repository.installed_changeset_revision}"
+        )
+        for file_elem in elem.findall("file"):
+            shared_path = file_elem.get("path")
+            if not shared_path:
+                continue
+            basename = os.path.basename(shared_path)
+            source_loc_sample = loc_basename_to_source.get(basename)
+            if not source_loc_sample or not os.path.exists(source_loc_sample):
+                continue
+            # Keep `${__HERE__}` literal so the appended rows match the shared loc's existing format.
+            new_rows = existing_table.parse_file_fields(source_loc_sample, here="${__HERE__}")
+            if new_rows:
+                existing_table.append_entries_with_attribution(new_rows, attribution)
+
+    def _shed_config_has_matching_entry(self, table_name: str, elem: Element) -> bool:
+        """Return True if ``shed_tool_data_table_config`` already has a ``<table>`` with the same
+        ``name`` and identical set of ``<file path>`` entries as ``elem``.
+
+        Used to avoid writing duplicate ``<table>`` entries for tables that have already been
+        registered by a prior install. The shed config remains the persistent source of the
+        shared loc file's association with the data table name — on reload, ``merge_tool_data_table``
+        re-applies the filename info to the in-memory table.
+        """
+        config = self.app.config.shed_tool_data_table_config
+        if not config or not os.path.exists(config):
+            return False
+        try:
+            tree, _ = xml_util.parse_xml(config)
+        except OSError as e:
+            log.warning("Could not read shed_tool_data_table_config '%s' for dedup check: %s", config, e)
+            return False
+        if tree is None:
+            return False
+        elem_paths = {fe.get("path") for fe in elem.findall("file") if fe.get("path")}
+        for existing_elem in tree.getroot().findall("table"):
+            if existing_elem.get("name") != table_name:
+                continue
+            existing_paths = {fe.get("path") for fe in existing_elem.findall("file") if fe.get("path")}
+            if elem_paths == existing_paths:
+                return True
+        return False
+
     def install_tool_data_tables(self, tool_shed_repository: "ToolShedRepository", tool_index_sample_files):
         TOOL_DATA_TABLE_FILE_NAME = "tool_data_table_conf.xml"
         TOOL_DATA_TABLE_FILE_SAMPLE_NAME = f"{TOOL_DATA_TABLE_FILE_NAME}.sample"
         SAMPLE_SUFFIX = ".sample"
         SAMPLE_SUFFIX_OFFSET = -len(SAMPLE_SUFFIX)
+        LOC_SAMPLE_SUFFIX = ".loc.sample"
         target_dir, tool_path, relative_target_dir = self.get_target_install_dir(tool_shed_repository)
+        # Galaxy-managed loc files for shed-installed tools live under tool_data_path/shed/ so they
+        # are clearly separated from admin-configured loc files in tool_data_path and from any
+        # entries shipped via tool_data_table_conf.xml.sample.
+        shared_loc_dir = os.path.join(self.app.config.tool_data_path, "shed")
+        os.makedirs(shared_loc_dir, exist_ok=True)
+        # Map shared loc basename -> source .loc.sample, used to merge entries when a table is reinstalled.
+        loc_basename_to_source: dict[str, str] = {}
         for sample_file in tool_index_sample_files:
             path, filename = os.path.split(sample_file)
             target_filename = filename
             if target_filename.endswith(SAMPLE_SUFFIX):
                 target_filename = target_filename[:SAMPLE_SUFFIX_OFFSET]
             source_file = os.path.join(tool_path, sample_file)
+            if filename.endswith(LOC_SAMPLE_SUFFIX):
+                target_path_filename = os.path.join(shared_loc_dir, target_filename)
+                install_dest_dir = shared_loc_dir
+                loc_basename_to_source[target_filename] = source_file
+            else:
+                target_path_filename = os.path.join(target_dir, target_filename)
+                install_dest_dir = target_dir
             # We're not currently uninstalling index files, do not overwrite existing files.
-            target_path_filename = os.path.join(target_dir, target_filename)
             if not os.path.exists(target_path_filename) or target_filename == TOOL_DATA_TABLE_FILE_NAME:
                 shutil.copy2(source_file, target_path_filename)
             else:
                 log.debug(
                     "Did not copy sample file '%s' to install directory '%s' because file already exists.",
                     filename,
-                    target_dir,
+                    install_dest_dir,
                 )
             # For provenance and to simplify introspection, let's keep the original data table sample file around.
             if filename == TOOL_DATA_TABLE_FILE_SAMPLE_NAME:
                 shutil.copy2(source_file, os.path.join(target_dir, filename))
         tool_data_table_conf_filename = os.path.join(target_dir, TOOL_DATA_TABLE_FILE_NAME)
-        elems = []
+        elems: list = []
         if os.path.exists(tool_data_table_conf_filename):
             tree, error_message = xml_util.parse_xml(tool_data_table_conf_filename)
             if tree:
@@ -191,20 +278,37 @@ class ShedToolDataTableManager(BaseShedToolDataTableManager):
                 tool_data_table_conf_filename,
                 TOOL_DATA_TABLE_FILE_SAMPLE_NAME,
             )
+        registered_tables = self.app.tool_data_tables.data_tables
+        kept_elems: list = []
         for elem in elems:
-            if elem.tag == "table":
-                for file_elem in elem.findall("file"):
-                    path = file_elem.get("path", None)
-                    if path:
-                        file_elem.set("path", os.path.normpath(os.path.join(target_dir, os.path.split(path)[1])))
-                # Store repository info in the table tag set for trace-ability.
-                self.generate_repository_info_elem_from_repository(tool_shed_repository, parent_elem=elem)
-        if elems:
+            if elem.tag != "table":
+                kept_elems.append(elem)
+                continue
+            for file_elem in elem.findall("file"):
+                path = file_elem.get("path", None)
+                if path:
+                    new_path = os.path.normpath(os.path.join(shared_loc_dir, os.path.split(path)[1]))
+                    file_elem.set("path", new_path)
+            table_name = elem.get("name") or ""
+            incoming_columns = _parse_table_columns(elem)
+            self.app.tool_data_tables.assert_data_table_consistency(table_name, incoming_columns)
+            existing = registered_tables.get(table_name) if table_name else None
+            if isinstance(existing, TabularToolDataTable) and existing.columns is not None:
+                # Already registered with matching columns. Merge any rows from this install's
+                # .loc.sample(s) into the shared loc file.
+                self._merge_loc_sample_entries(existing, elem, loc_basename_to_source, tool_shed_repository)
+                if self._shed_config_has_matching_entry(table_name, elem):
+                    # An identical <table> entry already exists in shed_tool_data_table_config.
+                    # Don't write another one (it would just duplicate the shared loc reference).
+                    continue
+            kept_elems.append(elem)
+        if kept_elems:
             # Remove old data_table
-            os.unlink(tool_data_table_conf_filename)
+            if os.path.exists(tool_data_table_conf_filename):
+                os.unlink(tool_data_table_conf_filename)
             # Persist new data_table content.
-            self.app.tool_data_tables.to_xml_file(tool_data_table_conf_filename, elems)
-        return tool_data_table_conf_filename, elems
+            self.app.tool_data_tables.to_xml_file(tool_data_table_conf_filename, kept_elems)
+        return tool_data_table_conf_filename, kept_elems
 
 
 # For backwards compatibility with exisiting data managers
@@ -212,6 +316,7 @@ ToolDataTableManager = ShedToolDataTableManager
 
 
 __all__ = (
+    "DataTableColumnMismatch",
     "ToolDataTableManager",
     "ShedToolDataTableManager",
 )

lib/galaxy/tool_util/data/__init__.py

@@ -140,6 +140,19 @@ class FileNameInfoT(TypedDict):
 LoadInfoT = Tuple[Tuple[Element, Optional[StrPath]], Dict[str, Any]]
 
 
+class DataTableColumnMismatch(Exception):
+    """Two data tables share a name but declare different columns."""
+
+    def __init__(self, table_name: str, existing_columns: Dict[str, int], incoming_columns: Dict[str, int]):
+        self.table_name = table_name
+        self.existing_columns = existing_columns
+        self.incoming_columns = incoming_columns
+        super().__init__(
+            f"Data table {table_name!r} is already registered with columns {existing_columns}, "
+            f"refusing to register conflicting columns {incoming_columns}."
+        )
+
+
 class ToolDataTable(Dictifiable):
     type_key: str
     data: List[List[str]]
@@ -516,6 +529,38 @@ class TabularToolDataTable(ToolDataTable):
     def get_version_fields(self):
         return (self._loaded_content_version, self.get_fields())
 
+    @staticmethod
+    def parse_column_spec_element(
+        config_element: Element,
+    ) -> Tuple[Dict[str, int], int, Dict[str, str]]:
+        """
+        Parse column definitions into ``(columns, largest_index, empty_field_values)``.
+        Does not mutate or assert — callers layer their own validation.
+        """
+        columns: Dict[str, int] = {}
+        empty_field_values: Dict[str, str] = {}
+        largest_index = 0
+        columns_elem = config_element.find("columns")
+        if columns_elem is not None:
+            column_names = util.xml_text(columns_elem)
+            for index, name in enumerate(n.strip() for n in column_names.split(",")):
+                columns[name] = index
+                largest_index = index
+        else:
+            for column_elem in config_element.findall("column"):
+                name = column_elem.get("name")
+                index_attr = column_elem.get("index")
+                if name is None or index_attr is None:
+                    continue
+                index = int(index_attr)
+                columns[name] = index
+                if index > largest_index:
+                    largest_index = index
+                empty_field_value = column_elem.get("empty_field_value", None)
+                if empty_field_value is not None:
+                    empty_field_values[name] = empty_field_value
+        return columns, largest_index, empty_field_values
+
     def parse_column_spec(self, config_element: Element) -> None:
         """
         Parse column definitions, which can either be a set of 'column' elements
@@ -525,27 +570,12 @@ class TabularToolDataTable(ToolDataTable):
 
         A column named 'value' is required.
         """
-        self.columns: Dict[str, int] = {}
-        if config_element.find("columns") is not None:
-            column_names = util.xml_text(config_element.find("columns"))
-            column_names = [n.strip() for n in column_names.split(",")]
-            for index, name in enumerate(column_names):
-                self.columns[name] = index
-                self.largest_index = index
-        else:
-            self.largest_index = 0
+        if config_element.find("columns") is None:
             for column_elem in config_element.findall("column"):
-                name = column_elem.get("name")
-                assert name is not None, "Required 'name' attribute missing from column def"
-                index_attr = column_elem.get("index")
-                assert index_attr is not None, "Required 'index' attribute missing from column def"
-                index = int(index_attr)
-                self.columns[name] = index
-                if index > self.largest_index:
-                    self.largest_index = index
-                empty_field_value = column_elem.get("empty_field_value", None)
-                if empty_field_value is not None:
-                    self.empty_field_values[name] = empty_field_value
+                assert column_elem.get("name") is not None, "Required 'name' attribute missing from column def"
+                assert column_elem.get("index") is not None, "Required 'index' attribute missing from column def"
+        self.columns, self.largest_index, parsed_empty_field_values = self.parse_column_spec_element(config_element)
+        self.empty_field_values.update(parsed_empty_field_values)
         assert "value" in self.columns, "Required 'value' column missing from column def"
         if "name" not in self.columns:
             self.columns["name"] = self.columns["value"]
@@ -654,6 +684,7 @@ class TabularToolDataTable(ToolDataTable):
                     source_repo_info_model = source.repo_info
                 source_repo_info = source_repo_info_model.model_dump() if source_repo_info_model else None
         filename = default
+        shared_fallback: Optional[str] = None
         for name, value in self.filenames.items():
             repo_info = value.get("tool_shed_repository")
             if (not source_repo_info and not repo_info) or (
@@ -661,6 +692,14 @@ class TabularToolDataTable(ToolDataTable):
             ):
                 filename = name
                 break
+            if source_repo_info and not repo_info and shared_fallback is None:
+                # Loc file registered without repo_info (shared install) — use as fallback
+                # when a tool-shed source has no exact match. Preserves legacy behavior where
+                # a `<tool_shed_repository>`-tagged filename matches exactly.
+                shared_fallback = name
+        else:
+            if shared_fallback is not None:
+                filename = shared_fallback
         return filename
 
     def _add_entry(
@@ -732,6 +771,62 @@ class TabularToolDataTable(ToolDataTable):
                 fields_collapsed = f"{self.separator.join(fields)}\n"
                 data_table_fh.write(fields_collapsed.encode("utf-8"))
 
+    def append_entries_with_attribution(
+        self,
+        entries: List[List[str]],
+        attribution: str,
+        allow_duplicates: bool = False,
+    ) -> int:
+        """Append ``entries`` to this table's loc file with an attribution comment line.
+
+        When ``allow_duplicates`` is False, dedupes entries by the ``value`` column
+        (the table's primary key) against existing in-memory rows and the pending
+        batch. Writes a single ``{comment_char} {attribution}`` line before the first
+        new row. No-op if no rows survive the dedup.
+        """
+        filename: Optional[str] = self.get_filename_for_source(None)
+        if filename is None:
+            for name in self.filenames:
+                filename = name
+                break
+        if filename is None:
+            raise MessageException(f"Unable to determine filename for appending entries to data table '{self.name}'.")
+        value_index = self.columns.get("value", 0)
+        existing_values: Optional[Set[str]] = None
+        if not allow_duplicates:
+            existing_values = {row[value_index] for row in self.data if value_index < len(row)}
+        new_rows: List[List[str]] = []
+        for entry in entries:
+            fields = self._replace_field_separators(list(entry))
+            if self.largest_index >= len(fields):
+                log.warning(
+                    "Skipping fields (%s) for data table '%s': only %d field(s) provided.",
+                    fields,
+                    self.name,
+                    len(fields),
+                )
+                continue
+            if existing_values is not None and fields[value_index] in existing_values:
+                continue
+            new_rows.append(fields)
+            if existing_values is not None:
+                existing_values.add(fields[value_index])
+        if not new_rows:
+            return self._loaded_content_version
+        with FileLock(filename):
+            if os.path.exists(filename) and os.stat(filename).st_size > 0:
+                with open(filename, "rb+") as fh:
+                    fh.seek(-1, 2)
+                    if fh.read(1) not in (b"\n", b"\r"):
+                        fh.write(b"\n")
+            with open(filename, "ab") as fh:
+                fh.write(f"{self.comment_char} {attribution}\n".encode())
+                for fields in new_rows:
+                    fh.write(f"{self.separator.join(fields)}\n".encode())
+        for fields in new_rows:
+            self.data.append(fields)
+        return self._update_version()
+
     def _locate_filename(self, tool_data_file_path, entry_source, bundle_mode):
         if tool_data_file_path is not None:
             filename = tool_data_file_path
@@ -965,6 +1060,18 @@ class ToolDataTableManager(Dictifiable):
     def get_tables(self) -> Dict[str, "ToolDataTable"]:
         return self.data_tables
 
+    def assert_data_table_consistency(
+        self,
+        candidate_name: str,
+        candidate_columns: Dict[str, int],
+    ) -> None:
+        """Raise if ``candidate_name`` is already registered with different columns."""
+        existing = self.data_tables.get(candidate_name)
+        if existing is not None:
+            existing_columns = getattr(existing, "columns", None)
+            if existing_columns is not None and existing_columns != candidate_columns:
+                raise DataTableColumnMismatch(candidate_name, existing_columns, candidate_columns)
+
     def to_dict(
         self, view: str = "collection", value_mapper: Optional[Dict[str, Callable]] = None
     ) -> Dict[str, Dict[str, Any]]:

lib/tool_shed/test/base/testcase.py

@@ -1709,7 +1709,9 @@ class ShedTestCase(ShedApiTestCase):
                     ), f"The hard-coded Galaxy tool-data directory is missing for the {required_data_table_entry} entry."
                     assert os.path.exists(galaxy_tool_data_dir), "The Galaxy tool-data directory does not exist."
                     # Make sure the loc_file_name was correctly copied into the configured directory location.
-                    configured_file_location = os.path.join(self.tool_data_path, loc_file_name)
+                    # Shed-installed loc files now live under tool_data_path/shed/ to keep them
+                    # separate from admin-configured ones.
+                    configured_file_location = os.path.join(self.tool_data_path, "shed", loc_file_name)
                     assert os.path.isfile(
                         configured_file_location
                     ), f'The expected copied file "{configured_file_location}" is missing.'

test/integration/test_repository_operations.py

@@ -1,4 +1,5 @@
 import os
+import xml.etree.ElementTree as ET
 from collections import namedtuple
 
 from sqlalchemy import select
@@ -61,6 +62,29 @@ class TestRepositoryInstallIntegrationTestCase(integration_util.IntegrationTestC
         self.install_repository(*repo)
         self.uninstall_repository(*repo)
 
+    def test_non_data_manager_install_lands_loc_files_under_shed_subdir(self):
+        """Non-Data-Manager repos install their .loc files under tool_data_path/shed/ (keeping them
+        separate from admin-configured loc files at the tool_data_path root and from anything
+        shipped via tool_data_table_conf.xml.sample). Any <table> entries written to
+        shed_tool_data_table_conf.xml have no <tool_shed_repository> sub-element."""
+        non_dm_repo = ("devteam", "bwa", "051eba708f43")
+        non_dm_loc_files = {"bwa_mem_index.loc"}
+        self.install_repository(*non_dm_repo)
+        shed_loc_dir = os.path.join(self._app.config.tool_data_path, "shed")
+        for loc_file in non_dm_loc_files:
+            shared_loc = os.path.join(shed_loc_dir, loc_file)
+            assert os.path.exists(shared_loc), f"Expected shared loc file at {shared_loc}"
+            # The loc file should NOT also be written at the tool_data_path root.
+            assert not os.path.exists(
+                os.path.join(self._app.config.tool_data_path, loc_file)
+            ), f"Shed loc file should not land at tool_data_path root: {loc_file}"
+        shed_conf = self._app.config.shed_tool_data_table_config
+        if os.path.exists(shed_conf):
+            for table_elem in ET.parse(shed_conf).getroot().findall("table"):
+                assert (
+                    table_elem.find("tool_shed_repository") is None
+                ), f"Table {table_elem.get('name')!r} should not have a <tool_shed_repository> sub-element"
+
     def test_repository_update(self):
         response = self._install_repository(revision=REVISION_4, version="0.0.3", allow_upgraded=True)[0]
         assert int(response["ctx_rev"]) >= 4

test/unit/app/test_galaxy_install.py

@@ -43,6 +43,8 @@ def test_against_production_shed(tmp_path: Path):
         assert tool_guid in f.read()
     repo_path = tmp_path / "tools" / "toolshed.g2.bx.psu.edu" / "repos" / repo_owner / repo_name / repo_revision
     assert repo_path.exists()
+    # All shed installs now register data tables; the per-revision config is written for any repo
+    # with sample files, regardless of whether it's a Data Manager.
     tool_data_table_path = (
         tmp_path
         / "tool_data"

test/unit/tool_shed/test_data_table_manager.py

@@ -0,0 +1,279 @@
+"""Unit tests for shed-side data table install behavior."""
+
+import os
+from unittest import mock
+
+import pytest
+
+from galaxy.tool_shed.tools.data_table_manager import (
+    DataTableColumnMismatch,
+    ShedToolDataTableManager,
+)
+from galaxy.tool_util.data import (
+    TabularToolDataTable,
+    ToolDataTableManager,
+)
+from galaxy.util import (
+    Element,
+    SubElement,
+)
+
+
+class _FakeTableRegistry(ToolDataTableManager):
+    def __init__(self):
+        self.data_tables: dict = {}
+        self.to_xml_calls: list = []
+
+    def to_xml_file(self, shed_tool_data_table_config, new_elems=None, remove_elems=None):
+        self.to_xml_calls.append((shed_tool_data_table_config, new_elems))
+        with open(shed_tool_data_table_config, "wb") as fh:
+            fh.write(b"<tables/>")
+
+
+SAMPLE_TABLE_CONF = """\
+<tables>
+    <table name="all_fasta" comment_char="#">
+        <columns>value, dbkey, name, path</columns>
+        <file path="tool-data/all_fasta.loc" />
+    </table>
+</tables>
+"""
+
+LOC_SAMPLE_CONTENT = "# all_fasta.loc sample\n"
+
+
+def _write(path: str, contents: str) -> None:
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w") as fh:
+        fh.write(contents)
+
+
+def _make_stdtm(tmp_path):
+    repo_dir = str(tmp_path / "repo")
+    tool_data_path = str(tmp_path / "tool-data")
+    shed_tool_data_path = str(tmp_path / "shed_tool_data")
+    shed_tool_data_table_config = str(tmp_path / "shed_data_table_conf.xml")
+    relative_target_dir = "owner/name/abc"
+
+    os.makedirs(tool_data_path)
+    os.makedirs(os.path.join(shed_tool_data_path, relative_target_dir))
+    _write(os.path.join(repo_dir, "tool_data_table_conf.xml.sample"), SAMPLE_TABLE_CONF)
+    _write(os.path.join(repo_dir, "tool-data", "all_fasta.loc.sample"), LOC_SAMPLE_CONTENT)
+
+    app = mock.MagicMock(name="app")
+    app.config.tool_data_path = tool_data_path
+    app.config.shed_tool_data_path = shed_tool_data_path
+    app.config.shed_tool_data_table_config = shed_tool_data_table_config
+    registry = _FakeTableRegistry()
+    app.tool_data_tables = registry
+    captured = {"to_xml_calls": registry.to_xml_calls}
+
+    stdtm = ShedToolDataTableManager(app)
+
+    repo = mock.MagicMock(name="tool_shed_repository")
+    repo.name = "data_manager_fetch_genome_dbkeys_all_fasta"
+    repo.owner = "iuc"
+    repo.installed_changeset_revision = "abc"
+    repo.tool_shed = "tool-shed"
+    repo.get_tool_relative_path.return_value = (repo_dir, relative_target_dir)
+
+    sample_files = [
+        "tool_data_table_conf.xml.sample",
+        os.path.join("tool-data", "all_fasta.loc.sample"),
+    ]
+    return stdtm, repo, sample_files, captured, tool_data_path, shed_tool_data_path, relative_target_dir
+
+
+def _registered_table(columns, filenames=None, type_key="tabular"):
+    existing = mock.MagicMock(spec=TabularToolDataTable)
+    existing.columns = columns
+    existing.filenames = filenames or {}
+    existing.type_key = type_key
+    existing.parse_file_fields.return_value = []
+    return existing
+
+
+def _write_shed_config_with_entry(stdtm, table_name, file_path):
+    """Pre-populate ``shed_tool_data_table_config`` with a single ``<table>`` matching ``elem``."""
+    shed_config = stdtm.app.config.shed_tool_data_table_config
+    contents = f"""<?xml version="1.0"?>
+<tables>
+    <table name="{table_name}" comment_char="#">
+        <columns>value, dbkey, name, path</columns>
+        <file path="{file_path}" />
+    </table>
+</tables>
+"""
+    with open(shed_config, "w") as fh:
+        fh.write(contents)
+
+
+def test_loc_file_lands_under_shed_subdir_not_per_revision(tmp_path):
+    stdtm, repo, samples, captured, tool_data_path, shed_tool_data_path, _ = _make_stdtm(tmp_path)
+    _, kept_elems = stdtm.install_tool_data_tables(repo, samples)
+
+    shared_loc = os.path.join(tool_data_path, "shed", "all_fasta.loc")
+    assert os.path.exists(shared_loc)
+    # The loc file does NOT land at the tool_data_path root (which is for admin-configured loc
+    # files) — Galaxy-managed shed loc files are isolated under tool_data_path/shed/.
+    assert not os.path.exists(os.path.join(tool_data_path, "all_fasta.loc"))
+    per_rev_loc = os.path.join(shed_tool_data_path, "owner/name/abc", "all_fasta.loc")
+    assert not os.path.exists(per_rev_loc)
+
+    assert len(kept_elems) == 1
+    file_elems = list(kept_elems[0].findall("file"))
+    assert len(file_elems) == 1
+    assert file_elems[0].get("path") == shared_loc
+    # New installs should not stamp a <tool_shed_repository> sub-element on the <table>:
+    # the loc-file location is deterministic and DMs on legacy installs fall through to
+    # the shared-no-repo_info match in get_filename_for_source.
+    assert kept_elems[0].find("tool_shed_repository") is None
+
+
+def test_existing_loc_file_is_not_overwritten(tmp_path):
+    stdtm, repo, samples, _, tool_data_path, _, _ = _make_stdtm(tmp_path)
+    shared_loc = os.path.join(tool_data_path, "shed", "all_fasta.loc")
+    os.makedirs(os.path.dirname(shared_loc), exist_ok=True)
+    _write(shared_loc, "preexisting DM-populated content\n")
+
+    stdtm.install_tool_data_tables(repo, samples)
+
+    with open(shared_loc) as fh:
+        assert fh.read() == "preexisting DM-populated content\n"
+
+
+def test_column_mismatch_raises(tmp_path):
+    stdtm, repo, samples, captured, _, _, _ = _make_stdtm(tmp_path)
+    stdtm.app.tool_data_tables.data_tables = {
+        "all_fasta": _registered_table({"value": 0, "name": 1, "path": 2}),
+    }
+
+    with pytest.raises(DataTableColumnMismatch) as exc_info:
+        stdtm.install_tool_data_tables(repo, samples)
+    assert exc_info.value.table_name == "all_fasta"
+    assert not captured["to_xml_calls"]
+
+
+def test_column_match_first_install_writes_table_entry(tmp_path):
+    """When a table is already registered in memory but has not yet been written to
+    ``shed_tool_data_table_config``, we still write a (stamp-less) ``<table>`` entry so the
+    shared loc file's association with the table survives reload."""
+    stdtm, repo, samples, captured, tool_data_path, _, _ = _make_stdtm(tmp_path)
+    matching_columns = {"value": 0, "dbkey": 1, "name": 2, "path": 3}
+    stdtm.app.tool_data_tables.data_tables = {
+        "all_fasta": _registered_table(matching_columns),
+    }
+
+    _, kept_elems = stdtm.install_tool_data_tables(repo, samples)
+
+    assert len(kept_elems) == 1
+    file_elems = list(kept_elems[0].findall("file"))
+    assert len(file_elems) == 1
+    assert file_elems[0].get("path") == os.path.join(tool_data_path, "shed", "all_fasta.loc")
+    assert kept_elems[0].find("tool_shed_repository") is None
+
+
+def test_column_match_subsequent_install_dedupes_shed_config_entry(tmp_path):
+    """If shed_tool_data_table_config already has a ``<table>`` with the same name and same
+    ``<file path>``, don't write another one."""
+    stdtm, repo, samples, captured, tool_data_path, _, _ = _make_stdtm(tmp_path)
+    matching_columns = {"value": 0, "dbkey": 1, "name": 2, "path": 3}
+    stdtm.app.tool_data_tables.data_tables = {
+        "all_fasta": _registered_table(matching_columns),
+    }
+    _write_shed_config_with_entry(stdtm, "all_fasta", os.path.join(tool_data_path, "shed", "all_fasta.loc"))
+
+    _, kept_elems = stdtm.install_tool_data_tables(repo, samples)
+
+    assert kept_elems == []
+    assert not captured["to_xml_calls"]
+
+
+def test_column_match_with_column_elements_writes_entry(tmp_path):
+    stdtm, repo, _, captured, tool_data_path, _, _ = _make_stdtm(tmp_path)
+    column_form_conf = """\
+<tables>
+    <table name="all_fasta" comment_char="#">
+        <column name="value" index="0" />
+        <column name="dbkey" index="1" />
+        <column name="name" index="2" />
+        <column name="path" index="3" />
+        <file path="tool-data/all_fasta.loc" />
+    </table>
+</tables>
+"""
+    repo_dir, _ = repo.get_tool_relative_path.return_value
+    _write(os.path.join(repo_dir, "tool_data_table_conf.xml.sample"), column_form_conf)
+    stdtm.app.tool_data_tables.data_tables = {
+        "all_fasta": _registered_table({"value": 0, "dbkey": 1, "name": 2, "path": 3}),
+    }
+
+    _, kept_elems = stdtm.install_tool_data_tables(
+        repo,
+        ["tool_data_table_conf.xml.sample", os.path.join("tool-data", "all_fasta.loc.sample")],
+    )
+    assert len(kept_elems) == 1
+    assert kept_elems[0].find("tool_shed_repository") is None
+
+
+def test_second_install_merges_loc_sample_rows_with_attribution(tmp_path):
+    stdtm, repo, samples, captured, tool_data_path, _, _ = _make_stdtm(tmp_path)
+    matching_columns = {"value": 0, "dbkey": 1, "name": 2, "path": 3}
+    existing = _registered_table(matching_columns)
+    incoming_rows = [["hg19", "hg19", "human (hg19)", "/data/hg19.fa"]]
+    existing.parse_file_fields.return_value = incoming_rows
+    stdtm.app.tool_data_tables.data_tables = {"all_fasta": existing}
+    # Pre-populate shed config so kept_elems stays empty — we're testing the row-merge path.
+    _write_shed_config_with_entry(stdtm, "all_fasta", os.path.join(tool_data_path, "shed", "all_fasta.loc"))
+
+    _, kept_elems = stdtm.install_tool_data_tables(repo, samples)
+
+    assert kept_elems == []
+    assert not captured["to_xml_calls"]
+    existing.append_entries_with_attribution.assert_called_once()
+    call_args = existing.append_entries_with_attribution.call_args
+    assert call_args.args[0] == incoming_rows
+    attribution = call_args.args[1]
+    assert "iuc/data_manager_fetch_genome_dbkeys_all_fasta" in attribution
+    assert "abc" in attribution
+
+
+def test_second_install_with_empty_loc_sample_does_not_append(tmp_path):
+    stdtm, repo, samples, captured, tool_data_path, _, _ = _make_stdtm(tmp_path)
+    matching_columns = {"value": 0, "dbkey": 1, "name": 2, "path": 3}
+    existing = _registered_table(matching_columns)
+    existing.parse_file_fields.return_value = []
+    stdtm.app.tool_data_tables.data_tables = {"all_fasta": existing}
+    _write_shed_config_with_entry(stdtm, "all_fasta", os.path.join(tool_data_path, "shed", "all_fasta.loc"))
+
+    _, kept_elems = stdtm.install_tool_data_tables(repo, samples)
+
+    assert kept_elems == []
+    assert not captured["to_xml_calls"]
+    existing.append_entries_with_attribution.assert_not_called()
+
+
+def test_merge_skipped_for_non_tabular_table_types(tmp_path):
+    """Refgenie-backed (and other non-tabular) tables don't get their .loc.sample parsed
+    during install — refgenie's ``parse_file_fields`` expects YAML, not a .loc, so calling
+    it on a ``.loc.sample`` is incorrect."""
+    stdtm, repo, samples, _, tool_data_path, _, _ = _make_stdtm(tmp_path)
+    matching_columns = {"value": 0, "dbkey": 1, "name": 2, "path": 3}
+    existing = _registered_table(matching_columns, type_key="refgenie")
+    stdtm.app.tool_data_tables.data_tables = {"all_fasta": existing}
+    _write_shed_config_with_entry(stdtm, "all_fasta", os.path.join(tool_data_path, "shed", "all_fasta.loc"))
+
+    _, kept_elems = stdtm.install_tool_data_tables(repo, samples)
+    assert kept_elems == []
+    existing.parse_file_fields.assert_not_called()
+    existing.append_entries_with_attribution.assert_not_called()
+
+
+def test_parse_table_columns_aliases_name_to_value():
+    from galaxy.tool_shed.tools.data_table_manager import _parse_table_columns
+
+    elem = Element("table")
+    cols = SubElement(elem, "columns")
+    cols.text = "value, path"
+    parsed = _parse_table_columns(elem)
+    assert parsed == {"value": 0, "path": 1, "name": 0}

test/unit/tool_shed/test_install_manager.py

@@ -0,0 +1,109 @@
+"""Unit tests for install-time data table registration gating."""
+
+from typing import (
+    Any,
+)
+from unittest import mock
+
+from galaxy.tool_shed.galaxy_install import install_manager
+
+INSTALL_MANAGER_MODULE = "galaxy.tool_shed.galaxy_install.install_manager"
+
+
+def _make_install_repository_manager():
+    app = mock.MagicMock(name="app")
+    app.install_model.context = mock.MagicMock(name="session")
+    register_mock = mock.MagicMock(name="add_new_entries_from_config_file")
+    app.tool_data_tables.add_new_entries_from_config_file = register_mock
+    irm = install_manager.InstallRepositoryManager.__new__(install_manager.InstallRepositoryManager)
+    irm.app = app
+    irm.install_model = app.install_model
+    irm.tpm = mock.MagicMock(name="tpm")
+    return irm, register_mock
+
+
+def _invoke_handle(irm, metadata_dict: dict[str, Any], repository_tools_tups: list[Any]):
+    irmm_instance = mock.MagicMock(name="irmm_instance")
+    irmm_instance.get_metadata_dict.return_value = metadata_dict
+    irmm_instance.get_repository_tools_tups.return_value = repository_tools_tups
+
+    stdtm_instance = mock.MagicMock(name="stdtm_instance")
+    stdtm_instance.get_tool_index_sample_files.return_value = []
+    stdtm_instance.install_tool_data_tables.return_value = (
+        "/dev/null/tool_data_table_conf.xml",
+        [object()],
+    )
+    stdtm_instance.handle_missing_data_table_entry.side_effect = lambda *_a, **_k: repository_tools_tups
+
+    patches = [
+        mock.patch(f"{INSTALL_MANAGER_MODULE}.InstalledRepositoryMetadataManager", return_value=irmm_instance),
+        mock.patch(f"{INSTALL_MANAGER_MODULE}.ShedToolDataTableManager", return_value=stdtm_instance),
+        mock.patch(
+            f"{INSTALL_MANAGER_MODULE}.repository_util.get_tool_shed_status_for_installed_repository", return_value=None
+        ),
+        mock.patch(f"{INSTALL_MANAGER_MODULE}.tool_util.copy_sample_files"),
+        mock.patch(
+            f"{INSTALL_MANAGER_MODULE}.tool_util.handle_missing_index_file",
+            side_effect=lambda *_a, **_k: (repository_tools_tups, []),
+        ),
+        mock.patch(f"{INSTALL_MANAGER_MODULE}.data_manager.DataManagerHandler"),
+    ]
+    for p in patches:
+        p.start()
+    try:
+        repo = mock.MagicMock(name="tool_shed_repository")
+        repo.changeset_revision = "abc"
+        repo.installed_changeset_revision = "abc"
+        irm._InstallRepositoryManager__handle_repository_contents(
+            tool_shed_repository=repo,
+            tool_path="/tmp/tool_path",
+            repository_clone_url="http://tool-shed/repos/owner/name",
+            relative_install_dir="owner/name/abc",
+            tool_shed="tool-shed",
+            tool_section=None,
+            shed_tool_conf=None,
+        )
+    finally:
+        for p in patches:
+            p.stop()
+    return stdtm_instance
+
+
+def test_non_data_manager_repo_registers_sample_files():
+    irm, register_mock = _make_install_repository_manager()
+    stdtm_instance = _invoke_handle(irm, {"sample_files": ["foo.loc.sample"]}, [])
+    stdtm_instance.install_tool_data_tables.assert_called_once()
+    register_mock.assert_called_once()
+
+
+def test_data_manager_repo_registers_sample_files():
+    irm, register_mock = _make_install_repository_manager()
+    metadata = {
+        "sample_files": ["foo.loc.sample"],
+        "tools": [{"id": "t"}],
+        "data_manager": {"data_managers": {}},
+    }
+    fake_tup = (mock.MagicMock(), "guid", mock.MagicMock())
+    stdtm_instance = _invoke_handle(irm, metadata, [fake_tup])
+    stdtm_instance.install_tool_data_tables.assert_called_once()
+    register_mock.assert_called_once()
+
+
+def test_non_data_manager_repo_skips_handle_missing_data_table_entry():
+    irm, _ = _make_install_repository_manager()
+    metadata = {"tools": [{"id": "t"}], "sample_files": []}
+    fake_tup = (mock.MagicMock(), "guid", mock.MagicMock())
+    stdtm_instance = _invoke_handle(irm, metadata, [fake_tup])
+    stdtm_instance.handle_missing_data_table_entry.assert_not_called()
+
+
+def test_data_manager_repo_invokes_handle_missing_data_table_entry():
+    irm, _ = _make_install_repository_manager()
+    metadata = {
+        "tools": [{"id": "t"}],
+        "sample_files": [],
+        "data_manager": {"data_managers": {}},
+    }
+    fake_tup = (mock.MagicMock(), "guid", mock.MagicMock())
+    stdtm_instance = _invoke_handle(irm, metadata, [fake_tup])
+    stdtm_instance.handle_missing_data_table_entry.assert_called_once()

test/unit/tool_util/data/test_tool_data.py

@@ -1,9 +1,21 @@
+import pytest
+
+from galaxy.tool_util.data import DataTableColumnMismatch
+
 LOC_ALPHA_CONTENTS_V2 = """
 data1	data1name	${__HERE__}/data1/entry.txt
 data2	data2name	${__HERE__}/data2/entry.txt
 data3	data3name	${__HERE__}/data3/entry.txt
 """
 
+COLUMN_DIVERGENT_TABLE_CONF_XML = """<tables>
+  <table name="testalpha" comment_char="#">
+    <columns>value, name, path, extra</columns>
+    <file path="{loc_path}" />
+  </table>
+</tables>
+"""
+
 
 def test_data_tables_as_dictionary(tdt_manager):
     assert "testalpha" in tdt_manager.data_tables
@@ -60,3 +72,94 @@ def test_to_json(merged_tdt_manager, tmp_path):
     assert not json_path.exists()
     merged_tdt_manager.to_json(json_path)
     assert json_path.exists()
+
+
+def test_assert_data_table_consistency_accepts_new_table(tdt_manager):
+    tdt_manager.assert_data_table_consistency(
+        "brand_new_table",
+        {"value": 0, "name": 1, "path": 2},
+    )
+
+
+def test_assert_data_table_consistency_accepts_matching_redefinition(tdt_manager):
+    existing = tdt_manager["testalpha"]
+    tdt_manager.assert_data_table_consistency("testalpha", existing.columns)
+
+
+def test_assert_data_table_consistency_raises_column_mismatch(tdt_manager):
+    with pytest.raises(DataTableColumnMismatch) as exc_info:
+        tdt_manager.assert_data_table_consistency(
+            "testalpha",
+            {"value": 0, "name": 1, "path": 2, "extra": 3},
+        )
+    assert exc_info.value.table_name == "testalpha"
+
+
+def test_get_filename_for_source_falls_back_to_shared_filename(tdt_manager):
+    table = tdt_manager["testalpha"]
+    [shared_filename] = list(table.filenames)
+    assert table.filenames[shared_filename].get("tool_shed_repository") is None
+    source_with_unknown_repo = {
+        "tool_shed": "tool-shed",
+        "repository_name": "repo",
+        "repository_owner": "owner",
+        "installed_changeset_revision": "abc",
+    }
+    assert table.get_filename_for_source(source_with_unknown_repo) == shared_filename
+
+
+def test_get_filename_for_source_prefers_exact_repo_match_over_shared(tdt_manager):
+    table = tdt_manager["testalpha"]
+    [shared_filename] = list(table.filenames)
+    legacy_info = {
+        "tool_shed": "tool-shed",
+        "repository_name": "legacy",
+        "repository_owner": "owner",
+        "installed_changeset_revision": "deadbeef",
+    }
+    legacy_filename = f"{shared_filename}.legacy"
+    table.filenames[legacy_filename] = dict(
+        found=True,
+        filename=legacy_filename,
+        from_shed_config=True,
+        tool_data_path=None,
+        config_element=None,
+        tool_shed_repository=legacy_info,
+        errors=[],
+    )
+    assert table.get_filename_for_source(legacy_info) == legacy_filename
+    other_info = dict(legacy_info, repository_name="unknown")
+    assert table.get_filename_for_source(other_info) == shared_filename
+
+
+def test_append_entries_with_attribution_appends_and_dedupes(tdt_manager):
+    table = tdt_manager["testalpha"]
+    [loc_filename] = list(table.filenames)
+    initial_rows = len(table.data)
+    new_entries = [
+        ["data3", "data3name", "${__HERE__}/data3/entry.txt"],
+        ["data1", "data1name", "${__HERE__}/data1/entry.txt"],  # duplicate, must be skipped
+    ]
+    table.append_entries_with_attribution(new_entries, "added by owner/foo@rev1")
+    with open(loc_filename) as fh:
+        contents = fh.read()
+    assert "# added by owner/foo@rev1" in contents
+    assert contents.count("data3\tdata3name") == 1
+    assert contents.count("data1\tdata1name") == 1
+    assert len(table.data) == initial_rows + 1
+
+
+def test_append_entries_with_attribution_noop_when_all_duplicates(tdt_manager):
+    table = tdt_manager["testalpha"]
+    [loc_filename] = list(table.filenames)
+    with open(loc_filename) as fh:
+        before = fh.read()
+    rows_before = list(table.data)
+    table.append_entries_with_attribution(
+        [["data1", "data1name", "${__HERE__}/data1/entry.txt"]],
+        "added by owner/foo@rev1",
+    )
+    with open(loc_filename) as fh:
+        after = fh.read()
+    assert after == before
+    assert table.data == rows_before