Add fail-safe guard and unit tests for App Engine cleanup

- cleanup_app_engine_versions.sh: abort before stopping if versions are
  SERVING but none report traffic, so a failed/empty traffic query cannot
  deactivate the live service. Delete already only targets STOPPED, so the
  live version can never be deleted.
- tests/unit/test_app_engine_cleanup_scripts.py: exercise both scripts with a
  stubbed gcloud (mirroring test_cloud_run_deploy_scripts.py) covering
  live/warm retention, protection of an older traffic-serving version,
  delete-only-stopped, the fail-safe guard, missing-staging robustness, and
  the stop script's required-version check.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: anth-volk

.github/scripts/cleanup_app_engine_versions.sh

@@ -73,6 +73,16 @@ while IFS= read -r v; do
   [[ -n "${v}" ]] && serving_all+=("${v}")
 done < <(read_versions "version.servingStatus=SERVING")
 
+# Fail safe: if there are SERVING versions but none report traffic, we cannot
+# identify the live service. Refuse to stop anything rather than risk
+# deactivating production (e.g. if the traffic query returns nothing). The
+# delete step never runs either, since we exit here.
+if [[ "${#serving_all[@]}" -gt 0 && "${#serving_traffic[@]}" -eq 0 ]]; then
+  echo "ERROR: ${#serving_all[@]} SERVING version(s) but none are receiving traffic;" >&2
+  echo "refusing to stop versions (cannot identify the live service). Aborting." >&2
+  exit 1
+fi
+
 # Newest KEEP_WARM_PROD prod-* versions to keep warm for instant rollback.
 warm_prod=()
 while IFS= read -r v; do

tests/unit/test_app_engine_cleanup_scripts.py

