Register shared loc path with existing table on merge

When install_tool_data_tables hits a table that's already registered
(e.g. __dbkeys__ from Galaxy's shipped refgenie config, or bwa_indexes
from the shipped tabular sample), we skip adding another <table> to
shed_tool_data_table_conf.xml — but that also skipped registering the
shared loc file path in the existing table's `filenames` dict. Data
Managers then couldn't locate where to persist entries
("Unable to determine filename for persisting data table '__dbkeys__'")
because get_filename_for_source had nothing to fall back to.

Register the shared loc path on the existing table with
tool_shed_repository=None so the new fallback in get_filename_for_source
finds it. Also fix the integration test for devteam/bwa: bwa_indexes /
bwa_mem_indexes already exist in tool_data_table_conf.xml.sample, so
their absence from shed_tool_data_table_conf.xml is correct dedup;
assert the shared loc file lands at tool_data_path instead.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Nate Coraor
2026-05-14 15:44:46 -04:00
co-authored by Claude Opus 4.7
parent 99f951108e
commit 962afa6200
3 changed files with 62 additions and 20 deletions
@@ -171,8 +171,11 @@ class ShedToolDataTableManager(BaseShedToolDataTableManager):
)
for file_elem in elem.findall("file"):
shared_path = file_elem.get("path")
basename = os.path.basename(shared_path) if shared_path else None
source_loc_sample = loc_basename_to_source.get(basename) if basename else None
if not shared_path:
continue
self._register_shared_loc_with_existing_table(existing_table, shared_path, elem)
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.
@@ -180,6 +183,30 @@ class ShedToolDataTableManager(BaseShedToolDataTableManager):
if new_rows:
existing_table.append_entries_with_attribution(new_rows, attribution)
def _register_shared_loc_with_existing_table(
self,
existing_table: TabularToolDataTable,
shared_path: str,
elem: Element,
) -> None:
"""Make the shared loc file path known to the existing table's ``filenames``.
Without this, Data Manager persist calls can't locate a destination loc file
when the table was first registered from a shipped or refgenie config that has
no tabular ``<file>`` entry (e.g. ``__dbkeys__``).
"""
if shared_path in existing_table.filenames:
return
existing_table.filenames[shared_path] = dict(
found=os.path.exists(shared_path),
filename=shared_path,
from_shed_config=True,
tool_data_path=self.app.config.tool_data_path,
config_element=elem,
tool_shed_repository=None,
errors=[],
)
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"
+14 -18
View File
@@ -62,27 +62,23 @@ class TestRepositoryInstallIntegrationTestCase(integration_util.IntegrationTestC
self.install_repository(*repo)
self.uninstall_repository(*repo)
def test_non_data_manager_install_registers_data_tables_without_repo_info(self):
"""Non-Data-Manager repos register tables in shed_tool_data_table_conf.xml without the
<tool_shed_repository> sub-element, and their loc files land at tool_data_path root."""
def test_non_data_manager_install_lands_loc_files_at_tool_data_path(self):
"""Non-Data-Manager repos install their .loc files to tool_data_path root, and any
<table> entries written to shed_tool_data_table_conf.xml have no <tool_shed_repository>
sub-element. Tables that already exist in Galaxy's shipped tool_data_table_conf.xml.sample
are deduped (not re-added to shed_tool_data_table_conf.xml)."""
non_dm_repo = ("devteam", "bwa", "051eba708f43")
non_dm_table_names = {"bwa_indexes", "bwa_mem_indexes"}
non_dm_loc_files = {"bwa_index.loc"}
self.install_repository(*non_dm_repo)
for loc_file in non_dm_loc_files:
shared_loc = os.path.join(self._app.config.tool_data_path, loc_file)
assert os.path.exists(shared_loc), f"Expected shared loc file at {shared_loc}"
shed_conf = self._app.config.shed_tool_data_table_config
table_elems = {t.get("name"): t for t in ET.parse(shed_conf).getroot().findall("table")}
missing = non_dm_table_names - table_elems.keys()
assert not missing, f"Expected tables not registered in {shed_conf}: {sorted(missing)}"
for name in non_dm_table_names:
assert (
table_elems[name].find("tool_shed_repository") is None
), f"Table {name!r} should not have a <tool_shed_repository> sub-element on new installs"
file_elems = table_elems[name].findall("file")
assert file_elems, f"Table {name!r} has no <file> entry"
for file_elem in file_elems:
loc_path = file_elem.get("path") or ""
assert loc_path.startswith(
self._app.config.tool_data_path
), f"Loc file for {name!r} not at tool_data_path root: {loc_path}"
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]
@@ -205,6 +205,25 @@ def test_second_install_with_empty_loc_sample_does_not_append(tmp_path):
existing.append_entries_with_attribution.assert_not_called()
def test_second_install_registers_shared_loc_with_existing_table(tmp_path):
"""When merging into an existing table whose ``filenames`` is empty (e.g. refgenie tables
or shipped tables whose loc resolution failed), the shared loc path must be registered so
Data Managers can persist entries."""
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)
existing.parse_file_fields.return_value = []
stdtm.app.tool_data_tables.data_tables = {"all_fasta": existing}
stdtm.install_tool_data_tables(repo, samples)
shared_loc = os.path.join(tool_data_path, "all_fasta.loc")
assert shared_loc in existing.filenames
info = existing.filenames[shared_loc]
assert info["tool_shed_repository"] is None
assert info["from_shed_config"] is True
def test_parse_table_columns_aliases_name_to_value():
from galaxy.tool_shed.tools.data_table_manager import _parse_table_columns