diff --git a/src/g3dt/cli/delete_cmds.py b/src/g3dt/cli/delete_cmds.py index aef43b6..db76dd1 100644 --- a/src/g3dt/cli/delete_cmds.py +++ b/src/g3dt/cli/delete_cmds.py @@ -56,16 +56,82 @@ def _normalise_version(raw: str, where: str) -> str: return match.group(1) +def _parse_study_specs(studies: str, fallback, env: str): + """Turn ``--studies`` into ``[(resolved_study_key, version), ...]``. + + Each comma-separated entry is ``name`` or ``name:version``. A bare name + takes *fallback* (the ``--version`` default); *fallback* is ``None`` when + ``--version`` was not given, which makes a bare name a usage error. + + Every entry is validated before anything is dispatched, so a typo in the + last study cannot leave the earlier ones already deleted. + """ + specs = [] + for entry in studies.split(","): + entry = entry.strip() + if not entry: + continue + + # partition() rather than split(), so a trailing colon ("ausdiab:") is + # distinguishable from a bare name and can be rejected instead of + # silently taking the fallback. + name, sep, raw_version = entry.partition(":") + name = name.strip() + + if not name: + typer.secho( + f"Invalid --studies entry '{entry}': missing study name.", + fg=typer.colors.RED, + err=True, + ) + raise typer.Exit(2) + + if sep and not raw_version.strip(): + typer.secho( + f"Invalid --studies entry '{entry}': ':' with no version. " + f"Use '{name}:0.9.8', '{name}:all', or a bare '{name}' to take " + "the --version default.", + fg=typer.colors.RED, + err=True, + ) + raise typer.Exit(2) + + if sep: + version = _normalise_version(raw_version, f"for study '{name}'") + elif fallback is not None: + version = fallback + else: + typer.secho( + f"No version for study '{name}': add ':' to it " + f"(e.g. '{name}:0.9.8'), or pass --version as the default for " + "every study. Use 'all' to delete every version.", + fg=typer.colors.RED, + err=True, + ) + raise typer.Exit(2) + + specs.append((study_of(name, env).key, version)) + + if not specs: + typer.secho("--studies is empty.", fg=typer.colors.RED, err=True) + raise typer.Exit(2) + return specs + + @app.command() def metadata( studies: str = typer.Option( - ..., "--studies", help="Comma-separated studies, e.g. ausdiab,caughtcad." + ..., + "--studies", + help="Comma-separated studies, each optionally 'name:version', " + "e.g. ausdiab:0.7.5,cdah:0.8.1,edcad.", ), env: str = typer.Option(..., "--env", "-e", help="Environment, e.g. test."), version: str = typer.Option( None, "--version", - help="Metadata version to delete, e.g. 0.9.8, or 'all' for every version.", + help="Default version for studies written without their own " + "':version', e.g. 0.9.8, or 'all' for every version.", ), node: str = typer.Option(None, "--node", help="Delete only this node type."), yes: bool = typer.Option( @@ -77,55 +143,65 @@ def metadata( Studies are processed one at a time. A study that exists but has no data at the requested version is skipped, and the job continues to the next study. - """ - if version is None: - typer.secho( - "--version is required: specify a version (e.g. 0.9.8) or 'all' " - "to delete every version.", - fg=typer.colors.RED, - err=True, - ) - raise typer.Exit(2) - version = _normalise_version(version, "for --version") + Each study may carry its own version as ``name:version``; ``--version`` + supplies the default for any study written bare. Examples: - names = [s.strip() for s in studies.split(",") if s.strip()] - keys = [study_of(name, env).key for name in names] - target = ",".join(keys) - all_versions = version == "all" - - if all_versions: - # Deleting every version is the most destructive path: always prompt - # (pass assume_yes=False so --yes can't bypass it; prod still types the - # target). - safety.confirm_destructive("deletion of ALL VERSIONS", target, env, False) + g3dt delete metadata --studies "ausdiab:0.7.5,cdah:0.8.1" --env staging + g3dt delete metadata --studies "ausdiab:all,cdah" --version 0.9.8 --env staging + """ + fallback = ( + _normalise_version(version, "for --version") if version is not None else None + ) + specs = _parse_study_specs(studies, fallback, env) + versions = [v for _, v in specs] + + # The typed production confirmation stays the study keys alone: short + # enough to retype accurately, while the per-study versions are spelled + # out in the action line printed directly above the prompt. + target = ",".join(key for key, _ in specs) + uniform = len(set(versions)) == 1 + any_all = "all" in versions + + if uniform and versions[0] == "all": + action = "deletion of ALL VERSIONS" + elif uniform: + action = f"deletion of v{versions[0]}" else: - safety.confirm_destructive(f"deletion of v{version}", target, env, yes) + plan = ", ".join(f"{key}:{v}" for key, v in specs) + action = f"deletion of per-study versions [{plan}]" + + # Deleting every version is the most destructive path: always prompt (pass + # assume_yes=False so --yes can't bypass it; prod still types the target). + # One 'all' anywhere in the list is enough to force the prompt, so an 'all' + # buried mid-list cannot ride along on a batch marked unattended. + safety.confirm_destructive(action, target, env, False if any_all else yes) def build_args(env_name): - a = [ - "--studies", - target, - "--env", - env_name, - "--version", - "all" if all_versions else version, - ] + if uniform: + # Canonical (and historical) shape: one --version for every study. + # Emitting it keeps a newer CLI compatible with an older installed + # service script on the box, which can lag a pip upgrade. + a = ["--studies", target, "--env", env_name, "--version", versions[0]] + else: + a = [ + "--studies", + ",".join(f"{key}:{v}" for key, v in specs), + "--env", + env_name, + ] if node: a += ["--node", node] return a def remote_cli(env_name): # --yes: confirmation already happened locally; the remote job must - # not prompt (SSM has no TTY). The remote re-check is version-specific - # only, and 'all' was already confirmed above. - a = [ - "delete", "metadata", - "--studies", studies, - "--env", env_name, - "--version", "all" if all_versions else version, - "--yes", - ] + # not prompt (SSM has no TTY). The raw --studies string is forwarded + # verbatim — the remote re-entry re-parses and re-validates it. + a = ["delete", "metadata", "--studies", studies, "--env", env_name] + if version is not None: + a += ["--version", version] + a.append("--yes") if node: a += ["--node", node] return a diff --git a/src/g3dt/cli/metadata.py b/src/g3dt/cli/metadata.py index 46cd655..a0b8381 100644 --- a/src/g3dt/cli/metadata.py +++ b/src/g3dt/cli/metadata.py @@ -22,10 +22,19 @@ def upload( study: str = typer.Option(..., "--study", "-s", help="Study, e.g. ausdiab."), env: str = typer.Option(..., "--env", "-e", help="Environment, e.g. test."), node: str = typer.Option(None, "--node", help="Submit only this node type."), + force_reupload: bool = typer.Option( + False, "--force-reupload", + help="Proceed even if this project+version was already uploaded to " + "this commons (uploads are additive: re-running duplicates records).", + ), on: Target = typer.Option(Target.local, "--on", help="Run local or on ec2."), ) -> None: """Upload a study's release metadata to Gen3 sheepdog. + The worker refuses (exit 2) when the audit table already records an + upload of the same project + version + endpoint — re-running would + duplicate every record. ``--force-reupload`` overrides. + Examples: g3dt metadata upload --study ausdiab --env staging g3dt metadata upload --study ausdiab --env staging --on ec2 @@ -36,12 +45,16 @@ def build_args(env_name): a = ["--study", s.key, "--env", env_name] if node: a += ["--specific-node", node] + if force_reupload: + a.append("--force-reupload") return a def remote_cli(env_name): a = ["metadata", "upload", "--study", study, "--env", env_name] if node: a += ["--node", node] + if force_reupload: + a.append("--force-reupload") return a dispatch.run_or_dispatch( @@ -64,6 +77,11 @@ def upload_all( help="Internal: set by the remote re-entry after the typed " "confirmation already happened locally. Never pass by hand.", ), + force_reupload: bool = typer.Option( + False, "--force-reupload", + help="Proceed even for project+versions the audit table says were " + "already uploaded to this commons.", + ), on: Target = typer.Option(Target.local, "--on", help="Run local or on ec2."), ) -> None: """Upload several studies sequentially (wraps upload_all_studies.sh). @@ -95,6 +113,8 @@ def build_args(env_name): a = ["--studies", ",".join(keys), "--env", env_name] if allow_prod: a.append("--allow-prod") + if force_reupload: + a.append("--force-reupload") return a def remote_cli(env_name): @@ -103,6 +123,8 @@ def remote_cli(env_name): # The typed confirmation already happened locally above; the box # has no TTY, so the re-entry must not prompt again. a += ["--allow-prod", "--prod-confirmed"] + if force_reupload: + a.append("--force-reupload") return a dispatch.run_or_dispatch( diff --git a/src/g3dt/services/delete/delete_metadata.sh b/src/g3dt/services/delete/delete_metadata.sh index 730a304..60e5f86 100755 --- a/src/g3dt/services/delete/delete_metadata.sh +++ b/src/g3dt/services/delete/delete_metadata.sh @@ -10,19 +10,21 @@ SKIP_EXIT_CODE=3 usage() { cat < --env --version [--node ] +Usage: $(basename "$0") --studies --env [--version ] [--node ] Delete metadata for each study sequentially, in a single job. Arguments: - --studies Comma-separated list of study config keys (e.g. ausdiab_staging,caughtcad_staging) + --studies Comma-separated study config keys, each optionally qualified with + its own version (e.g. ausdiab_staging:0.7.5,cdah_staging:0.8.1, + or bare ausdiab_staging to take the --version default) --env Environment string passed to the Python worker (e.g. staging_ec2) - --version Metadata version to delete (e.g. 0.9.8), or 'all' for every version + --version Default version for bare --studies entries (e.g. 0.9.8), or 'all' --node (optional) Restrict deletion to a single node type Behaviour: - * --version all -> delete_all_metadata_for_project.py (deletes whole nodes) - * --version -> delete_metadata_by_guid.py (Athena GUID lookup for that version) + * version 'all' -> delete_all_metadata_for_project.py (deletes whole nodes) + * version -> delete_metadata_by_guid.py (Athena GUID lookup for that version) A study that exists but has no data at the requested version is skipped and the loop continues. Only genuine errors (Gen3/AWS failures) count as failures. @@ -69,13 +71,39 @@ while [[ $# -gt 0 ]]; do esac done -if [[ -z "$STUDIES" || -z "$ENV" || -z "$VERSION" ]]; then - echo "ERROR: --studies, --env and --version are required." +if [[ -z "$STUDIES" || -z "$ENV" ]]; then + echo "ERROR: --studies and --env are required." usage fi -# Lower-case the version so 'ALL'/'All' are treated as 'all'. -VERSION_LC="$(echo "$VERSION" | tr '[:upper:]' '[:lower:]')" +# Expand '--studies name[:version],...' into two parallel arrays. An entry with +# no ':version' takes the --version default. Validating the whole list up front +# means a typo in the last entry cannot leave the earlier studies already +# deleted. +IFS=',' read -ra STUDY_ENTRIES <<< "$STUDIES" +STUDY_NAMES=() +STUDY_VERSIONS=() +for entry in "${STUDY_ENTRIES[@]}"; do + name="${entry%%:*}" + # Test for the ':' explicitly: for a bare 'name', "${entry#*:}" expands to + # 'name' rather than to the empty string, which would silently become the + # version. + if [[ "$entry" == *:* ]]; then + entry_version="${entry#*:}" + else + entry_version="$VERSION" + fi + if [[ -z "$name" ]]; then + echo "ERROR: empty study name in --studies entry '${entry}'." + usage + fi + if [[ -z "$entry_version" ]]; then + echo "ERROR: study '${name}' has no version: use '${name}:' in --studies, or pass --version as the default." + usage + fi + STUDY_NAMES+=("$name") + STUDY_VERSIONS+=("$entry_version") +done # ---------- Setup ---------- # Logs go outside the installed package. @@ -84,7 +112,6 @@ TIMESTAMP="$(date +%Y%m%d_%H%M%S)" FAILED_LOG="${LOG_DIR}/${TIMESTAMP}_delete_failed.log" mkdir -p "${LOG_DIR}" -IFS=',' read -ra STUDY_LIST <<< "$STUDIES" DELETED_COUNT=0 SKIPPED_COUNT=0 FAIL_COUNT=0 @@ -93,25 +120,30 @@ echo "============================================" echo "Metadata delete started at $(date)" echo "Environment : ${ENV}" echo "Studies : ${STUDIES}" -echo "Version : ${VERSION}" +echo "Version : ${VERSION:-(per study, from --studies)}" [[ -n "$NODE" ]] && echo "Node : ${NODE}" echo "Failure log : ${FAILED_LOG}" echo "============================================" echo "" # ---------- Sequential execution ---------- -for study in "${STUDY_LIST[@]}"; do +for i in "${!STUDY_NAMES[@]}"; do + study="${STUDY_NAMES[$i]}" + study_version="${STUDY_VERSIONS[$i]}" + # Lower-cased for the 'all' comparison only; the worker gets the original. + study_version_lc="$(echo "$study_version" | tr '[:upper:]' '[:lower:]')" + echo "--------------------------------------------" - echo "[$(date +%Y-%m-%d\ %H:%M:%S)] Starting deletion for study: ${study} (version: ${VERSION})" + echo "[$(date +%Y-%m-%d\ %H:%M:%S)] Starting deletion for study: ${study} (version: ${study_version})" echo "--------------------------------------------" - if [[ "$VERSION_LC" == "all" ]]; then + if [[ "$study_version_lc" == "all" ]]; then CMD=(python3 "${SCRIPT_DIR}/delete_all_metadata_for_project.py" --study "$study" --env "$ENV") [[ -n "$NODE" ]] && CMD+=(--node "$NODE") else CMD=(python3 "${SCRIPT_DIR}/delete_metadata_by_guid.py" - --study "$study" --env "$ENV" --version "$VERSION" --skip-if-empty) + --study "$study" --env "$ENV" --version "$study_version" --skip-if-empty) [[ -n "$NODE" ]] && CMD+=(--node "$NODE") fi @@ -125,12 +157,14 @@ for study in "${STUDY_LIST[@]}"; do echo "[$(date +%Y-%m-%d\ %H:%M:%S)] Completed: ${study}" DELETED_COUNT=$((DELETED_COUNT + 1)) elif [[ $EXIT_CODE -eq $SKIP_EXIT_CODE ]]; then - echo "[$(date +%Y-%m-%d\ %H:%M:%S)] Skipped (no data at version ${VERSION}): ${study}" + echo "[$(date +%Y-%m-%d\ %H:%M:%S)] Skipped (no data at version ${study_version}): ${study}" SKIPPED_COUNT=$((SKIPPED_COUNT + 1)) else FAIL_COUNT=$((FAIL_COUNT + 1)) echo "[$(date +%Y-%m-%d\ %H:%M:%S)] FAILED: ${study} (exit code ${EXIT_CODE})" - echo "[$(date +%Y-%m-%d\ %H:%M:%S)] ${study} exit_code=${EXIT_CODE}" >> "$FAILED_LOG" + # The version is recorded because one job can now delete two versions + # of the same study. + echo "[$(date +%Y-%m-%d\ %H:%M:%S)] ${study} version=${study_version} exit_code=${EXIT_CODE}" >> "$FAILED_LOG" fi echo "" @@ -139,7 +173,7 @@ done # ---------- Summary ---------- echo "============================================" echo "Metadata delete finished at $(date)" -echo "Total studies : ${#STUDY_LIST[@]}" +echo "Total studies : ${#STUDY_NAMES[@]}" echo "Deleted : ${DELETED_COUNT}" echo "Skipped : ${SKIPPED_COUNT}" echo "Failures : ${FAIL_COUNT}" diff --git a/src/g3dt/services/upload/metadata/upload_all_studies.sh b/src/g3dt/services/upload/metadata/upload_all_studies.sh index 475da11..faac071 100755 --- a/src/g3dt/services/upload/metadata/upload_all_studies.sh +++ b/src/g3dt/services/upload/metadata/upload_all_studies.sh @@ -12,10 +12,12 @@ Run upload_metadata.py sequentially for each study. Arguments: --studies Comma-separated list of study config keys (e.g. ausdiab_staging,caughtcad_staging) --env Environment string passed to the Python script (e.g. staging_ec2) - --allow-prod Permit running against a production environment. Without it any - 'prod' in --env or --studies aborts. The g3dt CLI adds this flag - only after a local typed confirmation — prefer invoking via - 'g3dt metadata upload-all' rather than passing it by hand. + --allow-prod Permit running against a production environment. Without it + any 'prod' in --env or --studies aborts. The g3dt CLI adds + this flag only after a local typed confirmation — prefer + invoking via 'g3dt metadata upload-all'. + --force-reupload Passed through to upload_metadata.py: proceed even when the + audit table already records this project+version+endpoint. Run via the g3dt CLI: g3dt metadata upload-all \\ @@ -31,6 +33,7 @@ EOF STUDIES="" ENV="" ALLOW_PROD="false" +FORCE_REUPLOAD="false" while [[ $# -gt 0 ]]; do case "$1" in @@ -46,6 +49,10 @@ while [[ $# -gt 0 ]]; do ALLOW_PROD="true" shift ;; + --force-reupload) + FORCE_REUPLOAD="true" + shift + ;; *) echo "ERROR: Unknown argument: $1" usage @@ -93,8 +100,9 @@ for study in "${STUDY_LIST[@]}"; do echo "[$(date +%Y-%m-%d\ %H:%M:%S)] Starting upload for study: ${study}" echo "--------------------------------------------" - if python3 "${SCRIPT_DIR}/upload_metadata.py" \ - --study "$study" --env "$ENV"; then + UPLOAD_ARGS=(--study "$study" --env "$ENV") + [[ "$FORCE_REUPLOAD" == "true" ]] && UPLOAD_ARGS+=(--force-reupload) + if python3 "${SCRIPT_DIR}/upload_metadata.py" "${UPLOAD_ARGS[@]}"; then echo "[$(date +%Y-%m-%d\ %H:%M:%S)] Completed successfully: ${study}" else EXIT_CODE=$? diff --git a/src/g3dt/services/upload/metadata/upload_metadata.py b/src/g3dt/services/upload/metadata/upload_metadata.py index 06ea20f..4938f19 100644 --- a/src/g3dt/services/upload/metadata/upload_metadata.py +++ b/src/g3dt/services/upload/metadata/upload_metadata.py @@ -8,6 +8,8 @@ find_data_import_order_file_s3, create_boto3_session, get_gen3_api_key_aws_secret, + fetch_prior_uploads, + infer_api_endpoint_from_jwt, MetadataSubmitter, ) @@ -39,6 +41,12 @@ def main(): help="Environment to upload to (e.g., test, staging, prod, staging_ec2, prod_ec2)" ) parser.add_argument("--specific-node", help="Submit only a specific node") + parser.add_argument( + "--force-reupload", action="store_true", + help="Proceed even if the audit table already holds rows for this " + "project + version + endpoint (uploads are additive: re-running " + "duplicates the records)", + ) parser.add_argument( "--debug", action="store_true", help="Enable debug logging" @@ -137,6 +145,32 @@ def main(): aws_region=aws_region, ) + # Pre-flight: uploads are additive, so re-running one silently doubles + # the project's records at that version. Refuse when the audit table + # already holds rows for this project + version + endpoint. + if not args.force_reupload: + version = submitter._collect_versions_from_metadata_file_list() + api_endpoint = infer_api_endpoint_from_jwt(api_key["api_key"]) + prior = fetch_prior_uploads( + database=database, + table=table, + project_id=project_id, + version=version, + api_endpoint=api_endpoint, + athena_s3_output=athena_s3_output, + workgroup=workgroup, + boto3_session=session, + ) + if not prior.empty: + logger.error( + "Refusing to upload: the audit table %s.%s already records an " + "upload of project '%s' version %s to %s. Re-running would " + "duplicate every record. Bump the release version, or pass " + "--force-reupload if the duplication is intended.", + database, table, project_id, version, api_endpoint, + ) + sys.exit(2) + try: logger.info("Submitting metadata to Gen3.") submitter.submit_metadata(specific_node=args.specific_node) diff --git a/src/g3dt/upload/metadata_submitter.py b/src/g3dt/upload/metadata_submitter.py index 15cedaa..e1a4ff5 100644 --- a/src/g3dt/upload/metadata_submitter.py +++ b/src/g3dt/upload/metadata_submitter.py @@ -12,6 +12,7 @@ import requests from typing import Any, Dict, List, Optional import re +import awswrangler as wr import pandas as pd import uuid from g3dt.utils.athena_utils import write_iceberg_to_db @@ -475,6 +476,56 @@ def get_gen3_api_key_aws_secret(secret_name: str, region_name: str, session) -> raise +def fetch_prior_uploads( + database: str, + table: str, + project_id: str, + version: str, + api_endpoint: str, + athena_s3_output: str, + workgroup: str = "primary", + boto3_session=None, +) -> pd.DataFrame: + """Return audit rows already recorded for this project+version+endpoint. + + ``metadata upload`` is purely additive: re-running it doubles the + project's records for that version with nothing to flag it — observed + live as an audit table holding exactly 2x the per-run row count for + several study/version pairs. Callers use this pre-flight to refuse (or + warn on) an upload the audit table says has already happened. + + Scoped to one api_endpoint so a staging upload can never mask — or + spuriously block — a prod one. + + A missing table means nothing has ever been uploaded in this + environment, which is a normal first-run state, not an error — an empty + frame is returned so the caller proceeds. + """ + sql = ( + f'SELECT DISTINCT project_id, version ' + f'FROM "{database}"."{table}" ' + f"WHERE project_id = '{project_id}' " + f"AND version = '{version}' " + f"AND api_endpoint = '{api_endpoint}'" + ) + try: + return wr.athena.read_sql_query( + sql, + database=database, + ctas_approach=False, + workgroup=workgroup, + s3_output=athena_s3_output, + boto3_session=boto3_session, + ) + except Exception as exc: + logger.warning( + "Could not read prior uploads from %s.%s (%s). " + "Treating this as a first upload.", + database, table, exc, + ) + return pd.DataFrame(columns=["project_id", "version"]) + + def commons_url_from_jwt(jwt_token: str) -> str: """Derive the bare commons base URL from an API key's JWT. diff --git a/tests/test_cli_commands.py b/tests/test_cli_commands.py index 688a0be..7ce3481 100644 --- a/tests/test_cli_commands.py +++ b/tests/test_cli_commands.py @@ -175,6 +175,65 @@ def test_metadata_upload_all_resolves_each_study(mock_run, _study, _env): assert argv[i + 1] == "ausdiab_staging,caughtcad_staging" +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_metadata_upload_forwards_force_reupload(mock_run, _study, _env): + """ + Background: + The duplicate-upload pre-flight lives in the worker + (upload_metadata.py exits 2 when the audit table already records + this project+version+endpoint). The CLI's only job is to thread the + override flag through — if it drops it, the worker can never be + overridden from the CLI. + + Inputs: g3dt metadata upload --study ausdiab --env staging --force-reupload + Expected Output: the worker argv contains --force-reupload. + """ + result = runner.invoke( + app, + ["metadata", "upload", "--study", "ausdiab", "--env", "staging", + "--force-reupload"], + ) + assert result.exit_code == 0, result.output + assert "--force-reupload" in _argv(mock_run) + + +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_metadata_upload_omits_force_reupload_by_default(mock_run, _study, _env): + """ + Inputs: g3dt metadata upload without the flag + Expected Output: the worker argv does NOT contain --force-reupload, so + the pre-flight guard stays active. + """ + result = runner.invoke( + app, + ["metadata", "upload", "--study", "ausdiab", "--env", "staging"], + ) + assert result.exit_code == 0, result.output + assert "--force-reupload" not in _argv(mock_run) + + +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_metadata_upload_all_forwards_force_reupload(mock_run, _study, _env): + """ + Inputs: g3dt metadata upload-all ... --force-reupload + Expected Output: the bulk wrapper argv carries --force-reupload (it + passes it through to each per-study worker). + """ + result = runner.invoke( + app, + ["metadata", "upload-all", "--studies", "ausdiab,caughtcad", + "--env", "staging", "--force-reupload"], + ) + assert result.exit_code == 0, result.output + assert "--force-reupload" in _argv(mock_run) + + @patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) @patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg) @patch("g3dt.cli._internal.runner.run") @@ -318,6 +377,134 @@ def test_delete_metadata_all_versions_passes_all(mock_run, _study, _env): assert argv[j + 1] == "all" +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_delete_metadata_per_study_versions_build_qualified_spec( + mock_run, _study, _env +): + """ + Background: + Release ladders diverge per study in practice (one study retiring + 0.7.5 while another retires 0.8.1), so one job must be able to carry + a different version per study instead of one job per study. + + Inputs: --studies "ausdiab:0.7.5,caughtcad:0.8.1" (no --version) + Expected Output: the wrapper receives the qualified key:version list and + NO --version flag. + """ + result = runner.invoke( + app, + ["delete", "metadata", "--studies", "ausdiab:0.7.5,caughtcad:0.8.1", + "--env", "staging", "--yes"], + ) + assert result.exit_code == 0, result.output + argv = _argv(mock_run) + i = argv.index("--studies") + assert argv[i + 1] == "ausdiab_staging:0.7.5,caughtcad_staging:0.8.1" + assert "--version" not in argv + + +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_delete_metadata_uniform_versions_collapse_to_legacy_argv( + mock_run, _study, _env +): + """ + Background: + When every study lands on the same version — every invocation that + existed before per-study specs — the wrapper must receive the + historical `--studies a,b --version X` shape, byte-identical, so a + newer CLI stays compatible with an older installed service script. + + Inputs: --studies "ausdiab:0.9.8,caughtcad" --version 0.9.8 + Expected Output: plain key list + one --version 0.9.8, no colons. + """ + result = runner.invoke( + app, + ["delete", "metadata", "--studies", "ausdiab:0.9.8,caughtcad", + "--env", "staging", "--version", "0.9.8", "--yes"], + ) + assert result.exit_code == 0, result.output + argv = _argv(mock_run) + i = argv.index("--studies") + assert argv[i + 1] == "ausdiab_staging,caughtcad_staging" + assert argv[argv.index("--version") + 1] == "0.9.8" + + +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_delete_metadata_bare_study_without_version_is_usage_error( + mock_run, _study, _env +): + """ + Background: + The whole list is validated before anything is dispatched — a typo in + the last study must not leave the earlier ones already deleted. + + Inputs: --studies "ausdiab:0.7.5,caughtcad" with NO --version + Expected Output: exit 2 naming the bare study, nothing dispatched. + """ + result = runner.invoke( + app, + ["delete", "metadata", "--studies", "ausdiab:0.7.5,caughtcad", + "--env", "staging", "--yes"], + ) + assert result.exit_code == 2 + mock_run.assert_not_called() + assert "caughtcad" in result.output + + +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_delete_metadata_rejects_colon_with_no_version(mock_run, _study, _env): + """ + Background: + A trailing colon is a half-finished edit, not a request for the + default. Silently falling back to --version would delete a version + the operator did not name — partition(':') makes the two cases + distinguishable. + + Inputs: --studies "ausdiab:" --version 0.9.8 + Expected Output: exit 2, nothing dispatched. + """ + result = runner.invoke( + app, + ["delete", "metadata", "--studies", "ausdiab:", + "--env", "staging", "--version", "0.9.8", "--yes"], + ) + assert result.exit_code == 2 + mock_run.assert_not_called() + + +@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) +@patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg) +@patch("g3dt.cli._internal.runner.run") +def test_delete_metadata_mixed_all_and_specific_always_prompts( + mock_run, _study, _env +): + """ + Background: + Deleting ALL versions is the most destructive path and always + prompts, even with --yes. One 'all' hidden mid-list must force the + same prompt — otherwise it rides along on a batch marked unattended. + + Inputs: --studies "ausdiab:0.9.8,caughtcad:all" --yes, prompt declined + Expected Output: non-zero exit, nothing dispatched. + """ + result = runner.invoke( + app, + ["delete", "metadata", "--studies", "ausdiab:0.9.8,caughtcad:all", + "--env", "staging", "--yes"], + input="n\n", + ) + assert result.exit_code != 0 + mock_run.assert_not_called() + + @patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg) @patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg) @patch("g3dt.cli._internal.runner.run") diff --git a/tests/test_delete_metadata_sh.py b/tests/test_delete_metadata_sh.py new file mode 100644 index 0000000..fd36ba5 --- /dev/null +++ b/tests/test_delete_metadata_sh.py @@ -0,0 +1,193 @@ +"""Routing tests for the bulk metadata-delete shell script. + +``services/delete/delete_metadata.sh`` is the layer that fans one job out over +several studies, and since each study may carry its own version, it is also +the layer that decides *which worker* each study goes to: a specific version +uses the Athena GUID lookup, ``all`` wipes whole nodes. Getting that branch +wrong would delete far more than intended, and it is not reachable from the +Python tests — the CLI stops at building argv. + +So these tests run the real script with a stubbed ``python3`` on PATH that +records the command line it was handed and exits with a chosen code. That lets +us assert the routing, the version each worker receives, and the +deleted/skipped/failed accounting without touching Gen3 or AWS. +""" +import os +import subprocess +from pathlib import Path + +import pytest + +SCRIPT = ( + Path(__file__).resolve().parent.parent + / "src" / "g3dt" / "services" / "delete" / "delete_metadata.sh" +) + + +@pytest.fixture +def stub_python(tmp_path): + """Put a fake ``python3`` on PATH that logs its arguments. + + The stub exits 0 normally, 3 for any study whose name contains ``skipme`` + (the worker's "no data at this version" code), and 4 for ``failme`` (a + genuine error). Returning the record file lets a test assert exactly which + worker each study was routed to. + """ + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + record = tmp_path / "record.txt" + stub = bin_dir / "python3" + stub.write_text( + "#!/usr/bin/env bash\n" + 'echo "$*" >> "$STUB_RECORD"\n' + 'case "$*" in\n' + " *skipme*) exit 3 ;;\n" + " *failme*) exit 4 ;;\n" + " *) exit 0 ;;\n" + "esac\n" + ) + stub.chmod(0o755) + + env = dict(os.environ) + env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}" + env["STUB_RECORD"] = str(record) + # Keep the script's failure log inside the tmpdir, not the real ~/.g3dt. + env["HOME"] = str(tmp_path) + return env, record + + +def _run(env, *args): + return subprocess.run( + ["bash", str(SCRIPT), *args], + capture_output=True, + text=True, + env=env, + ) + + +def test_each_study_is_routed_to_its_own_worker_and_version(stub_python): + """ + Background: + The whole point of per-study versions is that one job can mix paths — + retire one study entirely while removing a single version from + another. The script picks the worker per study, so this asserts the + branch is inside the loop rather than decided once for the batch. + + Inputs: --studies "ausdiab_staging:0.7.5,cdah_staging:all" + Expected Output: + - ausdiab goes to delete_metadata_by_guid.py with --version 0.7.5 + - cdah goes to delete_all_metadata_for_project.py with no --version + """ + env, record = stub_python + + _run(env, "--studies", "ausdiab_staging:0.7.5,cdah_staging:all", + "--env", "staging") + + calls = record.read_text().strip().splitlines() + assert len(calls) == 2 + assert "delete_metadata_by_guid.py" in calls[0] + assert "--study ausdiab_staging" in calls[0] + assert "--version 0.7.5" in calls[0] + assert "delete_all_metadata_for_project.py" in calls[1] + assert "--study cdah_staging" in calls[1] + assert "--version" not in calls[1] + + +def test_bare_entries_still_take_the_version_flag(stub_python): + """ + Background: + The EC2 box's installed script can lag or lead the operator's local + CLI, so the historical argv shape — plain study keys plus one + --version — must keep working unchanged. + + Inputs: --studies "a_staging,b_staging" --version 0.9.8 + Expected Output: both studies routed to the GUID worker with 0.9.8. + """ + env, record = stub_python + + _run(env, "--studies", "a_staging,b_staging", + "--env", "staging", "--version", "0.9.8") + + calls = record.read_text().strip().splitlines() + assert len(calls) == 2 + for call in calls: + assert "delete_metadata_by_guid.py" in call + assert "--version 0.9.8" in call + + +def test_mixed_bare_and_qualified_entries(stub_python): + """ + Inputs: --studies "a_staging:0.7.5,b_staging" --version 1.0.0 + Expected Output: a gets its own 0.7.5; bare b falls back to 1.0.0. + """ + env, record = stub_python + + _run(env, "--studies", "a_staging:0.7.5,b_staging", + "--env", "staging", "--version", "1.0.0") + + calls = record.read_text().strip().splitlines() + assert "--version 0.7.5" in calls[0] + assert "--version 1.0.0" in calls[1] + + +def test_missing_version_aborts_before_anything_is_deleted(stub_python): + """ + Background: + Validation is all-up-front: if the SECOND entry lacks a version, the + first must not already have been dispatched — otherwise a typo'd batch + is half-executed. + + Inputs: --studies "a_staging:0.7.5,b_staging" with NO --version + Expected Output: non-zero exit, and no worker was ever invoked. + """ + env, record = stub_python + + result = _run(env, "--studies", "a_staging:0.7.5,b_staging", + "--env", "staging") + + assert result.returncode != 0 + assert not record.exists() + + +def test_skip_exit_code_counts_as_skip_not_failure(stub_python): + """ + Background: + Worker exit 3 means "study has no data at this version" — the normal, + healthy outcome for an already-clean study. The batch must continue + and finish with exit 0, counting it as skipped rather than failed. + + Inputs: one skipping study and one deleting study + Expected Output: overall exit 0; summary shows 1 deleted / 1 skipped. + """ + env, _ = stub_python + + result = _run(env, "--studies", "skipme_staging:0.9.8,ok_staging:0.9.8", + "--env", "staging") + + assert result.returncode == 0 + assert "Deleted : 1" in result.stdout + assert "Skipped : 1" in result.stdout + + +def test_one_failure_fails_the_batch_and_logs_the_version(stub_python): + """ + Background: + A genuine worker error must fail the whole run (exit 1) — and the + failure log must record WHICH version failed, because one job can now + delete two versions of the same study. + + Inputs: one failing study (worker exit 4) and one succeeding + Expected Output: exit 1; the log line carries version=0.7.5; the other + study still ran. + """ + env, record = stub_python + + result = _run(env, "--studies", "failme_staging:0.7.5,ok_staging:0.9.8", + "--env", "staging") + + assert result.returncode == 1 + calls = record.read_text().strip().splitlines() + assert len(calls) == 2 # the failure did not stop the loop + logs = list((Path(env["HOME"]) / ".g3dt" / "logs").glob("*_delete_failed.log")) + assert len(logs) == 1 + assert "failme_staging version=0.7.5" in logs[0].read_text() diff --git a/tests/test_metadata_submitter.py b/tests/test_metadata_submitter.py index ad02a29..759cbc6 100644 --- a/tests/test_metadata_submitter.py +++ b/tests/test_metadata_submitter.py @@ -448,3 +448,70 @@ def test_infer_api_endpoint_from_jwt_appends_api_version(): assert infer_api_endpoint_from_jwt( _fake_jwt("https://staging.commons.example.org/user") ) == "https://staging.commons.example.org/api/v0" + + +def test_fetch_prior_uploads_scopes_to_project_version_and_endpoint(): + """ + Test that the duplicate-upload pre-flight query cannot mix environments. + + Background: + metadata upload is purely additive: re-running one silently doubles + the project's records at that version (observed live as an audit + table holding exactly 2x the per-run row count). The pre-flight asks + the audit table "has this exact upload happened before?" — and it + must scope by api_endpoint, or a staging upload would spuriously + block (or mask) a prod one. + + Inputs: project 'EDCAD-PMS', version 0.9.8, a specific endpoint + Expected Output: all three appear in the WHERE clause. + """ + from unittest.mock import MagicMock, patch + + from g3dt.upload.metadata_submitter import fetch_prior_uploads + + with patch( + "g3dt.upload.metadata_submitter.wr.athena.read_sql_query", + return_value=MagicMock(), + ) as mock_query: + fetch_prior_uploads( + database="db", table="metadata_upload_iceberg", + project_id="EDCAD-PMS", version="0.9.8", + api_endpoint="https://commons.example.org/api/v0", + athena_s3_output="s3://out/", + ) + + sql = mock_query.call_args[0][0] + assert "project_id = 'EDCAD-PMS'" in sql + assert "version = '0.9.8'" in sql + assert "api_endpoint = 'https://commons.example.org/api/v0'" in sql + + +def test_fetch_prior_uploads_treats_a_missing_table_as_first_upload(): + """ + Test that a brand-new environment does not fail the pre-flight. + + Background: + A fresh environment has no metadata_upload table yet. That is the + normal first-run state — the pre-flight must return "no prior + uploads" and let the upload proceed, not abort on the Athena error. + + Inputs: an Athena query that raises (table does not exist) + Expected Output: an empty DataFrame with the expected columns, no raise. + """ + from unittest.mock import patch + + from g3dt.upload.metadata_submitter import fetch_prior_uploads + + with patch( + "g3dt.upload.metadata_submitter.wr.athena.read_sql_query", + side_effect=Exception("TABLE_NOT_FOUND"), + ): + result = fetch_prior_uploads( + database="db", table="metadata_upload_iceberg", + project_id="EDCAD-PMS", version="0.9.8", + api_endpoint="https://commons.example.org/api/v0", + athena_s3_output="s3://out/", + ) + + assert result.empty + assert list(result.columns) == ["project_id", "version"]