@@ -0,0 +1,257 @@
+"""Unit tests for the App Engine idle-version cleanup scripts.
+
+`cleanup_app_engine_versions.sh` and `stop_app_engine_version.sh` are exercised
+by running them under bash with a stubbed ``gcloud`` on ``PATH`` (mirroring the
+approach in ``test_cloud_run_deploy_scripts.py``). The stub returns canned
+``gcloud app versions list`` output based on the ``--filter`` and records any
+``stop``/``delete`` invocations, so the tests assert the selection logic without
+touching real infrastructure. All cleanup runs use ``DRY_RUN=1``.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[2]
+CLEANUP_SCRIPT = ".github/scripts/cleanup_app_engine_versions.sh"
+STOP_SCRIPT = ".github/scripts/stop_app_engine_version.sh"
+
+# A stub `gcloud` that answers `app versions list` from MOCK_* env vars (each a
+# newline-joined, newest-first id list) and appends any stop/delete calls to
+# $GCLOUD_CALLS so tests can assert nothing mutating ran under DRY_RUN.
+_FAKE_GCLOUD = """#!/usr/bin/env bash
+filter=""
+for a in "$@"; do
+  case "$a" in --filter=*) filter="${a#--filter=}";; esac
+done
+case "$*" in
+  *"versions list"*)
+    case "$filter" in
+      *"traffic_split>0"*) [ -n "${MOCK_TRAFFIC:-}" ] && printf '%s\\n' "$MOCK_TRAFFIC";;
+      *"servingStatus=SERVING"*) [ -n "${MOCK_SERVING:-}" ] && printf '%s\\n' "$MOCK_SERVING";;
+      *"servingStatus=STOPPED"*) [ -n "${MOCK_STOPPED:-}" ] && printf '%s\\n' "$MOCK_STOPPED";;
+    esac;;
+  *"versions stop"*)   echo "STOP $*" >> "$GCLOUD_CALLS";;
+  *"versions delete"*) echo "DELETE $*" >> "$GCLOUD_CALLS";;
+esac
+exit 0
+"""
+
+
+def _run(
+    script: str,
+    tmp_path: Path,
+    *,
+    serving: list[str] | None = None,
+    traffic: list[str] | None = None,
+    stopped: list[str] | None = None,
+    extra_env: dict[str, str] | None = None,
+) -> tuple[subprocess.CompletedProcess[str], Path]:
+    stub_dir = tmp_path / "bin"
+    stub_dir.mkdir(exist_ok=True)
+    gcloud = stub_dir / "gcloud"
+    gcloud.write_text(_FAKE_GCLOUD, encoding="utf-8")
+    gcloud.chmod(0o755)
+
+    calls = tmp_path / "gcloud_calls.log"
+    calls.write_text("", encoding="utf-8")
+
+    env = {
+        "HOME": os.environ.get("HOME", ""),
+        "PATH": f"{stub_dir}:{os.environ['PATH']}",
+        "DRY_RUN": "1",
+        "APP_ENGINE_PROJECT": "test-project",
+        "GCLOUD_CALLS": str(calls),
+        "MOCK_SERVING": "\n".join(serving or []),
+        "MOCK_TRAFFIC": "\n".join(traffic or []),
+        "MOCK_STOPPED": "\n".join(stopped or []),
+    }
+    env.update(extra_env or {})
+
+    result = subprocess.run(
+        ["bash", script],
+        cwd=REPO,
+        env=env,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    return result, calls
+
+
+def _list_after(label: str, stdout: str) -> list[str]:
+    """Return the space-separated ids printed after a `Label[ (n)]: ...` line."""
+    match = re.search(rf"{re.escape(label)}(?: \(\d+\))?:\s*(.*)", stdout)
+    assert match is not None, f"missing '{label}' line in:\n{stdout}"
+    body = match.group(1).strip()
+    return [] if body == "<none>" else body.split()
+
+
+# --- syntax --------------------------------------------------------------
+
+
+def test_cleanup_scripts_are_shell_syntax_valid():
+    for script in (CLEANUP_SCRIPT, STOP_SCRIPT):
+        result = subprocess.run(
+            ["bash", "-n", script], cwd=REPO, capture_output=True, text=True
+        )
+        assert result.returncode == 0, f"{script}: {result.stderr}"
+
+
+# --- cleanup: stop selection ---------------------------------------------
+
+
+def test_keeps_live_and_warm_prod_and_stops_the_rest(tmp_path):
+    result, calls = _run(
+        CLEANUP_SCRIPT,
+        tmp_path,
+        serving=["prod-110", "prod-108", "staging-110", "staging-108", "prod-106"],
+        traffic=["prod-110"],
+        stopped=["prod-104", "prod-102"],
+    )
+    assert result.returncode == 0, result.stderr
+    assert _list_after("Serving traffic (never touched)", result.stdout) == ["prod-110"]
+    assert _list_after("Keeping warm (rollback window)", result.stdout) == [
+        "prod-110",
+        "prod-108",
+    ]
+    # Every SERVING version except the live one and the 2 newest prod is stopped.
+    assert _list_after("Stopping", result.stdout) == [
+        "staging-110",
+        "staging-108",
+        "prod-106",
+    ]
+    # The live version is never in the stop or delete sets.
+    assert "prod-110" not in _list_after("Stopping", result.stdout)
+    assert "prod-110" not in _list_after("Deleting", result.stdout)
+    # DRY_RUN performs no mutations.
+    assert calls.read_text() == ""
+
+
+def test_traffic_holder_is_protected_even_when_it_is_an_older_version(tmp_path):
+    # Simulates a manual rollback: traffic sits on prod-106 while prod-110/108
+    # are newer. The older, traffic-serving version must NOT be stopped.
+    result, _ = _run(
+        CLEANUP_SCRIPT,
+        tmp_path,
+        serving=["prod-110", "prod-108", "prod-106", "staging-110"],
+        traffic=["prod-106"],
+        stopped=[],
+    )
+    assert result.returncode == 0, result.stderr
+    assert "prod-106" not in _list_after("Stopping", result.stdout)
+    assert _list_after("Serving traffic (never touched)", result.stdout) == ["prod-106"]
+
+
+# --- cleanup: delete selection / retention -------------------------------
+
+
+def test_deletes_only_stopped_beyond_retention_window(tmp_path):
+    stopped_prod = [f"prod-{n}" for n in range(120, 98, -2)]  # 11 prod, newest first
+    stopped_staging = [f"staging-{n}" for n in range(120, 108, -2)]  # 6 staging
+    legacy = ["20260101t120000", "20251201t120000"]
+    result, _ = _run(
+        CLEANUP_SCRIPT,
+        tmp_path,
+        serving=["prod-130"],
+        traffic=["prod-130"],
+        stopped=stopped_prod + stopped_staging + legacy,
+    )
+    assert result.returncode == 0, result.stderr
+
+    kept_prod = _list_after("Keeping stopped prod", result.stdout)
+    kept_staging = _list_after("Keeping stopped staging", result.stdout)
+    deleted = _list_after("Deleting", result.stdout)
+
+    assert kept_prod == stopped_prod[:10]  # newest 10 prod
+    assert kept_staging == stopped_staging[:3]  # newest 3 staging
+    # Everything else stopped is deleted: 1 oldest prod + 3 older staging + 2 legacy.
+    assert set(deleted) == {stopped_prod[10]} | set(stopped_staging[3:]) | set(legacy)
+    # Never deletes a SERVING version.
+    assert "prod-130" not in deleted
+
+
+def test_delete_only_targets_stopped_versions(tmp_path):
+    # A SERVING (but idle) version must be eligible for STOP, never DELETE.
+    result, _ = _run(
+        CLEANUP_SCRIPT,
+        tmp_path,
+        serving=["prod-110", "prod-108", "prod-106"],
+        traffic=["prod-110"],
+        stopped=[],
+    )
+    assert result.returncode == 0, result.stderr
+    assert _list_after("Stopping", result.stdout) == ["prod-106"]
+    assert _list_after("Deleting", result.stdout) == []
+
+
+# --- cleanup: fail-safe guard --------------------------------------------
+
+
+def test_aborts_when_serving_versions_exist_but_none_serve_traffic(tmp_path):
+    # If the traffic query returns nothing while versions are SERVING, we must
+    # NOT stop anything (we cannot identify the live service).
+    result, calls = _run(
+        CLEANUP_SCRIPT,
+        tmp_path,
+        serving=["prod-104", "prod-102", "prod-100"],
+        traffic=[],
+        stopped=["prod-98"],
+    )
+    assert result.returncode != 0
+    assert "refusing to stop" in result.stderr.lower()
+    assert "Stopping" not in result.stdout
+    assert calls.read_text() == ""
+
+
+def test_no_serving_versions_is_not_an_error(tmp_path):
+    # Nothing serving (all already stopped) is a valid state, not the guard case.
+    result, _ = _run(
+        CLEANUP_SCRIPT,
+        tmp_path,
+        serving=[],
+        traffic=[],
+        stopped=["prod-104", "prod-102"],
+    )
+    assert result.returncode == 0, result.stderr
+    assert _list_after("Stopping", result.stdout) == []
+
+
+# --- cleanup: robustness -------------------------------------------------
+
+
+def test_handles_absence_of_staging_versions(tmp_path):
+    # grep '^staging-' matches nothing; the script must not crash under set -e.
+    result, _ = _run(
+        CLEANUP_SCRIPT,
+        tmp_path,
+        serving=["prod-110"],
+        traffic=["prod-110"],
+        stopped=["prod-108", "prod-106", "20250101t000000"],
+    )
+    assert result.returncode == 0, result.stderr
+    assert _list_after("Keeping stopped staging", result.stdout) == []
+    assert _list_after("Deleting", result.stdout) == ["20250101t000000"]
+
+
+# --- stop_app_engine_version.sh ------------------------------------------
+
+
+def test_stop_script_requires_a_version(tmp_path):
+    result, _ = _run(STOP_SCRIPT, tmp_path, extra_env={"APP_ENGINE_VERSION": ""})
+    assert result.returncode != 0
+    assert "APP_ENGINE_VERSION is required" in result.stderr
+
+
+def test_stop_script_stops_the_named_version(tmp_path):
+    result, calls = _run(
+        STOP_SCRIPT,
+        tmp_path,
+        extra_env={"APP_ENGINE_VERSION": "staging-2405-abc1234", "DRY_RUN": "0"},
+    )
+    assert result.returncode == 0, result.stderr
+    logged = calls.read_text()
+    assert "STOP" in logged and "staging-2405-abc1234" in logged