diff --git a/.ci/import_reference_data.sh b/.ci/import_reference_data.sh new file mode 100755 index 00000000..1aef4b31 --- /dev/null +++ b/.ci/import_reference_data.sh @@ -0,0 +1,52 @@ +#!/usr/bin/env bash +# Stage 3 of the IDC reference-data pipeline: import the reference-data bundles +# built on Galaxy (Stage 2) onto CVMFS. +# +# For each request under data-managers/, this locates the build's workflow +# invocation (via its history idc--), resolves the bundle +# dataset(s) from that invocation's named outputs, and imports each with +# galaxy-import-data-bundle - recording record// for idempotency +# (already-imported builds are skipped). +# +# It must run INSIDE a CVMFS transaction, i.e. where ${CVMFS_ROOT} is writable +# (the Jenkins job opens the transaction; see .ci/jenkins.sh). The Python +# environment must have bioblend + our scripts, and galaxy-import-data-bundle +# (from galaxy-maintenance-scripts) must be on PATH or given via IMPORT_CMD. +# +# Environment: +# GALAXY_URL Galaxy the bundles were built on (default test.galaxyproject.org) +# EPHEMERIS_API_KEY API key for GALAXY_URL (required) +# CVMFS_ROOT CVMFS repo root (default /cvmfs/idc.galaxyproject.org) +# PYTHON Python interpreter (default python3) +# IMPORT_CMD galaxy-import-data-bundle executable (default: on PATH) +set -euo pipefail + +: "${GALAXY_URL:=https://test.galaxyproject.org}" +: "${CVMFS_ROOT:=/cvmfs/idc.galaxyproject.org}" +: "${PYTHON:=python3}" +: "${IMPORT_CMD:=galaxy-import-data-bundle}" + +REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + +shopt -s nullglob +found=false +for req in "$REPO_DIR"/data-managers/*/*.yml "$REPO_DIR"/data-managers/*/*.yaml; do + found=true + dm="$(basename "$(dirname "$req")")" + version="$(basename "$req")"; version="${version%.*}" + echo "== reference data: ${dm}/${version} ==" + # Skip data that already exists in the Galaxy data table (the build stage + # skips it too, so there is no history to import). --print-new emits the + # request only if its data is not present. + if [ -z "$("$PYTHON" "$REPO_DIR/scripts/check_data_exists.py" "$req" --print-new --reference-galaxy "$GALAXY_URL")" ]; then + echo " already exists on $GALAXY_URL; skipping" + continue + fi + "$PYTHON" "$REPO_DIR/scripts/import_bundles.py" \ + --galaxy-url "$GALAXY_URL" \ + --history-name "idc-${dm}-${version}" \ + --dm "$dm" --version "$version" \ + --cvmfs-root "$CVMFS_ROOT" \ + --import-cmd "$IMPORT_CMD" +done +$found || echo "No reference-data requests under data-managers/." diff --git a/.ci/jenkins.sh b/.ci/jenkins.sh index 59282b70..30fe9740 100755 --- a/.ci/jenkins.sh +++ b/.ci/jenkins.sh @@ -4,8 +4,23 @@ set -euo pipefail # Set this variable to 'true' to publish on successful installation : ${PUBLISH:=false} +# Set to 'true' to import ONLY reference-data (data-managers/) bundles and skip +# the genome (tool-data) build+import entirely. Also enabled by commenting +# "@galaxybot deploy reference-data" on the PR (see check_bot_command). +: ${REFERENCE_DATA_ONLY:=false} + BUILD_GALAXY_URL="http://idc-build" PUBLISH_GALAXY_URL="https://usegalaxy.org" +# Galaxy where the IDC reference-data (workflow-bundle) builds run - Stage 2 +# (.github/workflows/build.yml) builds bundles here for Stage 3 to import. +REFERENCE_DATA_GALAXY_URL="https://test.galaxyproject.org" +# API key for $REFERENCE_DATA_GALAXY_URL. Stage 3 uses it (via bioblend) to look +# up the built bundles' dataset ids on that server; the bundle download itself is +# unauthenticated. This is a DISTINCT server from $PUBLISH_GALAXY_URL, so it needs +# its own key - do not reuse $EPHEMERIS_API_KEY, which authenticates the genome +# import against $PUBLISH_GALAXY_URL. Falls back to $EPHEMERIS_API_KEY only for +# backward compatibility when a reference-data-only key is not provided. +: ${REFERENCE_DATA_API_KEY:=${EPHEMERIS_API_KEY:-}} SSH_MASTER_SOCKET_DIR="${HOME}/.cache/idc" MAIN_BRANCH='main' @@ -34,7 +49,14 @@ USE_LOCAL_OVERLAYFS=false # Set to true to run the importer in a docker container USE_DOCKER="$USE_LOCAL_OVERLAYFS" -REMOTE_PYTHON=/opt/rh/rh-python38/root/usr/bin/python3 +# Python for the remote ephemeris/maintenance venvs. The Stratum 0's system +# python3 is only 3.9 (too old for galaxy-maintenance-scripts' deps, e.g. +# yacman>=1.0 which needs 3.10+), and CVMFS-provided Pythons aren't reliably +# present on every worker, so setup_remote_python() bootstraps a pinned +# standalone CPython with uv. Set REMOTE_PYTHON in the environment to use a +# specific interpreter instead and skip the uv bootstrap. +REMOTE_PYTHON="${REMOTE_PYTHON:-}" +: "${REMOTE_PYTHON_VERSION:=3.13}" REMOTE_WORKDIR_PARENT=/srv/idc # $EPHEMERIS_API_KEY and $IDC_VAULT_PASS should be set in the environment @@ -154,10 +176,15 @@ function check_bot_command() { log 'Checking for Github PR Bot commands' log_debug "Value of \$ghprbCommentBody is: ${ghprbCommentBody:-UNSET}" case "${ghprbCommentBody:-UNSET}" in + "@galaxybot deploy reference-data"*) + PUBLISH=true + REFERENCE_DATA_ONLY=true + ;; "@galaxybot deploy"*) PUBLISH=true ;; esac + $REFERENCE_DATA_ONLY && log "Reference-data-only deploy: skipping genome build/import" if $PUBLISH; then log "Publish requested; running build and import" else @@ -225,6 +252,27 @@ function setup_ephemeris() { } +function setup_remote_python() { + # Sets global $REMOTE_PYTHON to a pinned standalone CPython bootstrapped with + # uv, so the remote venvs don't depend on the host OS Python (too old) or on + # a CVMFS-provided Python being mounted on this particular worker/Stratum 0. + # Honor an explicit REMOTE_PYTHON from the environment and skip the bootstrap. + if [ -n "$REMOTE_PYTHON" ]; then + log "Using preset REMOTE_PYTHON=${REMOTE_PYTHON}" + return + fi + log "Bootstrapping remote Python ${REMOTE_PYTHON_VERSION} with uv" + # uv is a single static binary; install it into the (ephemeral) workdir and + # let it fetch a managed CPython. UV_INSTALL_DIR sets the binary location; + # INSTALLER_NO_MODIFY_PATH keeps it from touching the idc user's profile. + exec_on "curl -LsSf https://astral.sh/uv/install.sh | env UV_INSTALL_DIR='${REMOTE_WORKDIR}/uv' INSTALLER_NO_MODIFY_PATH=1 sh" + local uv="${REMOTE_WORKDIR}/uv/uv" + exec_on "$uv" python install "$REMOTE_PYTHON_VERSION" + REMOTE_PYTHON="$(exec_on "$uv" python find "$REMOTE_PYTHON_VERSION")" + log "Remote Python: ${REMOTE_PYTHON}" +} + + function setup_remote_ephemeris() { # Sets global $EPHEMERIS_BIN EPHEMERIS_BIN="${REMOTE_WORKDIR}/ephemeris/bin" @@ -424,6 +472,10 @@ function wait_for_build_galaxy() { function stop_build_galaxy() { + # The build Galaxy (and its ansible-venv) is only set up when there are + # genome data managers to run; an import-only run (e.g. reference-data-only, + # or "Nothing to build") never starts it, so there is nothing to tear down. + $BUILD_GALAXY_UP || return 0 . ./ansible-venv/bin/activate log "Stopping Build Galaxy" pushd ansible @@ -595,6 +647,45 @@ function import_tool_data_bundles() { } +# --------------------------------------------------------------------------- +# IDC reference-data (workflow-bundle) pipeline - Stage 3 import. +# NOTE: UNTESTED against live Jenkins/Stratum 0 - review before enabling. +# These import bundles built on $REFERENCE_DATA_GALAXY_URL by the Stage 2 build +# (.github/workflows/build.yml) for requests under data-managers/. Only the +# remote (non-docker) path is wired here; the USE_DOCKER/local-overlay path +# would need the container invocation like import_tool_data_bundles. +# --------------------------------------------------------------------------- + +function has_reference_data_requests() { + compgen -G "data-managers/*/*.yaml" >/dev/null 2>&1 || compgen -G "data-managers/*/*.yml" >/dev/null 2>&1 +} + + +function import_reference_data_bundles() { + local req dm version + log "Importing IDC reference-data bundles" + copy_to scripts/get_bundle_urls.py + copy_to scripts/import_bundles.py + for req in data-managers/*/*.yaml data-managers/*/*.yml; do + [ -e "$req" ] || continue + dm="$(basename "$(dirname "$req")")" + version="$(basename "$req")"; version="${version%.*}" + # import_bundles.py skips gracefully when there is no build history + # idc-- - which is the case when the build stage skipped this + # request because its data already exists. + log "Importing reference-data bundles for '${dm}/${version}'" + exec_on mkdir -p "/cvmfs/${REPO}/data" "/cvmfs/${REPO}/record/${dm}" + # import_bundles.py resolves the build's bundles from its workflow + # invocation (history idc--) and imports each, recording + # record// for idempotency. API key filtered by Jenkins. + exec_on "EPHEMERIS_API_KEY='$REFERENCE_DATA_API_KEY' TMPDIR='${REMOTE_WORKDIR}' ${EPHEMERIS_BIN}/python3 ${REMOTE_WORKDIR}/import_bundles.py \ + --galaxy-url '$REFERENCE_DATA_GALAXY_URL' --history-name 'idc-${dm}-${version}' \ + --dm '$dm' --version '$version' --cvmfs-root '/cvmfs/${REPO}' \ + --import-cmd '${GALAXY_MAINTENANCE_SCRIPTS_BIN}/galaxy-import-data-bundle'" + done +} + + function show_logs() { local lines= if [ -n "${1:-}" ]; then @@ -682,13 +773,21 @@ function do_import_local() { function do_import_remote() { start_ssh_control create_remote_workdir + setup_remote_python setup_remote_ephemeris # from this point forward $EPHEMERIS_BIN refers to remote - if generate_import_tasks; then + local have_genome_tasks=false have_reference_data=false + if ! $REFERENCE_DATA_ONLY && generate_import_tasks; then + have_genome_tasks=true + fi + # UNTESTED: also open a transaction when only reference-data requests exist + has_reference_data_requests && have_reference_data=true + if $have_genome_tasks || $have_reference_data; then setup_galaxy_maintenance_scripts "$WORKDIR" "$REMOTE_PYTHON" begin_transaction update_tool_data_table_conf - import_tool_data_bundles + $have_genome_tasks && import_tool_data_bundles + $have_reference_data && import_reference_data_bundles check_for_repo_changes post_install else @@ -706,7 +805,7 @@ function main() { detect_changes set_repo_vars setup_ephemeris - if generate_data_manager_tasks; then + if ! $REFERENCE_DATA_ONLY && generate_data_manager_tasks; then run_build_galaxy wait_for_build_galaxy #install_data_managers diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 00000000..bde5ee33 --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,94 @@ +name: Build reference-data bundles + +# Stage 2 of the IDC reference-data pipeline: when a request is merged to main, +# generate its data-manager-bundle workflow and run it on test.galaxyproject.org +# (via planemo). The build runs asynchronously (--no_wait); the resulting bundle +# datasets live in a history named idc--, which Stage 3 (Jenkins) +# later imports onto CVMFS. +# +# Requires the repository secret TEST_API_KEY (a test.galaxyproject.org API key). + +on: + push: + branches: + - main + paths: + - "data-managers/**" + workflow_dispatch: + inputs: + request: + description: "Specific request file to build (default: all requests changed in the push)" + required: false + default: "" + +concurrency: + group: idc-build + cancel-in-progress: false + +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 2 # need the previous commit to diff changed requests + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Install dependencies + run: pip install planemo "pydantic>=2" pyyaml gxformat2 + + - name: Select requests to build + run: | + set -euo pipefail + if [ -n "${{ github.event.inputs.request }}" ]; then + printf '%s\n' "${{ github.event.inputs.request }}" > candidates.txt + else + base="${{ github.event.before }}" + # branch-creation / unknown base reports an all-zero SHA; use previous commit + if [ -z "$base" ] || ! git cat-file -e "${base}^{commit}" 2>/dev/null; then + base="HEAD~1" + fi + git diff --name-only --diff-filter=AM "$base" "${{ github.sha }}" \ + -- 'data-managers/*/*.yml' 'data-managers/*/*.yaml' > candidates.txt || true + fi + # drop requests whose version is already published (offline ledger) ... + python scripts/pending_requests.py --from-file candidates.txt > pending.txt + # ... and whose data already exists in a Galaxy data table (authoritative; + # e.g. already provided by the byhand CVMFS). --print-new emits only the + # requests that still need building. + python scripts/check_data_exists.py --from-file pending.txt --print-new \ + --reference-galaxy https://test.galaxyproject.org > requests.txt + echo "Requests to build:"; cat requests.txt || true + + - name: Build bundles on test.galaxyproject.org + env: + TEST_API_KEY: ${{ secrets.TEST_API_KEY }} + run: | + set -euo pipefail + if [ ! -s requests.txt ]; then echo "Nothing to build."; exit 0; fi + while read -r req; do + [ -n "$req" ] || continue + dm=$(basename "$(dirname "$req")") + version=$(basename "$req"); version="${version%.*}" + echo "::group::Building ${dm}/${version}" + # --reference-galaxy: if a chained request's upstream database already + # exists on the Galaxy, reference it instead of rebuilding it. + python scripts/generate_build.py "$req" --outdir build \ + --reference-galaxy https://test.galaxyproject.org + b="build/${dm}/${version}" + planemo run "$b/workflow.gxwf.yml" "$b/job.yml" \ + --galaxy_url https://test.galaxyproject.org \ + --galaxy_user_key "$TEST_API_KEY" \ + --history_name "idc-${dm}-${version}" \ + --tags idc --no_wait \ + --output_json "$b/invocation.json" + echo "::endgroup::" + done < requests.txt + + - uses: actions/upload-artifact@v4 + if: always() + with: + name: idc-build + path: build/ + if-no-files-found: ignore diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml new file mode 100644 index 00000000..4506a03d --- /dev/null +++ b/.github/workflows/lint.yml @@ -0,0 +1,48 @@ +name: Lint reference-data requests + +# Stage 1 of the IDC reference-data pipeline: validate contributed request +# YAMLs and the build workflows they generate. Pure validation, no secrets, so +# it is safe to run on pull requests from forks. + +on: + pull_request: + paths: + - "data-managers/**" + - "published.yml" + - "scripts/**" + - "tests/**" + - ".github/workflows/lint.yml" + push: + branches: + - main + paths: + - "data-managers/**" + - "published.yml" + - "scripts/**" + - "tests/**" + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Install dependencies + run: pip install "pydantic>=2" pyyaml gxformat2 pytest + - name: Lint request files + run: python scripts/request_models.py + - name: Validate generated build workflows + # Generate a gxformat2 bundle workflow for every request and validate it + # with gxformat2 (strict schema, format2->native, semantic lint). Catches + # wiring errors - e.g. a chained request with no CHAIN_WIRING entry. + run: python scripts/generate_build.py --all --outdir "${RUNNER_TEMP}/idc-build-check" + - name: Run pipeline unit tests + run: python -m pytest tests/ -q + - name: Check whether requested data already exists + # Informational (non-blocking): query the tool data tables on + # test.galaxyproject.org (public, no key) and warn if a request's data is + # already available there from any source (e.g. the byhand CVMFS). The + # build stage skips these; here we just surface it to reviewers. + run: python scripts/check_data_exists.py --all --warn diff --git a/.gitignore b/.gitignore index 01b46db3..e7982e4b 100644 --- a/.gitignore +++ b/.gitignore @@ -106,3 +106,10 @@ working/ # Ansible installed artifacts ansible/collections ansible/roles/*.* + +# macOS +.DS_Store + +# ephemeris/shed-tools test output +tool_test_output.json +tool_test_output.html diff --git a/README.md b/README.md index 686681f3..d49a931e 100644 --- a/README.md +++ b/README.md @@ -123,6 +123,122 @@ genomes: - bfast ``` +## Contributing versioned reference data (workflow bundles) + +Beyond the genome-indexing pipeline above (`genomes.yml` + `data_managers.yml`), +the IDC supports **versioned reference databases** built by data managers — e.g. +`metaphlan_database_versioned`, `motus_db_versioned`, `samestr_db`. Each requested +version is built by running a Galaxy **data-manager-bundle workflow** on +[test.galaxyproject.org](https://test.galaxyproject.org), and the resulting +bundle is imported onto CVMFS by Jenkins. + +### How to request a new reference-data version + +Open a PR that adds a single YAML file at: + +``` +data-managers//.yaml +``` + +- `` is the name of the data manager's primary **data table** + (e.g. `motus_db_versioned`). It must be one of the file's `data_tables`. +- `` (the file name, without extension) is the version identity used for + the build history and for idempotency — it must be unique per data manager. + +**Standalone request** (a self-contained download/build), e.g. +`data-managers/motus_db_versioned/3.1.0.yaml`: + +```yaml +# Full, version-pinned Tool Shed GUID of the data manager tool (required). +tool_id: toolshed.g2.bx.psu.edu/repos/bgruening/data_manager_motus/motus_db_fetcher/3.1.0+galaxy0 +# Data table(s) the data manager populates (required). +data_tables: [motus_db_versioned] +# Tool parameters for this build. Each becomes a workflow input (use "|" for +# nested/conditional params, e.g. "db|build"). +params: + version: "3.1.0" +# Optional, human-facing provenance: +description: mOTUs profiler database, version 3.1.0 +doi: +``` + +**Chained request** (a data manager that builds from another database), e.g. +`data-managers/samestr_db/marker_db_mpa_vJan21.yaml` — SameStr builds from a +MetaPhlAn database: + +```yaml +tool_id: toolshed.g2.bx.psu.edu/repos/iuc/data_manager_samestr/samestr_db/1.2025.111+galaxy1 +data_tables: [samestr_db] +# The upstream table -> version this build depends on. A request file must exist +# at data-managers/metaphlan_database_versioned/.yaml so it is built +# first; its bundle is wired into this data manager's input. +depends_on: + metaphlan_database_versioned: mpa_vJan21_CHOCOPhlAnSGB_202103 +params: {} +description: SameStr marker database derived from MetaPhlAn mpa_vJan21 +``` + +### What happens to your PR + +1. **Lint** (GitHub Actions, on the PR): `scripts/request_models.py` validates the + request (version-pinned GUID, folder/table match, `depends_on` resolves, not + already in `published.yml`) and gxformat2-validates the workflow it generates. +2. **Build** (GitHub Actions, on merge to `main`): `scripts/generate_build.py` + turns the request into a gxformat2 data-manager-bundle workflow, and + `planemo run` executes it on test.galaxyproject.org into a history named + `idc--`. Each data manager runs in + `__data_manager_mode: bundle`, producing a downloadable bundle dataset. +3. **Import** (Jenkins): the bundle(s) are resolved from the build's workflow + invocation and imported onto CVMFS with `galaxy-import-data-bundle` + (`.ci/import_reference_data.sh` / `.ci/jenkins.sh`), recording + `record//` for idempotency. + +### Avoiding rebuilds of existing data ("does this already exist?") + +Reference data that a Galaxy already has is never rebuilt or re-imported. The +authoritative check is the target Galaxy's tool data table: +`GET /api/tool_data/` (public, no key) lists what is actually available +there — from *any* source, including the byhand `data.galaxyproject.org` CVMFS — +so we don't duplicate data that already exists. `scripts/check_data_exists.py` +performs this check and is used at three points: + +- **Lint** (informational): warns on the PR if a request's data already exists. +- **Build** (`build.yml`): skips building requests whose data already exists. +- **Import**: `import_bundles.py` skips gracefully when there is no build history + for a request (which is the case when the build skipped it). + +Version matching is heuristic (the identifying column differs per data manager), +so a request is considered present if its version, any `params` value, or any +`depends_on` version matches a table entry. + +For **chained** requests this also avoids redundant upstream work: if the +upstream database a request `depends_on` already exists in the data table, the +generated workflow references that existing entry directly (a single downstream +step) instead of rebuilding the upstream. If it does not exist, the upstream data +manager is added as a step and built first. + +### Adding a brand-new data manager + +To onboard a data manager that isn't used yet: + +1. Get the tool installed on test.galaxyproject.org by adding it to + [usegalaxy-tools](https://github.com/galaxyproject/usegalaxy-tools) + (`test.galaxyproject.org/data_managers.yml`). +2. Add its data table(s) to `config/tool_data_table_conf.xml` (columns must match + the data manager's `` definition). +3. If it builds from another database, add a wiring entry to `CHAIN_WIRING` in + `scripts/generate_build.py` describing the conditional selector and the input + parameter that receives the upstream bundle. + +### Testing locally + +```bash +pip install "pydantic>=2" pyyaml gxformat2 pytest +python scripts/request_models.py # lint all requests +python scripts/generate_build.py --all --outdir build # generate + gxformat2-validate +pytest tests/ # pipeline unit tests +``` + ## Testing This repo can be tested using a machine with Docker installed and by a user with Docker privledges. As a warning however, some of the genomes will take a LOT (>64GB) of RAM to index. @@ -136,4 +252,3 @@ Work has been done on some of the other data types, tools and data managers such If you want to use the reference data, please have a look at our [ansible-role](https://github.com/galaxyproject/ansible-cvmfs ) and the [example playbook](https://github.com/usegalaxy-eu/cvmfs-example). - diff --git a/config/tool_data_table_conf.xml b/config/tool_data_table_conf.xml index 20c9ee8a..7866893b 100644 --- a/config/tool_data_table_conf.xml +++ b/config/tool_data_table_conf.xml @@ -87,4 +87,16 @@ value, name, path
+ + value, name, dbkey, path, db_version + +
+ + value, name, path, version_file, samestr_version + +
+ + value, version, name, path + +
diff --git a/data-managers/metaphlan_database_versioned/mpa_vJan21_CHOCOPhlAnSGB_202103.yaml b/data-managers/metaphlan_database_versioned/mpa_vJan21_CHOCOPhlAnSGB_202103.yaml new file mode 100644 index 00000000..b64938b6 --- /dev/null +++ b/data-managers/metaphlan_database_versioned/mpa_vJan21_CHOCOPhlAnSGB_202103.yaml @@ -0,0 +1,9 @@ +# MetaPhlAn clade-specific marker gene database (Jan 2021 CHOCOPhlAn SGB 202103). +# Standalone build: the data manager downloads and builds the DB by version. +tool_id: toolshed.g2.bx.psu.edu/repos/iuc/data_manager_metaphlan_database_downloader/data_manager_metaphlan_download/4.0.6+galaxy5 +data_tables: + - metaphlan_database_versioned +params: + index: mpa_vJan21_CHOCOPhlAnSGB_202103 # the tool's "Version" select param +description: MetaPhlAn clade-specific marker genes (mpa_vJan21_CHOCOPhlAnSGB_202103) +doi: 10.1101/2020.11.19.388223 diff --git a/data-managers/motus_db_versioned/3.1.0.yaml b/data-managers/motus_db_versioned/3.1.0.yaml new file mode 100644 index 00000000..2716c498 --- /dev/null +++ b/data-managers/motus_db_versioned/3.1.0.yaml @@ -0,0 +1,8 @@ +# mOTUs database v3.1.0 (fetched from Zenodo record 7778108). +# Standalone build: the data manager downloads the DB by version. +tool_id: toolshed.g2.bx.psu.edu/repos/bgruening/data_manager_motus/motus_db_fetcher/3.1.0+galaxy1 +data_tables: + - motus_db_versioned +params: + version: "3.1.0" # the tool's "Database Version" select param +description: mOTUs profiler database, version 3.1.0 diff --git a/data-managers/samestr_db/marker_db_mpa_vJan21.yaml b/data-managers/samestr_db/marker_db_mpa_vJan21.yaml new file mode 100644 index 00000000..9f27020f --- /dev/null +++ b/data-managers/samestr_db/marker_db_mpa_vJan21.yaml @@ -0,0 +1,13 @@ +# SameStr marker database, built from a MetaPhlAn database. +# Chained build: samestr's "database" input is backed by the +# metaphlan_database_versioned data table, so the build workflow first runs the +# MetaPhlAn data manager (bundle mode) and feeds its output into this step. +tool_id: toolshed.g2.bx.psu.edu/repos/iuc/data_manager_samestr/samestr_db/1.2025.111+galaxy2 +data_tables: + - samestr_db +depends_on: + # Which MetaPhlAn version this SameStr DB is built against. A request file must + # exist at data-managers/metaphlan_database_versioned/.yaml. + metaphlan_database_versioned: mpa_vJan21_CHOCOPhlAnSGB_202103 +params: {} +description: SameStr marker database derived from MetaPhlAn mpa_vJan21_CHOCOPhlAnSGB_202103 diff --git a/published.yml b/published.yml new file mode 100644 index 00000000..88deb2bc --- /dev/null +++ b/published.yml @@ -0,0 +1,12 @@ +# Reference-data versions that have been built and published to CVMFS. +# +# This file is the source of truth for idempotency: the Stage 1 lint rejects a +# request whose (data_manager, version) already appears here, and the build +# stage skips it. Entries are added (by the publish automation) only after a +# bundle has been successfully imported onto CVMFS. +# +# Shape: +# published: +# : +# - +published: {} diff --git a/scripts/check_data_exists.py b/scripts/check_data_exists.py new file mode 100644 index 00000000..3d6e97c0 --- /dev/null +++ b/scripts/check_data_exists.py @@ -0,0 +1,144 @@ +#!/usr/bin/env python3 +"""Check whether a request's reference data already exists in a Galaxy data table. + +The most authoritative "does this already exist?" signal is the target Galaxy's +tool data table: ``GET /api/tool_data/`` (public, no API key) lists the +entries actually available there - from *any* source, including the byhand +``data.galaxyproject.org`` CVMFS - so we never rebuild or re-import data a Galaxy +already has. + +Matching the request's version to a table entry is done heuristically, because +the identifying column differs per data manager (e.g. MetaPhlAn keys on ``dbkey``, +mOTUs on ``value``, SameStr on the upstream MetaPhlAn value). An entry counts as +present if any of the request's identity strings - its version, its ``params`` +values, or its ``depends_on`` versions - equals any field of a row, or is the +``value`` column optionally followed by a ``-`` (e.g. a build date). + +Usage:: + + python scripts/check_data_exists.py --all # exit 1 if any exist + python scripts/check_data_exists.py --all --warn # annotate, exit 0 + python scripts/check_data_exists.py data-managers/motus_db_versioned/3.1.0.yaml +""" +import argparse +import json +import sys +import urllib.request +from pathlib import Path + +import yaml + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from request_models import ( # noqa: E402 + Request, + data_manager_name, + iter_request_files, + version_id, +) + +DEFAULT_GALAXY = "https://test.galaxyproject.org" + + +def fetch_table(galaxy_url: str, table: str) -> dict: + """GET /api/tool_data/
-> {columns, fields}. Public, no key needed.""" + url = f"{galaxy_url.rstrip('/')}/api/tool_data/{table}" + with urllib.request.urlopen(url, timeout=30) as resp: # noqa: S310 (fixed https host) + return json.load(resp) + + +def identity_strings(request: Request, version: str) -> set[str]: + """The strings that could identify this build in a data table entry.""" + candidates = {version} + candidates |= {str(v) for v in (request.params or {}).values()} + candidates |= {str(v) for v in (request.depends_on or {}).values()} + return {c for c in candidates if c} + + +def matching_value(table_data: dict, candidates: set[str]) -> str | None: + """The ``value`` column of the first row matching any candidate, else None.""" + for row in table_data.get("fields", []): + row_strings = [str(x) for x in row] + value = row_strings[0] if row_strings else "" + for candidate in candidates: + if candidate in row_strings or value == candidate or value.startswith(candidate + "-"): + return value + return None + + +def entry_exists(table_data: dict, candidates: set[str]) -> bool: + return matching_value(table_data, candidates) is not None + + +def resolve_existing_value(galaxy_url: str, table: str, version: str) -> str | None: + """The data-table ``value`` for an existing entry of ``version``, else None. + + Used to reference an already-built upstream database (e.g. a MetaPhlAn DB a + SameStr build depends on) instead of rebuilding it. + """ + try: + table_data = fetch_table(galaxy_url, table) + except Exception: + return None + return matching_value(table_data, {version}) + + +def request_exists(request: Request, version: str, galaxy_url: str) -> bool: + candidates = identity_strings(request, version) + for table in request.data_tables: + try: + table_data = fetch_table(galaxy_url, table) + except Exception: + continue # unknown/empty table -> treat as not-present + if entry_exists(table_data, candidates): + return True + return False + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("requests", nargs="*", help="Request file(s) to check") + parser.add_argument("--all", action="store_true", help="Check every request under data-managers/") + parser.add_argument("--from-file", help="Read request paths from this file (one per line)") + parser.add_argument("--reference-galaxy", default=DEFAULT_GALAXY, help="Galaxy whose data tables to query") + parser.add_argument("--warn", action="store_true", help="Annotate and exit 0 instead of failing") + parser.add_argument( + "--print-new", + action="store_true", + help="Print (stdout) the requests whose data does NOT exist yet; always exit 0. For build/import filtering.", + ) + args = parser.parse_args(argv) + + if args.all: + paths = iter_request_files() + else: + raw = list(args.requests) + if args.from_file: + raw += [ln.strip() for ln in Path(args.from_file).read_text().splitlines() if ln.strip()] + paths = [Path(r) for r in raw if Path(r).is_file()] + + new, existing = [], [] + for path in paths: + request = Request(**yaml.safe_load(Path(path).read_text())) + version = version_id(Path(path)) + if request_exists(request, version, args.reference_galaxy): + existing.append((path, data_manager_name(Path(path)), version)) + else: + new.append(path) + + if args.print_new: + for path in new: + print(path) + return 0 + + for path, dm, version in existing: + prefix = "::warning:: " if args.warn else "" + print(f"{prefix}{dm}/{version} already exists on {args.reference_galaxy} ({path})", file=sys.stderr) + + if not existing: + print(f"No requested reference data already exists on {args.reference_galaxy}.") + return 0 + return 0 if args.warn else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/generate_build.py b/scripts/generate_build.py new file mode 100644 index 00000000..7db27001 --- /dev/null +++ b/scripts/generate_build.py @@ -0,0 +1,313 @@ +#!/usr/bin/env python3 +"""Generate a Galaxy data-manager *bundle* workflow from an IDC request file. + +Given a request at ``data-managers//.yaml`` this emits a gxformat2 +workflow (``class: GalaxyWorkflow``) whose data-manager tool steps each run in +``__data_manager_mode: bundle``, plus the planemo job file supplying the build +parameters. Running that workflow on a Galaxy server (Stage 2, via planemo) +produces the reference-data *bundle* dataset(s) that Jenkins later imports onto +CVMFS. + +Build parameters are **exposed as workflow inputs** and connected to the tool +parameters (a ``string`` input feeds a ``select`` parameter just fine - see +Galaxy's ``lib/galaxy_test/workflow/multiple_text.gxwf.yml``); their concrete +values live in the generated ``job.yml``. Only the structural selector of a +chained tool (e.g. samestr's ``db_source.db_type``) is baked into ``tool_state``. + +Chained builds (a request with ``depends_on``) become multi-step workflows: the +upstream data manager runs first (also in bundle mode) and its ``out_file`` +bundle is wired into the downstream tool's data-table-backed input - the pattern +Galaxy's ``test_data_manager_workflow_bundle`` integration test uses to feed a +fetched genome into an indexer. + +Usage:: + + python scripts/generate_build.py data-managers/motus_db_versioned/3.1.0.yaml + python scripts/generate_build.py data-managers/samestr_db/marker_db_mpa_vJan21.yaml --outdir build + +For each request it writes ``///workflow.gxwf.yml`` and +``job.yml`` and prints the planemo command to run it. +""" +import argparse +import copy +import sys +from pathlib import Path + +import yaml + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from request_models import ( # noqa: E402 + DATA_MANAGERS_DIR, + Request, + data_manager_name, + iter_request_files, + version_id, +) + +# How to wire an upstream bundle into a downstream (chained) data manager. Keyed +# on (downstream data manager, upstream data table). ``db_type`` selects the +# branch of the downstream tool's conditional (baked into tool_state); +# ``connect_param`` is the downstream input parameter (gxformat2 ``|`` notation) +# that receives the upstream step's bundle output. +CHAIN_WIRING = { + ("samestr_db", "metaphlan_database_versioned"): { + "tool_state": {"db_source": {"db_type": "metaphlan"}}, + "connect_param": "db_source|database", + }, + ("samestr_db", "motus_db_versioned"): { + "tool_state": {"db_source": {"db_type": "motus"}}, + "connect_param": "db_source|motus_db", + }, +} + +OUT_FILE = "out_file" # every data manager tool's bundle output + + +def tool_version_of(tool_id: str) -> str: + """The trailing version component of a version-pinned toolshed GUID.""" + return tool_id.rsplit("/", 1)[1] + + +def deep_merge(base: dict, overlay: dict) -> dict: + out = dict(base) + for key, value in overlay.items(): + if isinstance(value, dict) and isinstance(out.get(key), dict): + out[key] = deep_merge(out[key], value) + else: + out[key] = value + return out + + +def load_request(path: Path) -> tuple[Request, str, str]: + doc = yaml.safe_load(path.read_text()) + return Request(**doc), data_manager_name(path), version_id(path) + + +class WorkflowBuilder: + """Accumulates gxformat2 steps, workflow inputs, and the planemo job.""" + + def __init__(self) -> None: + self.steps: dict = {} + self.inputs: dict = {} + self.outputs: dict = {} + self.job: dict = {} + + def add_step( + self, + step_key: str, + tool_id: str, + params: dict, + *, + baked_state: dict | None = None, + connections: dict | None = None, + input_prefix: str = "", + ) -> None: + tool_state = dict(baked_state or {}) + tool_state["__data_manager_mode"] = "bundle" + + in_map: dict = {} + # Each build parameter -> a workflow input (string), connected to the + # tool parameter and given its value in the job file. + for param_path, value in params.items(): + input_name = (input_prefix + param_path).replace("|", "_") + self.inputs[input_name] = {"type": "string"} + self.job[input_name] = value + in_map[param_path] = {"source": input_name} + # Upstream-bundle connections (chained builds). + for param_path, source in (connections or {}).items(): + in_map[param_path] = {"source": source} + + step: dict = { + "tool_id": tool_id, + "tool_version": tool_version_of(tool_id), + "tool_state": tool_state, + } + if in_map: + step["in"] = in_map + self.steps[step_key] = step + # Expose this data manager's bundle as a named workflow output, so the + # invocation surfaces the bundle dataset directly (consumed in Stage 3). + self.outputs[f"{step_key}_bundle"] = {"outputSource": f"{step_key}/{OUT_FILE}"} + + def workflow(self, label: str) -> dict: + return { + "class": "GalaxyWorkflow", + "label": label, + "inputs": self.inputs, + "outputs": self.outputs, + "steps": self.steps, + } + + +def _resolve_request_path(dm: str, version: str) -> Path: + for ext in (".yaml", ".yml"): + candidate = DATA_MANAGERS_DIR / dm / f"{version}{ext}" + if candidate.exists(): + return candidate + raise SystemExit( + f"Cannot resolve upstream request data-managers/{dm}/{version}.yaml - " + f"it must exist so the chained build's upstream step can be generated." + ) + + +def build(request: Request, dm: str, version: str, reference_galaxy: str | None = None) -> tuple[dict, dict]: + """Return (gxformat2 workflow dict, planemo job dict) for one request. + + For a chained request (``depends_on``): if ``reference_galaxy`` is given and + the upstream database already exists in that Galaxy's data table, the existing + entry is referenced directly (no upstream build step). Otherwise the upstream + data manager is added as a step and rebuilt. + """ + from check_data_exists import resolve_existing_value + + wb = WorkflowBuilder() + + connections: dict = {} + baked_state: dict = {} + for up_table, up_version in (request.depends_on or {}).items(): + wiring = CHAIN_WIRING.get((dm, up_table)) + if wiring is None: + raise SystemExit( + f"No chain wiring defined for downstream {dm!r} depending on {up_table!r}. " + f"Add an entry to CHAIN_WIRING in scripts/generate_build.py." + ) + baked_state = deep_merge(baked_state, wiring["tool_state"]) + + existing_value = ( + resolve_existing_value(reference_galaxy, up_table, up_version) if reference_galaxy else None + ) + if existing_value is not None: + # Reference the already-built upstream entry via a workflow input. + input_name = wiring["connect_param"].replace("|", "_") + wb.inputs[input_name] = {"type": "string"} + wb.job[input_name] = existing_value + connections[wiring["connect_param"]] = input_name + else: + # Rebuild the upstream data manager as a step and wire its bundle. + up_request, up_dm, _ = load_request(_resolve_request_path(up_table, up_version)) + wb.add_step(up_dm, up_request.tool_id, up_request.params, input_prefix=f"{up_dm}_") + connections[wiring["connect_param"]] = f"{up_dm}/{OUT_FILE}" + + wb.add_step( + dm, + request.tool_id, + request.params, + baked_state=baked_state or None, + connections=connections or None, + ) + return wb.workflow(f"IDC bundle: {dm} {version}"), wb.job + + +def validate_workflow(workflow: dict) -> None: + """Validate a generated gxformat2 workflow with gxformat2 itself. + + Runs three checks and raises ValueError on the first failure: + 1. strict schema validation (Format2StrictModel), + 2. format2 -> native conversion (structural / connection sanity), + 3. the core gxformat2 semantic linter. + """ + try: + from gxformat2 import python_to_workflow + from gxformat2.lint import ( + Format2StrictModel, + LintContext, + lint_format2, + ) + except ImportError as exc: # pragma: no cover - depends on environment + raise ValueError( + "gxformat2 is required to validate generated workflows; install it " + "(it ships with planemo) or pass --no-validate." + ) from exc + + try: + Format2StrictModel(**workflow) + except Exception as exc: + raise ValueError(f"gxformat2 strict schema validation failed: {exc}") from exc + + try: + python_to_workflow(copy.deepcopy(workflow)) + except Exception as exc: + raise ValueError(f"gxformat2 format2->native conversion failed: {exc}") from exc + + lint_context = LintContext(level="error") + lint_format2(lint_context, workflow, raw_dict=workflow) + if lint_context.found_errors: + messages = "; ".join(str(m) for m in lint_context.error_messages) + raise ValueError(f"gxformat2 lint reported errors: {messages}") + + +def write_build( + request: Request, + dm: str, + version: str, + outdir: Path, + validate: bool = True, + reference_galaxy: str | None = None, +) -> Path: + workflow, job = build(request, dm, version, reference_galaxy=reference_galaxy) + if validate: + validate_workflow(workflow) + build_dir = outdir / dm / version + build_dir.mkdir(parents=True, exist_ok=True) + wf_path = build_dir / "workflow.gxwf.yml" + (wf_path).write_text(yaml.safe_dump(workflow, sort_keys=False)) + (build_dir / "job.yml").write_text(yaml.safe_dump(job, sort_keys=False) if job else "{}\n") + return wf_path + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("requests", nargs="*", help="Request YAML file(s) under data-managers/") + parser.add_argument( + "--all", + action="store_true", + help="Process every request file under data-managers/ (instead of listing them)", + ) + parser.add_argument("--outdir", default="build", help="Where to write generated builds (default: build/)") + parser.add_argument( + "--galaxy-url", + default="https://test.galaxyproject.org", + help="Galaxy URL for the printed planemo command", + ) + parser.add_argument( + "--no-validate", + dest="validate", + action="store_false", + help="Skip gxformat2 validation of the generated workflow", + ) + parser.add_argument( + "--reference-galaxy", + default=None, + help=( + "If a chained request's upstream database already exists in this " + "Galaxy's data table, reference it instead of rebuilding it." + ), + ) + args = parser.parse_args(argv) + + if args.all: + request_paths = iter_request_files() + elif args.requests: + request_paths = [Path(r).resolve() for r in args.requests] + else: + parser.error("provide request file(s) or --all") + + outdir = Path(args.outdir) + for path in request_paths: + request, dm, version = load_request(path) + wf_path = write_build( + request, dm, version, outdir, validate=args.validate, reference_galaxy=args.reference_galaxy + ) + job_path = wf_path.parent / "job.yml" + history = f"idc-{dm}-{version}" + print(f"# {dm} {version}") + print(f"planemo run {wf_path} {job_path} \\") + print(f" --galaxy_url {args.galaxy_url} --galaxy_user_key $TEST_GALAXY_KEY \\") + print(f' --history_name "{history}" --tags idc --no_wait \\') + print(f" --output_json {wf_path.parent / 'invocation.json'}") + print() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/get_bundle_urls.py b/scripts/get_bundle_urls.py new file mode 100644 index 00000000..bb161796 --- /dev/null +++ b/scripts/get_bundle_urls.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +"""Resolve the reference-data *bundle* download URLs produced by a build. + +Stage 2 runs a data-manager-bundle workflow on a Galaxy server; each data +manager step declares its bundle as a named workflow output (``_bundle``, +see scripts/generate_build.py). This script turns a finished **workflow +invocation** into the list of bundle dataset download URLs that Stage 3 +(``galaxy-import-data-bundle``) imports onto CVMFS. + +Unlike the legacy ``.ci/get-bundle-url.py`` - which returned a single bundle +from a history - a chained build (e.g. samestr, which also builds MetaPhlAn) +produces **several** bundles in one invocation, and all of them are returned. + +The invocation can be supplied three ways: + +* ``--invocation-json FILE`` - a saved invocation dict (no Galaxy needed; used + by the offline unit tests and handy for debugging), +* ``--invocation-id ID`` - fetched from Galaxy via bioblend, +* ``--history-name NAME`` - fallback that scans the history for + ``data_manager_json`` datasets (order preserved), for builds not driven by a + named-output workflow. + +Prints one bundle URL per line (import loop friendly) and, with +``--record-file``, writes a YAML provenance record. +""" +import argparse +import os +import sys + +EXT = "data_manager_json" +DEFAULT_BUNDLE_SUFFIX = "_bundle" +HDA_SRC = "hda" + + +def bundle_url(galaxy_url: str, dataset_id: str) -> str: + """Download URL for a bundle dataset (composite zip of the bundle dir).""" + return f"{galaxy_url.rstrip('/')}/api/datasets/{dataset_id}/display?to_ext={EXT}" + + +def bundle_dataset_ids_from_invocation( + invocation: dict, suffix: str = DEFAULT_BUNDLE_SUFFIX +) -> dict[str, str]: + """Map bundle output label -> dataset id for a workflow invocation. + + Considers the invocation's named ``outputs`` (``dict[label -> {id, src}]``). + Keeps HDA outputs whose label ends with ``suffix`` (pass ``suffix=""`` to + take every HDA output). Order follows the invocation's output ordering. + """ + result: dict[str, str] = {} + for label, output in (invocation.get("outputs") or {}).items(): + if output.get("src", HDA_SRC) != HDA_SRC: + continue + if suffix and not label.endswith(suffix): + continue + dataset_id = output.get("id") + if not dataset_id: + raise ValueError(f"Invocation output {label!r} has no dataset id: {output!r}") + result[label] = dataset_id + return result + + +def bundles_from_history(gi, history_name: str, suffix: str = DEFAULT_BUNDLE_SUFFIX) -> dict[str, str]: + """Resolve a build's bundles from its history name (the stable key shared by + the build and import stages). + + Prefers the workflow **invocation** in that history: its named ``*_bundle`` + outputs are exactly the bundles this build produced, so this is precise even + if the history also holds a re-run or a failed job's output. Falls back to + scanning ``data_manager_json`` datasets only if the history has no invocation. + """ + histories = gi.histories.get_histories(name=history_name, deleted=False) + if not histories: + # No build history - e.g. the build was skipped because the data already + # exists. Return nothing so callers can skip gracefully. + return {} + history_id = histories[0]["id"] + + invocations = gi.invocations.get_invocations(history_id=history_id) + if invocations: + latest = sorted(invocations, key=lambda i: i.get("create_time", ""))[-1] + invocation = gi.invocations.show_invocation(latest["id"]) + return bundle_dataset_ids_from_invocation(invocation, suffix=suffix) + + datasets = gi.datasets.get_datasets(history_id=history_id, extension=EXT, order="create_time-asc") + return {f"{history_name}_{i}": d["id"] for i, d in enumerate(datasets)} + + +def _galaxy_connection(args): + from bioblend.galaxy import GalaxyInstance + + api_key = args.galaxy_api_key or os.environ.get("EPHEMERIS_API_KEY") + if not api_key: + raise SystemExit("No Galaxy API key (use --galaxy-api-key or set $EPHEMERIS_API_KEY)") + return GalaxyInstance(url=args.galaxy_url, key=api_key) + + +def _load_invocation(args) -> dict: + if args.invocation_json: + import json + + with open(args.invocation_json) as fh: + return json.load(fh) + if args.invocation_id: + gi = _galaxy_connection(args) + return gi.invocations.show_invocation(args.invocation_id) + raise SystemExit("internal: no invocation source") # guarded in main() + + +def _bundles_from_history(args) -> dict[str, str]: + """Fallback: every data_manager_json dataset in a named history, in order.""" + return bundles_from_history(_galaxy_connection(args), args.history_name) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("-g", "--galaxy-url", default="http://localhost:8080", help="Galaxy server URL") + parser.add_argument("-a", "--galaxy-api-key", help="Galaxy API key (or set $EPHEMERIS_API_KEY)") + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument("--invocation-json", help="Path to a saved invocation dict (offline)") + source.add_argument("--invocation-id", help="Workflow invocation id to fetch from Galaxy") + source.add_argument("--history-name", help="History to scan for data_manager_json datasets (fallback)") + parser.add_argument( + "--bundle-suffix", + default=DEFAULT_BUNDLE_SUFFIX, + help="Only take invocation outputs whose label ends with this (default: %(default)s; '' for all)", + ) + parser.add_argument("-r", "--record-file", help="Write a YAML provenance record here") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + + if args.history_name: + bundles = _bundles_from_history(args) + else: + invocation = _load_invocation(args) + bundles = bundle_dataset_ids_from_invocation(invocation, suffix=args.bundle_suffix) + + if not bundles: + raise SystemExit("No bundle datasets found for this build") + + urls = {label: bundle_url(args.galaxy_url, ds_id) for label, ds_id in bundles.items()} + + if args.record_file: + import yaml + + record = { + "galaxy_url": args.galaxy_url, + "invocation_id": args.invocation_id, + "history_name": args.history_name, + "bundles": [ + {"label": label, "dataset_id": bundles[label], "url": urls[label]} + for label in bundles + ], + } + with open(args.record_file, "w") as fh: + yaml.safe_dump(record, fh, sort_keys=False) + + for label in bundles: + print(urls[label]) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/import_bundles.py b/scripts/import_bundles.py new file mode 100644 index 00000000..d8a5299a --- /dev/null +++ b/scripts/import_bundles.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +"""Import a build's reference-data bundles onto CVMFS (Stage 3). + +Given a finished data-manager-bundle workflow invocation, this resolves every +bundle it produced (see scripts/get_bundle_urls.py) and imports each onto CVMFS +with ``galaxy-import-data-bundle`` (from galaxy-maintenance-scripts), which moves +the data under ``/data``, appends the new ``.loc`` rows, relativizes +symlinks, and reloads the tables. + +Idempotency mirrors the existing IDC importer's ``record/`` markers, generalized +to the reference-data identity: a build is skipped if +``/record//`` already exists, and that marker is written +after a successful import. + +This is meant to run inside the Jenkins CVMFS transaction (see .ci/jenkins.sh); +``--dry-run`` prints the exact commands without importing, so the wiring is +testable offline. +""" +import argparse +import os +import subprocess +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from get_bundle_urls import ( # noqa: E402 + DEFAULT_BUNDLE_SUFFIX, + bundle_dataset_ids_from_invocation, + bundle_url, + bundles_from_history, +) + + +def import_command(import_cmd: str, cvmfs_root: str, url: str) -> list[str]: + """The galaxy-import-data-bundle invocation for one bundle URL.""" + return [ + import_cmd, + "--tool-data-path", + f"{cvmfs_root}/data", + "--data-table-config-path", + f"{cvmfs_root}/config/tool_data_table_conf.xml", + url, + ] + + +def record_marker(cvmfs_root: str, dm: str, version: str) -> Path: + return Path(cvmfs_root) / "record" / dm / version + + +def _galaxy_connection(args): + from bioblend.galaxy import GalaxyInstance + + api_key = args.galaxy_api_key or os.environ.get("EPHEMERIS_API_KEY") + if not api_key: + raise SystemExit("No Galaxy API key (use --galaxy-api-key or set $EPHEMERIS_API_KEY)") + return GalaxyInstance(url=args.galaxy_url, key=api_key) + + +def resolve_bundles(args) -> dict[str, str]: + """Map bundle label -> dataset id, from whichever source was given.""" + if args.invocation_json: + import json + + with open(args.invocation_json) as fh: + invocation = json.load(fh) + return bundle_dataset_ids_from_invocation(invocation, suffix=args.bundle_suffix) + if args.invocation_id: + gi = _galaxy_connection(args) + invocation = gi.invocations.show_invocation(args.invocation_id) + return bundle_dataset_ids_from_invocation(invocation, suffix=args.bundle_suffix) + # history-name fallback (the stable key shared by the build and import stages) + return bundles_from_history(_galaxy_connection(args), args.history_name) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("-g", "--galaxy-url", default="http://localhost:8080", help="Galaxy server URL") + parser.add_argument("-a", "--galaxy-api-key", help="Galaxy API key (or set $EPHEMERIS_API_KEY)") + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument("--invocation-json", help="Path to a saved invocation dict (offline)") + source.add_argument("--invocation-id", help="Workflow invocation id to fetch from Galaxy") + source.add_argument("--history-name", help="History to scan for data_manager_json bundles") + parser.add_argument("--dm", required=True, help="Data manager name (record identity)") + parser.add_argument("--version", required=True, help="Version being imported (record identity)") + parser.add_argument("--cvmfs-root", default="/cvmfs/idc.galaxyproject.org", help="CVMFS repo root") + parser.add_argument("--bundle-suffix", default=DEFAULT_BUNDLE_SUFFIX, help="Bundle output label suffix") + parser.add_argument( + "--import-cmd", + default="galaxy-import-data-bundle", + help="galaxy-import-data-bundle executable (path)", + ) + parser.add_argument("--overwrite", action="store_true", help="Import even if a record marker exists") + parser.add_argument("--dry-run", action="store_true", help="Print commands without importing") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + + marker = record_marker(args.cvmfs_root, args.dm, args.version) + if marker.exists() and not args.overwrite: + print(f"Already imported: {args.dm}/{args.version} (record {marker} exists); skipping") + return 0 + + bundles = resolve_bundles(args) + if not bundles: + # Nothing to import - no build history/invocation (e.g. the build was + # skipped because the data already exists). Not an error. + print(f"No bundles to import for {args.dm}/{args.version}; skipping") + return 0 + + for label, dataset_id in bundles.items(): + url = bundle_url(args.galaxy_url, dataset_id) + cmd = import_command(args.import_cmd, args.cvmfs_root, url) + print(f"# import {label}") + print(" ".join(cmd)) + if not args.dry_run: + subprocess.run(cmd, check=True) + + if args.dry_run: + print(f"# (dry-run) would record: {marker}") + return 0 + + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text("\n".join(f"{label}: {ds}" for label, ds in bundles.items()) + "\n") + print(f"Recorded import: {marker}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/pending_requests.py b/scripts/pending_requests.py new file mode 100644 index 00000000..4b6e03ff --- /dev/null +++ b/scripts/pending_requests.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Filter request files down to those not yet published (pending a build). + +Given candidate request paths, print (one per line) those whose (data_manager, +version) does not already appear in published.yml. Used by the Stage 2 build +workflow to skip requests whose reference data is already on CVMFS. + +Usage:: + + python scripts/pending_requests.py --from-file candidates.txt + python scripts/pending_requests.py data-managers/motus_db_versioned/3.1.0.yaml + python scripts/pending_requests.py --all +""" +import argparse +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from request_models import ( # noqa: E402 + data_manager_name, + iter_request_files, + load_published, + version_id, +) + + +def pending(paths: list[Path], published: dict[str, list[str]]) -> list[Path]: + result = [] + for path in paths: + if version_id(path) not in (published.get(data_manager_name(path)) or []): + result.append(path) + return result + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("requests", nargs="*", help="Candidate request file(s)") + parser.add_argument("--from-file", help="Read candidate request paths from this file (one per line)") + parser.add_argument("--all", action="store_true", help="Consider every request under data-managers/") + args = parser.parse_args(argv) + + if args.all: + candidates = iter_request_files() + else: + raw = list(args.requests) + if args.from_file: + raw += [line.strip() for line in Path(args.from_file).read_text().splitlines() if line.strip()] + candidates = [Path(r) for r in raw if Path(r).is_file()] + + for path in pending(candidates, load_published()): + print(path) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/request_models.py b/scripts/request_models.py new file mode 100644 index 00000000..c82b5c28 --- /dev/null +++ b/scripts/request_models.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Schema + linter for IDC reference-data *request* files. + +A contribution to the IDC is a single flat YAML file placed under:: + + data-managers//.yaml + +where ```` is the name of the data manager / primary data table +(e.g. ``metaphlan_database_versioned``) and ```` is the version being +requested (the file stem, e.g. ``mpa_vJan21_CHOCOPhlAnSGB_202103``). + +The file describes how to build one reference-data *bundle* on a Galaxy server +(the build itself runs a gxformat2 data-manager-bundle workflow via planemo - +see ``workflows/`` and ``scripts/generate_build.py``). This module validates +those files and is run as the Stage 1 CI lint:: + + python scripts/request_models.py # lint everything + python scripts/request_models.py data-managers/motus_db_versioned/3.1.0.yaml + +Exit code is non-zero if any file fails validation. +""" +import sys +from pathlib import Path +from typing import Optional + +import yaml +from pydantic import ( + BaseModel, + ConfigDict, + field_validator, +) + +REPO_ROOT = Path(__file__).resolve().parent.parent +DATA_MANAGERS_DIR = REPO_ROOT / "data-managers" +PUBLISHED_PATH = REPO_ROOT / "published.yml" + +# A toolshed GUID looks like: +# toolshed.g2.bx.psu.edu/repos//// +TOOL_ID_PREFIX = "toolshed.g2.bx.psu.edu/repos/" + + +class Request(BaseModel): + """One requested reference-data version -> one bundle build.""" + + model_config = ConfigDict(extra="forbid") + + # Full toolshed GUID of the data manager tool that builds this data. + tool_id: str + # Data table(s) the data manager populates (its bundle carries these rows). + data_tables: list[str] + # Tool parameters for this specific build, e.g. {"index": "mpa_vJan21_..."}. + params: dict[str, object] = {} + # For chained builds: maps an upstream data table name -> the upstream + # version this build depends on. e.g. samestr depends on a metaphlan db: + # depends_on: {metaphlan_database_versioned: mpa_vJan21_CHOCOPhlAnSGB_202103} + depends_on: Optional[dict[str, str]] = None + + # Human-facing provenance (unused by the build, but reviewed in the PR). + description: Optional[str] = None + doi: Optional[str] = None + + @field_validator("tool_id") + @classmethod + def _tool_id_is_a_guid(cls, v: str) -> str: + if not v.startswith(TOOL_ID_PREFIX): + raise ValueError( + f"tool_id must be a production Tool Shed GUID starting with {TOOL_ID_PREFIX!r}, got: {v!r}" + ) + # Require the full, version-pinned GUID: + # host/repos/owner/repo/tool/version (6 slash-separated parts) + # The pinned tool version is what makes a reference-data build + # reproducible and auditable, so it is mandatory - not left to whatever + # revision happens to be installed at build time. + if len(v.split("/")) < 6: + raise ValueError( + f"tool_id must be a version-pinned GUID host/repos/owner/repo/tool/version " + f"(the trailing tool version is required for reproducibility): {v!r}" + ) + return v + + @field_validator("data_tables") + @classmethod + def _data_tables_non_empty(cls, v: list[str]) -> list[str]: + if not v: + raise ValueError("data_tables must list at least one data table") + return v + + +class LintError(Exception): + pass + + +def data_manager_name(path: Path) -> str: + """The data manager identity == the parent directory name.""" + return path.parent.name + + +def version_id(path: Path) -> str: + """The requested version == the file stem.""" + return path.stem + + +def load_published() -> dict[str, list[str]]: + if not PUBLISHED_PATH.exists(): + return {} + doc = yaml.safe_load(PUBLISHED_PATH.read_text()) or {} + return doc.get("published", {}) or {} + + +def iter_request_files() -> list[Path]: + if not DATA_MANAGERS_DIR.is_dir(): + return [] + return sorted(p for p in DATA_MANAGERS_DIR.rglob("*.y*ml") if p.is_file()) + + +def lint_file(path: Path, published: dict[str, list[str]]) -> list[str]: + """Return a list of error strings for one request file (empty == ok).""" + errors: list[str] = [] + try: + rel = path.relative_to(REPO_ROOT) + except ValueError: + rel = path + + # Structural: must be data-managers//.yaml (exactly one level deep). + try: + depth = path.relative_to(DATA_MANAGERS_DIR).parts + except ValueError: + return [f"{rel}: request files must live under data-managers/"] + if len(depth) != 2: + errors.append( + f"{rel}: request files must be at data-managers//.yaml " + f"(got {len(depth)} path component(s) under data-managers/)" + ) + return errors + + dm = data_manager_name(path) + version = version_id(path) + + try: + doc = yaml.safe_load(path.read_text()) + except yaml.YAMLError as exc: + return [f"{rel}: invalid YAML: {exc}"] + if not isinstance(doc, dict): + return [f"{rel}: top-level YAML must be a mapping"] + + try: + req = Request(**doc) + except Exception as exc: # pydantic ValidationError et al. + return [f"{rel}: schema validation failed:\n{exc}"] + + # The directory name should be one of the data tables the DM populates, + # so the on-disk identity matches what the bundle actually writes. + if dm not in req.data_tables: + errors.append( + f"{rel}: directory name {dm!r} is not in data_tables {req.data_tables} - " + f"the folder must be named after the data manager's primary data table" + ) + + # Already published? Then this request is a no-op / duplicate. + if version in published.get(dm, []): + errors.append( + f"{rel}: version {version!r} is already published for {dm!r} (see published.yml)" + ) + + # Chained builds: the upstream version must itself have a request file, so the + # generator can build the upstream step of the workflow. + for up_table, up_version in (req.depends_on or {}).items(): + up_path = DATA_MANAGERS_DIR / up_table / f"{up_version}.yaml" + up_path_yml = DATA_MANAGERS_DIR / up_table / f"{up_version}.yml" + if not up_path.exists() and not up_path_yml.exists(): + errors.append( + f"{rel}: depends_on {up_table}=={up_version} but no request file " + f"exists at data-managers/{up_table}/{up_version}.yaml - " + f"add the upstream request so it can be built first" + ) + + return errors + + +def main(argv: Optional[list[str]] = None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + published = load_published() + + if argv: + files = [Path(a).resolve() for a in argv] + else: + files = iter_request_files() + + if not files: + print("No request files found under data-managers/ - nothing to lint.") + return 0 + + all_errors: list[str] = [] + for path in files: + errs = lint_file(path, published) + if errs: + all_errors.extend(errs) + else: + print(f"ok: {path.relative_to(REPO_ROOT)}") + + if all_errors: + print("\n".join(["", "Lint failed:", *all_errors]), file=sys.stderr) + return 1 + print(f"\nAll {len(files)} request file(s) valid.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 00000000..a9ea44ef --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,325 @@ +"""Offline tests for the IDC reference-data pipeline scripts. + +These exercise the pure, network-free logic: request validation, build-workflow +generation (+ gxformat2 validation), bundle-URL resolution from an invocation, +and the CVMFS import command assembly / record idempotency. No Galaxy or +toolshed access is required. +""" +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +SCRIPTS = REPO_ROOT / "scripts" +sys.path.insert(0, str(SCRIPTS)) + +import check_data_exists as cde # noqa: E402 +import generate_build as gb # noqa: E402 +import get_bundle_urls as gburls # noqa: E402 +import import_bundles as imp # noqa: E402 +import pending_requests as pr # noqa: E402 +import request_models as rm # noqa: E402 + +SEEDS = { + "metaphlan": REPO_ROOT / "data-managers/metaphlan_database_versioned/mpa_vJan21_CHOCOPhlAnSGB_202103.yaml", + "motus": REPO_ROOT / "data-managers/motus_db_versioned/3.1.0.yaml", + "samestr": REPO_ROOT / "data-managers/samestr_db/marker_db_mpa_vJan21.yaml", +} + + +# --------------------------------------------------------------------------- # +# request_models +# --------------------------------------------------------------------------- # +def test_seed_requests_lint_clean(): + published: dict = {} + for path in SEEDS.values(): + assert rm.lint_file(path, published) == [], path + + +def test_tool_id_must_be_version_pinned(): + with pytest.raises(Exception): + rm.Request(tool_id="toolshed.g2.bx.psu.edu/repos/iuc/repo/tool", data_tables=["t"]) + with pytest.raises(Exception): + rm.Request(tool_id="testtoolshed.g2.bx.psu.edu/repos/iuc/repo/tool/1.0", data_tables=["t"]) + # 6-part (version-pinned) is accepted + rm.Request(tool_id="toolshed.g2.bx.psu.edu/repos/iuc/repo/tool/1.0", data_tables=["t"]) + + +def test_request_rejects_unimplemented_checksum_field(): + with pytest.raises(Exception): + rm.Request( + tool_id="toolshed.g2.bx.psu.edu/repos/iuc/repo/tool/1.0", + data_tables=["t"], + checksum="sha256:abc", + ) + + +def test_lint_rejects_dir_table_mismatch(tmp_path): + p = rm.DATA_MANAGERS_DIR / "motus_db_versioned" / "_probe.yaml" + p.write_text("tool_id: toolshed.g2.bx.psu.edu/repos/iuc/a/b/1\ndata_tables: [other]\n") + try: + errors = rm.lint_file(p, {}) + finally: + p.unlink() + assert any("directory name" in e for e in errors) + + +def test_lint_rejects_already_published(): + errors = rm.lint_file(SEEDS["motus"], {"motus_db_versioned": ["3.1.0"]}) + assert any("already published" in e for e in errors) + + +# --------------------------------------------------------------------------- # +# generate_build +# --------------------------------------------------------------------------- # +def test_standalone_build_single_bundle_step(): + request, dm, version = gb.load_request(SEEDS["motus"]) + workflow, job = gb.build(request, dm, version) + assert list(workflow["steps"]) == ["motus_db_versioned"] + step = workflow["steps"]["motus_db_versioned"] + assert step["tool_state"]["__data_manager_mode"] == "bundle" + assert workflow["outputs"] == {"motus_db_versioned_bundle": {"outputSource": "motus_db_versioned/out_file"}} + assert job == {"version": "3.1.0"} + gb.validate_workflow(workflow) # gxformat2 strict + native + lint + + +def test_chained_build_wires_upstream_bundle(): + request, dm, version = gb.load_request(SEEDS["samestr"]) + workflow, _ = gb.build(request, dm, version) + assert list(workflow["steps"]) == ["metaphlan_database_versioned", "samestr_db"] + samestr = workflow["steps"]["samestr_db"] + # structural selector baked, database wired from the metaphlan step's bundle + assert samestr["tool_state"]["db_source"]["db_type"] == "metaphlan" + assert samestr["in"]["db_source|database"]["source"] == "metaphlan_database_versioned/out_file" + # both bundles exposed as workflow outputs + assert set(workflow["outputs"]) == {"metaphlan_database_versioned_bundle", "samestr_db_bundle"} + gb.validate_workflow(workflow) + + +def test_chained_build_references_existing_upstream(monkeypatch): + # When the upstream metaphlan already exists, reference it instead of rebuilding. + monkeypatch.setattr(cde, "resolve_existing_value", lambda url, table, version: "mpa_vJan21_CHOCOPhlAnSGB_202103-04042023") + request, dm, version = gb.load_request(SEEDS["samestr"]) + workflow, job = gb.build(request, dm, version, reference_galaxy="https://test.galaxyproject.org") + # single step (samestr only) - no metaphlan build step + assert list(workflow["steps"]) == ["samestr_db"] + samestr = workflow["steps"]["samestr_db"] + assert samestr["tool_state"]["db_source"]["db_type"] == "metaphlan" + # database wired from a workflow input carrying the existing table value + assert samestr["in"]["db_source|database"]["source"] == "db_source_database" + assert job["db_source_database"] == "mpa_vJan21_CHOCOPhlAnSGB_202103-04042023" + assert set(workflow["outputs"]) == {"samestr_db_bundle"} + gb.validate_workflow(workflow) + + +def test_validate_rejects_broken_connection(): + workflow = { + "class": "GalaxyWorkflow", + "inputs": {}, + "outputs": {"b": {"outputSource": "s/out_file"}}, + "steps": { + "s": { + "tool_id": "toolshed.g2.bx.psu.edu/repos/iuc/r/t/1", + "tool_version": "1", + "tool_state": {"__data_manager_mode": "bundle"}, + "in": {"x": {"source": "nonexistent"}}, + } + }, + } + with pytest.raises(ValueError): + gb.validate_workflow(workflow) + + +# --------------------------------------------------------------------------- # +# get_bundle_urls +# --------------------------------------------------------------------------- # +STANDALONE_INV = {"outputs": {"motus_db_versioned_bundle": {"id": "ds1", "src": "hda"}}} +CHAIN_INV = { + "outputs": { + "metaphlan_database_versioned_bundle": {"id": "dsMETA", "src": "hda"}, + "samestr_db_bundle": {"id": "dsSAM", "src": "hda"}, + "some_report": {"id": "dsREP", "src": "hda"}, + } +} + + +def test_resolve_standalone_bundle(): + assert gburls.bundle_dataset_ids_from_invocation(STANDALONE_INV) == {"motus_db_versioned_bundle": "ds1"} + + +def test_resolve_chain_returns_both_bundles_excluding_non_bundle(): + result = gburls.bundle_dataset_ids_from_invocation(CHAIN_INV) + assert result == {"metaphlan_database_versioned_bundle": "dsMETA", "samestr_db_bundle": "dsSAM"} + + +def test_resolve_empty_suffix_takes_all_hda_outputs(): + assert set(gburls.bundle_dataset_ids_from_invocation(CHAIN_INV, suffix="")) == { + "metaphlan_database_versioned_bundle", + "samestr_db_bundle", + "some_report", + } + + +def test_bundle_url_format_and_trailing_slash(): + assert ( + gburls.bundle_url("https://test.galaxyproject.org/", "dsX") + == "https://test.galaxyproject.org/api/datasets/dsX/display?to_ext=data_manager_json" + ) + + +# --------------------------------------------------------------------------- # +# import_bundles +# --------------------------------------------------------------------------- # +def test_import_command_assembly(): + cmd = imp.import_command("galaxy-import-data-bundle", "/cvmfs/idc.galaxyproject.org", "http://u/bundle") + assert cmd == [ + "galaxy-import-data-bundle", + "--tool-data-path", + "/cvmfs/idc.galaxyproject.org/data", + "--data-table-config-path", + "/cvmfs/idc.galaxyproject.org/config/tool_data_table_conf.xml", + "http://u/bundle", + ] + + +def test_pending_requests_filters_published(): + paths = list(SEEDS.values()) + published = {"motus_db_versioned": ["3.1.0"]} + remaining = pr.pending(paths, published) + names = {p.parent.name for p in remaining} + assert "motus_db_versioned" not in names # already published -> filtered out + assert {"metaphlan_database_versioned", "samestr_db"} <= names + + +# --------------------------------------------------------------------------- # +# check_data_exists +# --------------------------------------------------------------------------- # +_META_TABLE = { + "columns": ["value", "name", "dbkey", "path", "db_version"], + "fields": [["mpa_vJan21_CHOCOPhlAnSGB_202103-04042023", "n", "mpa_vJan21_CHOCOPhlAnSGB_202103", "/p", "SGB"]], +} +_MOTUS_TABLE = {"columns": ["value", "version", "name", "path"], "fields": [["3.1.0", "3.1.0", "n", "/p"]]} + + +def test_entry_exists_matches_exact_field_and_value_prefix(): + assert cde.entry_exists(_META_TABLE, {"mpa_vJan21_CHOCOPhlAnSGB_202103"}) # dbkey exact + value prefix + assert not cde.entry_exists(_META_TABLE, {"mpa_vOct22_CHOCOPhlAnSGB_202212"}) + assert cde.entry_exists(_MOTUS_TABLE, {"3.1.0"}) # value exact + assert not cde.entry_exists(_MOTUS_TABLE, {"3.0.0"}) + assert not cde.entry_exists({"fields": []}, {"anything"}) + + +def test_identity_strings_include_params_and_depends_on(): + req = rm.Request( + tool_id="toolshed.g2.bx.psu.edu/repos/iuc/data_manager_samestr/samestr_db/1", + data_tables=["samestr_db"], + params={}, + depends_on={"metaphlan_database_versioned": "mpa_vJan21_CHOCOPhlAnSGB_202103"}, + ) + ids = cde.identity_strings(req, "marker_db_mpa_vJan21") + assert "marker_db_mpa_vJan21" in ids and "mpa_vJan21_CHOCOPhlAnSGB_202103" in ids + + +def test_request_exists_uses_table_lookup(monkeypatch): + req = gb.load_request(SEEDS["metaphlan"])[0] + monkeypatch.setattr(cde, "fetch_table", lambda url, table: _META_TABLE) + assert cde.request_exists(req, "mpa_vJan21_CHOCOPhlAnSGB_202103", "http://g") + monkeypatch.setattr(cde, "fetch_table", lambda url, table: {"fields": []}) + assert not cde.request_exists(req, "mpa_vJan21_CHOCOPhlAnSGB_202103", "http://g") + + +class _FakeGi: + """Minimal stand-in for a bioblend GalaxyInstance for history resolution.""" + + def __init__(self, invocations, invocation_detail, datasets=None): + self._invocations = invocations + self._invocation_detail = invocation_detail + self._datasets = datasets or [] + + outer = self + + class _Histories: + def get_histories(self, name, deleted=False): + return [{"id": "hist1", "name": name}] + + class _Invocations: + def get_invocations(self, history_id): + return outer._invocations + + def show_invocation(self, invocation_id): + return outer._invocation_detail + + class _Datasets: + def get_datasets(self, history_id, extension, order): + return outer._datasets + + self.histories = _Histories() + self.invocations = _Invocations() + self.datasets = _Datasets() + + +def test_history_resolution_prefers_latest_invocation(): + gi = _FakeGi( + invocations=[ + {"id": "old", "create_time": "2026-01-01T00:00:00"}, + {"id": "new", "create_time": "2026-02-01T00:00:00"}, + ], + invocation_detail=CHAIN_INV, # named *_bundle outputs + ) + result = gburls.bundles_from_history(gi, "idc-samestr_db-v1") + # precise: exactly the two bundle outputs, not a dataset scan + assert result == {"metaphlan_database_versioned_bundle": "dsMETA", "samestr_db_bundle": "dsSAM"} + + +def test_history_resolution_returns_empty_when_history_missing(): + class _NoHistory: + class histories: + @staticmethod + def get_histories(name, deleted=False): + return [] + + assert gburls.bundles_from_history(_NoHistory(), "idc-missing-1") == {} + + +def test_import_skips_gracefully_when_no_bundles(tmp_path, capsys): + inv = tmp_path / "inv.json" + inv.write_text('{"outputs": {}}') # invocation with no bundle outputs + rc = imp.main([ + "--invocation-json", str(inv), + "--dm", "motus_db_versioned", "--version", "3.1.0", + "--cvmfs-root", str(tmp_path), + ]) + assert rc == 0 + assert "skipping" in capsys.readouterr().out + + +def test_history_resolution_falls_back_to_dataset_scan_without_invocation(): + gi = _FakeGi( + invocations=[], + invocation_detail={}, + datasets=[{"id": "d0"}, {"id": "d1"}], + ) + result = gburls.bundles_from_history(gi, "idc-motus_db_versioned-3.1.0") + assert list(result.values()) == ["d0", "d1"] + + +def test_import_dry_run_and_idempotency(tmp_path, capsys): + inv = tmp_path / "inv.json" + inv.write_text('{"outputs":{"samestr_db_bundle":{"id":"dsSAM","src":"hda"}}}') + args = [ + "--galaxy-url", "https://test.galaxyproject.org", + "--invocation-json", str(inv), + "--dm", "samestr_db", "--version", "v1", + "--cvmfs-root", str(tmp_path), + ] + # dry-run imports nothing and writes no marker + assert imp.main(args + ["--dry-run"]) == 0 + assert not imp.record_marker(str(tmp_path), "samestr_db", "v1").exists() + + # write the marker directly, then a real run must skip (idempotent) + marker = imp.record_marker(str(tmp_path), "samestr_db", "v1") + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text("done\n") + assert imp.main(args) == 0 + assert "skipping" in capsys.readouterr().out