# =============================================================================
# L3 FAULT-INJECTION — DR restore from an OFF-CLUSTER backup  (R-STATE-03)
#                      *SAFETY-CRITICAL*                    [RUNNABLE / self-skips]
# Catalog: docs/14-testing-and-conformance.md "DR restore (R-STATE-03, L3)"
# DoD:     agents/definition-of-done.md — "DR-critical data restores from an
#          **off-cluster** backup (drill for at least one service/phase)."
# Spec:    docs/04-storage-and-state.md; ADR-0003 (Longhorn + Velero).
# Impl:    platform/velero/** (Velero + kopia File System Backup -> the versitygw S3
#          endpoint on the Proxmox HOST), reconciled by clusters/base/backup.yaml.
#
# GIVEN a DR-critical stateful workload with known data on the replicated `longhorn`
#       tier, and a Velero BackupStorageLocation whose object-store endpoint is
#       PROVABLY OUTSIDE this cluster,
# WHEN  a backup of that namespace + its volume data is written to that off-cluster
#       store, and then EVERY in-cluster copy of the data is DESTROYED (namespace,
#       PVC and the underlying PersistentVolume/Longhorn replicas all deleted),
# THEN  a Velero restore into a CLEAN, previously-nonexistent namespace brings the
#       workload back and the known data reads back byte-identical.
#
# R-TEST-05 (fault injection, not happy-path): the restore is only meaningful because
# the source is annihilated first. After the destroy step there is no PVC, no PV and
# no Longhorn volume left in the cluster holding this data — so the bytes that come
# back can ONLY have come from the off-cluster object store. A restore that silently
# rebound a surviving in-cluster PV is explicitly excluded: the test captures the
# original PV name, asserts it is GONE, and asserts the restored PVC is bound to a
# DIFFERENT, newly provisioned PV.
#
# HOW "OFF-CLUSTER" IS PROVEN (not assumed) — step `assert-backup-target-is-off-cluster`:
#   1. The BackupStorageLocation reports phase `Available` — Velero itself has
#      successfully reached and validated that object store.
#   2. Its endpoint host is resolved to an IPv4 address and asserted to be NONE of:
#      loopback; any Node InternalIP/ExternalIP; any Service ClusterIP; any Pod IP;
#      inside any node's `spec.podCIDR(s)`; inside any `ServiceCIDR` (best-effort).
#      It is also asserted not to be a cluster-internal DNS name (*.svc,
#      *.cluster.local, localhost). => the bytes leave the cluster's address space.
#   3. The backup's PodVolumeBackup carries a non-empty `status.snapshotID` — the
#      kopia snapshot really landed in the remote repository, i.e. VOLUME DATA (not
#      merely Kubernetes resource JSON) was uploaded off-cluster.
#   4. Plus the destroy-then-restore sequence above, which removes every in-cluster
#      alternative source for the data.
#
# KNOWN, DELIBERATE GAPS (asserted nowhere — do not read this test as claiming them):
#   * OFF-HOST is NOT proven, only OFF-CLUSTER — TRUE AS RUN, and being fixed. On the
#     dev estate the k3s nodes are VMs on the same Proxmox host that runs the
#     versitygw/MinIO S3 target (192.168.1.9), so this drill does not survive loss of
#     that hypervisor: one machine takes the cluster AND its backups.
#     That is exactly why the operator moved off-cluster DR to Cloudflare R2 on
#     2026-09-05 (commit 32d6766 — S3_BACKUP_ENDPOINT is now an R2 endpoint). Once that
#     merges and this drill is RE-RUN against R2, off-host stops being a gap and becomes
#     proven, because the target no longer shares a failure domain with the cluster.
#     UNTIL THEN THIS TEST'S PASS IS AGAINST MinIO. R-STATE-03 is currently proven only
#     for the same-host target; re-proving it against R2 is a prerequisite for claiming
#     off-host DR, and the ledger row that proves R-STATE-03 predates the move.
#   * The test does NOT read the backup objects out of the bucket with an S3 client.
#     Doing so would require handing the conformance runner the backup credential,
#     widening a privileged test identity for weaker evidence than (1)-(4) above.
#   * The test restores its OWN scratch workload, never a real service's data. A
#     per-service drill (Nextcloud/Gitea/CNPG) stays an operator exercise.
#   * Bare-metal DR (rebuild the cluster itself, then restore) is NOT exercised; this
#     asserts namespace+volume restore into a live cluster, which is what the catalog
#     entry and the DoD line ask for.
#   * Backup ENCRYPTION at rest and retention/GC behaviour are not asserted.
#
# RUNNABLE / SELF-SKIPPING (a skip is honest; a fake pass is a defect). Skips CLEANLY
# (green) when a precondition is absent: no `longhorn` StorageClass; no Velero API
# (CRDs absent); no BackupStorageLocation configured at all; Velero server or its
# node-agent DaemonSet not ready. It runs the full drill the moment those exist — no
# edit needed. RBAC denial is NOT a skip: it fails loudly as a harness defect.
#
# DESTRUCTIVE BY DESIGN, SCOPED BY CONSTRUCTION: every object it destroys is one it
# created itself, in two namespaces it names after its own Chainsaw namespace
# (`<ns>-dr-src`, `<ns>-dr-dst`) and labels `conformance.home.k3s/created-by=<ns>`.
# It never touches a real service's namespace, PVC or PV, and it never deletes a
# Backup it did not create.
#
# TEARDOWN ALWAYS RUNS (`finally`, even on failure/timeout/panic): both scratch
# namespaces, any leftover PV, the Restore CR, and the Backup — the latter via a
# Velero DeleteBackupRequest so the objects are removed from the bucket too, with a
# short `ttl` on the Backup as a belt-and-braces backstop if that request is lost.
# =============================================================================
apiVersion: chainsaw.kyverno.io/v1alpha1
kind: Test
metadata:
  name: dr-restore
  labels:
    conformance.home.k3s/R-STATE-03: "true"
  annotations:
    conformance.home.k3s/requirements: "R-STATE-03"
    conformance.home.k3s/layer: "L3"
    conformance.home.k3s/safety-critical: "true"
    # executable => runs against a cluster; self-skips (green) until Velero + a
    # BackupStorageLocation + the longhorn tier exist.
    conformance.home.k3s/executable: "true"
    conformance.home.k3s/intent: >-
      Seed known data on longhorn, back the namespace + volume data up to the Velero
      object store whose endpoint is proven to be outside this cluster, destroy every
      in-cluster copy (namespace, PVC and PV), restore into a clean namespace, and read
      the known data back from a newly provisioned volume. Skips cleanly until Velero,
      a BackupStorageLocation and a longhorn StorageClass exist.
    conformance.home.k3s/never-executed: >-
      AUTHORED, NOT YET EXECUTED. No workstation in this repo has cluster access, and
      the drill is destructive, so it must first be run by the operator against the dev
      cluster in a maintenance window before R-STATE-03 may be called proven. Repo-derived
      status deliberately reports proof as not-derivable (conformance/lib/status.py).
spec:
  # A full backup -> destroy -> restore cycle over kopia File System Backup, plus
  # Longhorn provisioning on HDD, is far slower than any other test here. Bound it
  # generously but explicitly (no sleep-and-hope; every wait below is a bounded poll).
  # Worst-case bound of the drill step, summing every poll below (seed ~5m + stale-target
  # clean ~5m + backup 15m + namespace destroy 12m + PV destroy 5m + restore 15m + restored
  # workload Ready 10m ~= 67m). Declared at 75m so the step's own bounded polls, not this
  # timeout, are what fail first and report WHICH phase stalled. Nothing here sleeps blindly.
  timeouts:
    exec: 75m
    delete: 10m
    cleanup: 15m
  steps:
    # -----------------------------------------------------------------------------
    # Preconditions + the OFF-CLUSTER property of the backup target. This step makes
    # no changes to the cluster; it either skips cleanly or proves the target is
    # genuinely outside the cluster's address space.
    # -----------------------------------------------------------------------------
    - name: assert-backup-target-is-off-cluster
      try:
        - script:
            content: |
              set -eu

              # ---- helpers -------------------------------------------------------
              # Classify a kubectl failure: an RBAC denial is a HARNESS DEFECT and must
              # fail loudly; an absent API is a genuine precondition gap and may skip.
              die_or_skip() {
                case "$2" in
                  *orbidden*|*Unauthorized*|*cannot\ list*|*cannot\ get*)
                    echo "FAIL: conformance runner lacks RBAC for $1 -- fix conformance/incluster/rbac.yaml" >&2
                    echo "$2" >&2
                    exit 1 ;;
                  *)
                    echo "SKIP: $1 not available on this cluster -- $2"
                    echo DR_RESTORE_SKIPPED_OK
                    exit 0 ;;
                esac
              }

              ip2int() {
                _q="$(echo "$1" | tr '.' ' ')"
                set -- $_q
                [ $# -eq 4 ] || { echo -1; return 0; }
                echo $(( ($1 * 16777216) + ($2 * 65536) + ($3 * 256) + $4 ))
              }

              # in_cidr <ip> <a.b.c.d/len> -> 0 when the ip is inside the prefix
              in_cidr() {
                case "$2" in *"/"*) ;; *) return 1 ;; esac
                _net="${2%/*}"; _len="${2#*/}"
                case "$_len" in ''|*[!0-9]*) return 1 ;; esac
                [ "$_len" -ge 0 ] && [ "$_len" -le 32 ] || return 1
                _i="$(ip2int "$1")"; _n="$(ip2int "$_net")"
                [ "$_i" -ge 0 ] && [ "$_n" -ge 0 ] || return 1
                [ "$_len" -eq 0 ] && return 0
                _d=1; _s=0
                while [ "$_s" -lt $((32 - _len)) ]; do _d=$((_d * 2)); _s=$((_s + 1)); done
                [ $(( _i / _d )) -eq $(( _n / _d )) ]
              }

              # ---- Gate 1: the replicated storage tier the DR-critical data lives on --
              if ! kubectl get storageclass longhorn >/dev/null 2>&1; then
                echo "SKIP: 'longhorn' StorageClass absent -- no DR-critical data tier on this cluster yet"
                echo DR_RESTORE_SKIPPED_OK
                exit 0
              fi

              # ---- Gate 2: the Velero API + a configured BackupStorageLocation --------
              BSL_LINES="$(kubectl get backupstoragelocations.velero.io -A \
                -o jsonpath='{range .items[*]}{.metadata.namespace}{" "}{.metadata.name}{" "}{.status.phase}{" "}{.spec.provider}{" "}{.spec.default}{" "}{.spec.config.s3Url}{" "}{.spec.objectStorage.bucket}{"\n"}{end}' \
                2>&1)" || die_or_skip "velero.io/backupstoragelocations" "$BSL_LINES"
              if [ -z "$(echo "$BSL_LINES" | tr -d '[:space:]')" ]; then
                echo "SKIP: Velero API present but NO BackupStorageLocation is configured -- no off-cluster backup target yet"
                echo DR_RESTORE_SKIPPED_OK
                exit 0
              fi
              echo "backup storage locations:"
              echo "$BSL_LINES"

              VNS="$(echo "$BSL_LINES" | head -1 | cut -d' ' -f1)"

              # ---- Gate 3: Velero server + node-agent (kopia FSB) actually ready -------
              # File System Backup is the volume-DATA path (platform/velero/helmrelease.yaml:
              # deployNodeAgent=true, uploaderType=kopia, snapshotsEnabled=false). Without a
              # ready node-agent no volume data can be captured, so the drill is not runnable.
              VREADY=0; NAREADY=0
              for i in $(seq 1 12); do
                VREADY="$(kubectl -n "$VNS" get deploy velero -o jsonpath='{.status.readyReplicas}' 2>/dev/null || true)"
                NAREADY="$(kubectl -n "$VNS" get daemonset node-agent -o jsonpath='{.status.numberReady}' 2>/dev/null || true)"
                echo "poll $i: velero readyReplicas=${VREADY:-0} node-agent numberReady=${NAREADY:-0}"
                if [ "${VREADY:-0}" -ge 1 ] && [ "${NAREADY:-0}" -ge 1 ]; then break; fi
                sleep 10
              done
              if [ "${VREADY:-0}" -lt 1 ] || [ "${NAREADY:-0}" -lt 1 ]; then
                echo "SKIP: Velero server and/or its node-agent DaemonSet are not ready in ns '$VNS' -- backup/restore path not available"
                echo DR_RESTORE_SKIPPED_OK
                exit 0
              fi

              # ---- Pick the BSL this drill will use: prefer the default, and it MUST be
              #      Available. A CONFIGURED-BUT-UNAVAILABLE target is NOT a precondition
              #      gap -- it is a live R-STATE-03 defect, so it FAILS (after a bounded
              #      wait for first-validation on a cold start), it does not skip. --------
              BSL_NS=""; BSL_NAME=""; BSL_URL=""; BSL_BUCKET=""
              for i in $(seq 1 12); do
                BSL_LINES="$(kubectl get backupstoragelocations.velero.io -A \
                  -o jsonpath='{range .items[*]}{.metadata.namespace}{" "}{.metadata.name}{" "}{.status.phase}{" "}{.spec.provider}{" "}{.spec.default}{" "}{.spec.config.s3Url}{" "}{.spec.objectStorage.bucket}{"\n"}{end}' 2>/dev/null || true)"
                CAND="$(echo "$BSL_LINES" | grep ' Available ' | grep ' true ' | head -1 || true)"
                [ -n "$CAND" ] || CAND="$(echo "$BSL_LINES" | grep ' Available ' | head -1 || true)"
                if [ -n "$CAND" ]; then
                  BSL_NS="$(echo "$CAND" | cut -d' ' -f1)"
                  BSL_NAME="$(echo "$CAND" | cut -d' ' -f2)"
                  BSL_URL="$(echo "$CAND" | cut -d' ' -f6)"
                  BSL_BUCKET="$(echo "$CAND" | cut -d' ' -f7)"
                  break
                fi
                echo "poll $i: no Available BackupStorageLocation yet"
                sleep 10
              done
              if [ -z "$BSL_NAME" ]; then
                echo "FAIL: a BackupStorageLocation is configured but NONE reached phase 'Available'." >&2
                echo "      The off-cluster backup target is unreachable -- R-STATE-03 is BROKEN, not merely untested." >&2
                echo "$BSL_LINES" >&2
                exit 1
              fi
              echo "using BSL ${BSL_NS}/${BSL_NAME}  bucket=${BSL_BUCKET}  endpoint=${BSL_URL}"

              # ---- OFF-CLUSTER PROOF (1): an endpoint we can evaluate ------------------
              if [ -z "$BSL_URL" ]; then
                echo "FAIL: BackupStorageLocation ${BSL_NS}/${BSL_NAME} exposes no spec.config.s3Url;" >&2
                echo "      the OFF-CLUSTER property of the backup target cannot be established, so this" >&2
                echo "      test refuses to claim it. (Extend this check for a non-S3 provider.)" >&2
                exit 1
              fi
              HOST="${BSL_URL#*://}"      # strip scheme
              HOST="${HOST%%/*}"          # strip path
              HOST="${HOST%%:*}"          # strip port
              echo "backup endpoint host: $HOST"

              case "$HOST" in
                localhost|*.svc|*.svc.cluster.local|*.cluster.local)
                  echo "FAIL: backup endpoint '$HOST' is a CLUSTER-INTERNAL name -- the backup target is IN the cluster." >&2
                  echo "      An in-cluster backup store cannot satisfy R-STATE-03 (off-cluster backup)." >&2
                  exit 1 ;;
              esac

              # Resolve to an IPv4 address. If it cannot be resolved we CANNOT establish the
              # off-cluster property -- that is a failure, never a silent pass.
              case "$HOST" in
                *[!0-9.]*)
                  # Endpoint given as a name. Try the resolvers that may exist in the
                  # runner image, in order; if none is available or none resolves, the
                  # test FAILS below rather than guessing.
                  EPIP="$(getent ahostsv4 "$HOST" 2>/dev/null | head -1 | cut -d' ' -f1 || true)"
                  [ -n "$EPIP" ] || EPIP="$(getent hosts "$HOST" 2>/dev/null | head -1 | cut -d' ' -f1 || true)"
                  [ -n "$EPIP" ] || EPIP="$(nslookup "$HOST" 2>/dev/null | grep -A2 '^Name:' | grep '^Address' | head -1 | sed 's/.*: *//' || true)" ;;
                *) EPIP="$HOST" ;;
              esac
              if [ -z "$EPIP" ]; then
                echo "FAIL: could not resolve backup endpoint host '$HOST' to an IPv4 address;" >&2
                echo "      the OFF-CLUSTER property cannot be asserted, so this test does not pass." >&2
                exit 1
              fi
              echo "backup endpoint resolves to: $EPIP"

              # ---- OFF-CLUSTER PROOF (2): the endpoint is outside every cluster address --
              case "$EPIP" in 127.*|0.0.0.0) echo "FAIL: backup endpoint $EPIP is loopback -- not off-cluster" >&2; exit 1 ;; esac

              NODE_IPS="$(kubectl get nodes -o jsonpath='{range .items[*]}{range .status.addresses[*]}{.address}{"\n"}{end}{end}' 2>/dev/null || true)"
              for a in $NODE_IPS; do
                if [ "$a" = "$EPIP" ]; then
                  echo "FAIL: backup endpoint $EPIP is a NODE address of this cluster -- the backup is not off-cluster." >&2
                  exit 1
                fi
              done
              echo "endpoint is not any node address (checked $(echo "$NODE_IPS" | grep -c . || true) node addresses)"

              SVC_IPS="$(kubectl get svc -A -o jsonpath='{range .items[*]}{.spec.clusterIP}{"\n"}{end}' 2>/dev/null || true)"
              for a in $SVC_IPS; do
                if [ "$a" = "$EPIP" ]; then
                  echo "FAIL: backup endpoint $EPIP is a Service ClusterIP -- the backup target lives INSIDE the cluster." >&2
                  exit 1
                fi
              done
              echo "endpoint is not any Service ClusterIP (checked $(echo "$SVC_IPS" | grep -c . || true))"

              POD_IPS="$(kubectl get pods -A -o jsonpath='{range .items[*]}{.status.podIP}{"\n"}{end}' 2>/dev/null || true)"
              for a in $POD_IPS; do
                if [ "$a" = "$EPIP" ]; then
                  echo "FAIL: backup endpoint $EPIP is a POD IP -- the backup target is an in-cluster pod." >&2
                  exit 1
                fi
              done
              echo "endpoint is not any Pod IP (checked $(echo "$POD_IPS" | grep -c . || true))"

              POD_CIDRS="$(kubectl get nodes -o jsonpath='{range .items[*]}{.spec.podCIDR}{"\n"}{range .spec.podCIDRs[*]}{@}{"\n"}{end}{end}' 2>/dev/null || true)"
              for c in $POD_CIDRS; do
                case "$c" in *:*) continue ;; esac      # IPv6 prefixes are out of scope here
                if in_cidr "$EPIP" "$c"; then
                  echo "FAIL: backup endpoint $EPIP falls inside node podCIDR $c -- it is in the cluster pod network." >&2
                  exit 1
                fi
              done
              echo "endpoint is outside every node podCIDR (checked: $(echo "$POD_CIDRS" | tr '\n' ' '))"

              # Best-effort (needs networking.k8s.io/ServiceCIDR, which the least-privilege
              # runner may not read): the checks above are the authoritative ones.
              SVC_CIDRS="$(kubectl get servicecidrs.networking.k8s.io -o jsonpath='{range .items[*]}{range .spec.cidrs[*]}{@}{"\n"}{end}{end}' 2>/dev/null || true)"
              if [ -n "$SVC_CIDRS" ]; then
                for c in $SVC_CIDRS; do
                  case "$c" in *:*) continue ;; esac
                  if in_cidr "$EPIP" "$c"; then
                    echo "FAIL: backup endpoint $EPIP falls inside the Service CIDR $c -- it is in the cluster." >&2
                    exit 1
                  fi
                done
                echo "endpoint is outside every ServiceCIDR ($(echo "$SVC_CIDRS" | tr '\n' ' '))"
              else
                echo "(ServiceCIDR objects unreadable -- skipping that best-effort check; ClusterIP/PodIP/podCIDR checks stand)"
              fi

              echo "OFF_CLUSTER_TARGET_OK  bsl=${BSL_NS}/${BSL_NAME} endpoint=${EPIP} bucket=${BSL_BUCKET}"
            check:
              # green on either a clean precondition skip or a proven off-cluster target.
              (contains($stdout, 'OFF_CLUSTER_TARGET_OK') || contains($stdout, 'DR_RESTORE_SKIPPED_OK')): true

    # -----------------------------------------------------------------------------
    # R-STATE-03's "scheduled" half: the backup is not a one-off human action.
    # Non-destructive, and skipped in lock-step with the step above.
    # -----------------------------------------------------------------------------
    - name: assert-backups-are-scheduled
      try:
        - script:
            content: |
              set -eu
              SCHED="$(kubectl get schedules.velero.io -A \
                -o jsonpath='{range .items[*]}{.metadata.namespace}{"/"}{.metadata.name}{" phase="}{.status.phase}{" cron="}{.spec.schedule}{" lastBackup="}{.status.lastBackup}{"\n"}{end}' 2>/dev/null || true)"
              if [ -z "$(echo "$SCHED" | tr -d '[:space:]')" ]; then
                # Distinguish "Velero absent" (skip, in lock-step with step 1) from
                # "Velero present but nothing scheduled" (a real R-STATE-03 defect).
                if kubectl get backupstoragelocations.velero.io -A >/dev/null 2>&1; then
                  if [ -z "$(kubectl get backupstoragelocations.velero.io -A -o name 2>/dev/null || true)" ]; then
                    echo "SKIP: no BackupStorageLocation configured -- scheduling not applicable yet"
                    echo DR_SCHEDULE_SKIPPED_OK
                    exit 0
                  fi
                  echo "FAIL: Velero is configured with a backup target but declares NO Schedule --" >&2
                  echo "      R-STATE-03 requires SCHEDULED off-cluster backups, not manual ones." >&2
                  exit 1
                fi
                echo "SKIP: Velero API absent -- backup scheduling not applicable on this cluster yet"
                echo DR_SCHEDULE_SKIPPED_OK
                exit 0
              fi
              echo "velero schedules:"
              echo "$SCHED"
              # A Schedule that exists but is paused/disabled backs nothing up.
              if echo "$SCHED" | grep -q 'phase=Enabled'; then
                echo BACKUPS_ARE_SCHEDULED_OK
              else
                echo "FAIL: no Velero Schedule is in phase 'Enabled' -- scheduled off-cluster backups are not running." >&2
                exit 1
              fi
            check:
              (contains($stdout, 'BACKUPS_ARE_SCHEDULED_OK') || contains($stdout, 'DR_SCHEDULE_SKIPPED_OK')): true

    # -----------------------------------------------------------------------------
    # The drill itself: seed -> back up off-cluster -> ANNIHILATE -> restore -> read.
    # -----------------------------------------------------------------------------
    - name: data-restores-from-the-off-cluster-backup
      try:
        - script:
            content: |
              set -eu
              NS="$NAMESPACE"
              SRC="${NS}-dr-src"          # scratch source namespace (destroyed by design)
              DST="${NS}-dr-dst"          # clean restore target (must not pre-exist)
              BK="${NS}-dr-backup"
              RS="${NS}-dr-restore"
              OWNER_LABEL="conformance.home.k3s/created-by=${NS}"

              # ---- Re-derive the preconditions (steps do not share shell state). Every
              #      skip below mirrors step 1 exactly, so the whole test skips or runs
              #      as a unit -- it can never half-run. -----------------------------
              if ! kubectl get storageclass longhorn >/dev/null 2>&1; then
                echo "SKIP: 'longhorn' StorageClass absent"; echo DR_RESTORE_SKIPPED_OK; exit 0
              fi
              BSL_LINES="$(kubectl get backupstoragelocations.velero.io -A \
                -o jsonpath='{range .items[*]}{.metadata.namespace}{" "}{.metadata.name}{" "}{.status.phase}{" "}{.spec.default}{"\n"}{end}' 2>/dev/null || true)"
              CAND="$(echo "$BSL_LINES" | grep ' Available ' | grep ' true$' | head -1 || true)"
              [ -n "$CAND" ] || CAND="$(echo "$BSL_LINES" | grep ' Available ' | head -1 || true)"
              if [ -z "$CAND" ]; then
                echo "SKIP: no Available BackupStorageLocation (see the off-cluster step for the authoritative verdict)"
                echo DR_RESTORE_SKIPPED_OK; exit 0
              fi
              VNS="$(echo "$CAND" | cut -d' ' -f1)"
              BSL_NAME="$(echo "$CAND" | cut -d' ' -f2)"
              if [ "$(kubectl -n "$VNS" get daemonset node-agent -o jsonpath='{.status.numberReady}' 2>/dev/null || echo 0)" -lt 1 ]; then
                echo "SKIP: Velero node-agent (kopia FSB) not ready -- volume data cannot be backed up"
                echo DR_RESTORE_SKIPPED_OK; exit 0
              fi
              echo "velero ns=$VNS bsl=$BSL_NAME src=$SRC dst=$DST"

              # A clean restore target means PREVIOUSLY NONEXISTENT, and the source name must
              # be free. If either is lying around from an aborted run, remove it before we
              # claim "clean" -- and fail loudly if it will not go, because a half-deleted
              # namespace would silently invalidate the destroy proof below.
              kubectl delete namespace "$DST" "$SRC" --ignore-not-found --wait=true --timeout=300s

              # ============================ GIVEN ==============================
              # A DR-critical stateful workload with known data on the replicated tier.
              CANARY="dr-canary-$(date +%s%N)"
              kubectl create namespace "$SRC"
              kubectl label namespace "$SRC" "$OWNER_LABEL" --overwrite

              kubectl -n "$SRC" apply -f - <<EOF
              apiVersion: v1
              kind: PersistentVolumeClaim
              metadata:
                name: dr-data
                labels:
                  conformance.home.k3s/created-by: ${NS}
              spec:
                accessModes: [ReadWriteOnce]
                storageClassName: longhorn
                resources:
                  requests:
                    storage: 1Gi
              EOF

              kubectl -n "$SRC" apply -f - <<EOF
              apiVersion: v1
              kind: Pod
              metadata:
                name: dr-workload
                labels:
                  app: dr-workload
                  conformance.home.k3s/created-by: ${NS}
              spec:
                restartPolicy: Never
                containers:
                  - name: app
                    image: busybox:1.36.1
                    command: ["sh", "-c", "sleep 7200"]
                    volumeMounts:
                      - name: data
                        mountPath: /data
                volumes:
                  - name: data
                    persistentVolumeClaim:
                      claimName: dr-data
              EOF

              kubectl -n "$SRC" wait --for=condition=Ready pod/dr-workload --timeout=300s
              printf '%s' "$CANARY" | kubectl -n "$SRC" exec -i dr-workload -- sh -c 'cat > /data/canary && sync'
              echo "seeded canary: $(kubectl -n "$SRC" exec dr-workload -- cat /data/canary)"

              ORIG_PV="$(kubectl -n "$SRC" get pvc dr-data -o jsonpath='{.spec.volumeName}')"
              [ -n "$ORIG_PV" ] || { echo "FAIL: source PVC never bound to a PV" >&2; exit 1; }
              echo "original in-cluster PersistentVolume: $ORIG_PV"

              # ======================= WHEN (part 1): back up off-cluster =======
              # Created as a Velero Backup CR -- exactly what `velero backup create` does.
              # defaultVolumesToFsBackup:true selects the kopia File System Backup path, so
              # the VOLUME DATA (not just the resource JSON) is uploaded to the object store.
              # ttl is short: if the finally-step DeleteBackupRequest is ever lost, Velero's
              # own GC still removes this drill's objects from the bucket.
              kubectl apply -f - <<EOF
              apiVersion: velero.io/v1
              kind: Backup
              metadata:
                name: ${BK}
                namespace: ${VNS}
                labels:
                  conformance.home.k3s/created-by: ${NS}
              spec:
                includedNamespaces: [${SRC}]
                storageLocation: ${BSL_NAME}
                snapshotVolumes: false
                defaultVolumesToFsBackup: true
                ttl: 2h0m0s
              EOF

              BPHASE=""
              for i in $(seq 1 90); do
                BPHASE="$(kubectl -n "$VNS" get backups.velero.io "$BK" -o jsonpath='{.status.phase}' 2>/dev/null || true)"
                echo "poll $i: backup phase=${BPHASE:-<pending>}"
                case "${BPHASE:-}" in
                  Completed|Failed|PartiallyFailed|FailedValidation) break ;;
                esac
                sleep 10
              done
              if [ "${BPHASE:-}" != "Completed" ]; then
                echo "FAIL: off-cluster backup did not complete (phase=${BPHASE:-<timeout>})" >&2
                kubectl -n "$VNS" get backups.velero.io "$BK" -o yaml >&2 || true
                exit 1
              fi
              echo "backup $BK Completed to ${BSL_NAME}"

              # ---- OFF-CLUSTER PROOF (3): the VOLUME DATA really landed in the remote
              #      kopia repository -- a PodVolumeBackup Completed with a snapshot ID.
              PVB="$(kubectl -n "$VNS" get podvolumebackups.velero.io \
                       -l velero.io/backup-name="$BK" \
                       -o jsonpath='{range .items[*]}{.metadata.name}{" "}{.status.phase}{" "}{.status.snapshotID}{"\n"}{end}' 2>/dev/null || true)"
              echo "pod volume backups:"
              echo "$PVB"
              if [ -z "$(echo "$PVB" | tr -d '[:space:]')" ]; then
                echo "FAIL: the backup captured NO pod volume data (no PodVolumeBackup)." >&2
                echo "      Resource-only backups cannot restore DR-critical DATA." >&2
                exit 1
              fi
              # A line is "<name> <phase> <snapshotID>"; require Completed AND a
              # non-empty snapshot ID (an empty ID means nothing reached the repository).
              GOOD_PVB="$(echo "$PVB" | grep -cE ' Completed [^[:space:]]+$' || true)"
              if [ "${GOOD_PVB:-0}" -lt 1 ]; then
                echo "FAIL: no PodVolumeBackup reached phase Completed with a snapshot ID --" >&2
                echo "      the volume data did not reach the off-cluster repository." >&2
                exit 1
              fi
              echo "volume data uploaded off-cluster (PodVolumeBackup Completed with a kopia snapshotID)"

              # ============ WHEN (part 2): FAULT INJECTION -- annihilate the source ====
              # Destroy EVERY in-cluster copy: the pod, the PVC, the namespace, and the
              # underlying PersistentVolume (and with it the Longhorn replicas). After this
              # there is no in-cluster source the restore could quietly read from.
              kubectl delete namespace "$SRC" --wait=true --timeout=420s
              for i in $(seq 1 30); do
                if ! kubectl get namespace "$SRC" >/dev/null 2>&1; then break; fi
                echo "poll $i: waiting for namespace $SRC to disappear"
                sleep 10
              done
              if kubectl get namespace "$SRC" >/dev/null 2>&1; then
                echo "FAIL: source namespace $SRC did not terminate; the destroy step is not convincing" >&2
                exit 1
              fi
              echo "source namespace destroyed"

              # The longhorn StorageClass reclaims Delete, so the PV should already be gone;
              # force it if a Retain policy (or a stuck finalizer) left it behind, because a
              # surviving PV would make the restore prove nothing about off-cluster recovery.
              kubectl delete pv "$ORIG_PV" --ignore-not-found --wait=false || true
              for i in $(seq 1 30); do
                if ! kubectl get pv "$ORIG_PV" >/dev/null 2>&1; then break; fi
                echo "poll $i: waiting for PersistentVolume $ORIG_PV to disappear"
                sleep 10
              done
              if kubectl get pv "$ORIG_PV" >/dev/null 2>&1; then
                echo "FAIL: the original PersistentVolume $ORIG_PV SURVIVED the destroy step." >&2
                echo "      An in-cluster copy of the data still exists, so a successful restore" >&2
                echo "      would NOT prove recovery from the off-cluster backup. Refusing to pass." >&2
                exit 1
              fi
              echo "original PersistentVolume $ORIG_PV is GONE -- no in-cluster copy of the data remains"

              # Best-effort corroboration (the least-privilege runner may not read longhorn.io):
              # the Longhorn Volume CR and its replicas should be gone too.
              if kubectl -n longhorn-system get volumes.longhorn.io "$ORIG_PV" >/dev/null 2>&1; then
                echo "FAIL: Longhorn Volume $ORIG_PV still exists after PV deletion -- replicas survive in-cluster" >&2
                exit 1
              fi
              echo "(longhorn Volume CR $ORIG_PV not present -- either deleted or not readable by this identity)"

              # ============================ RESTORE ============================
              # Into a CLEAN, previously-nonexistent namespace (Velero creates it), which is
              # the catalog's "restored into a clean state".
              kubectl apply -f - <<EOF
              apiVersion: velero.io/v1
              kind: Restore
              metadata:
                name: ${RS}
                namespace: ${VNS}
                labels:
                  conformance.home.k3s/created-by: ${NS}
              spec:
                backupName: ${BK}
                includedNamespaces: [${SRC}]
                namespaceMapping:
                  ${SRC}: ${DST}
                existingResourcePolicy: none
              EOF

              RPHASE=""
              for i in $(seq 1 90); do
                RPHASE="$(kubectl -n "$VNS" get restores.velero.io "$RS" -o jsonpath='{.status.phase}' 2>/dev/null || true)"
                echo "poll $i: restore phase=${RPHASE:-<pending>}"
                case "${RPHASE:-}" in
                  Completed|Failed|PartiallyFailed|FailedValidation) break ;;
                esac
                sleep 10
              done
              if [ "${RPHASE:-}" != "Completed" ]; then
                echo "FAIL: restore did not complete (phase=${RPHASE:-<timeout>})" >&2
                kubectl -n "$VNS" get restores.velero.io "$RS" -o yaml >&2 || true
                kubectl -n "$VNS" get podvolumerestores.velero.io -l velero.io/restore-name="$RS" -o wide >&2 || true
                exit 1
              fi
              echo "restore $RS Completed into $DST"

              # ============================= THEN ==============================
              # The service returns WITH ITS DATA. With kopia FSB the restored pod carries a
              # `restore-wait` init container that blocks until the node-agent has written the
              # volume contents back, so Ready here already implies the data landed.
              if ! kubectl -n "$DST" wait --for=condition=Ready pod/dr-workload --timeout=600s; then
                echo "FAIL: restored workload never became Ready in the clean namespace $DST" >&2
                kubectl -n "$DST" describe pod dr-workload >&2 || true
                kubectl -n "$VNS" get podvolumerestores.velero.io -l velero.io/restore-name="$RS" -o wide >&2 || true
                exit 1
              fi

              NEW_PV="$(kubectl -n "$DST" get pvc dr-data -o jsonpath='{.spec.volumeName}' 2>/dev/null || true)"
              echo "restored PVC bound to PersistentVolume: ${NEW_PV:-<none>}"
              [ -n "$NEW_PV" ] || { echo "FAIL: restored PVC did not bind a PV" >&2; exit 1; }
              if [ "$NEW_PV" = "$ORIG_PV" ]; then
                echo "FAIL: the restored PVC rebound the ORIGINAL PersistentVolume ($ORIG_PV)." >&2
                echo "      That is an in-cluster survival, not an off-cluster restore." >&2
                exit 1
              fi
              echo "restored onto a NEW volume ($NEW_PV != $ORIG_PV) -- the data came from outside the cluster"

              GOT="$(kubectl -n "$DST" exec dr-workload -- cat /data/canary)"
              echo "recovered='$GOT' expected='$CANARY'"
              if [ "$GOT" != "$CANARY" ]; then
                echo "FAIL: restored data does not match what was backed up -- DR restore lost/corrupted data" >&2
                exit 1
              fi

              echo DR_RESTORED_FROM_OFF_CLUSTER_BACKUP_OK
            check:
              # green on either a clean precondition skip or a proven off-cluster restore.
              (contains($stdout, 'DR_RESTORED_FROM_OFF_CLUSTER_BACKUP_OK') || contains($stdout, 'DR_RESTORE_SKIPPED_OK')): true
      finally:
        # ALWAYS RUNS -- on pass, fail, timeout or panic. This test creates namespaces
        # and Velero objects OUTSIDE its Chainsaw namespace, so Chainsaw's own namespace
        # teardown cannot reclaim them; everything below is keyed to names/labels derived
        # from THIS test's namespace, so it can never touch anything it did not create.
        - script:
            content: |
              set -u
              NS="$NAMESPACE"
              SRC="${NS}-dr-src"; DST="${NS}-dr-dst"; BK="${NS}-dr-backup"; RS="${NS}-dr-restore"

              VNS="$(kubectl get backupstoragelocations.velero.io -A -o jsonpath='{.items[0].metadata.namespace}' 2>/dev/null || true)"

              # Velero objects first: a DeleteBackupRequest makes Velero remove the backup
              # DATA from the off-cluster bucket, not just the CR (self-cleaning contract).
              if [ -n "$VNS" ]; then
                kubectl -n "$VNS" delete restores.velero.io "$RS" --ignore-not-found --wait=false || true
                if kubectl -n "$VNS" get backups.velero.io "$BK" >/dev/null 2>&1; then
                  kubectl apply -f - <<EOF || true
              apiVersion: velero.io/v1
              kind: DeleteBackupRequest
              metadata:
                name: ${BK}-delete
                namespace: ${VNS}
                labels:
                  conformance.home.k3s/created-by: ${NS}
              spec:
                backupName: ${BK}
              EOF
                  for i in $(seq 1 18); do
                    kubectl -n "$VNS" get backups.velero.io "$BK" >/dev/null 2>&1 || break
                    sleep 10
                  done
                  # Backstop if Velero could not process the request (the 2h ttl also GCs it).
                  kubectl -n "$VNS" delete backups.velero.io "$BK" --ignore-not-found --wait=false || true
                  kubectl -n "$VNS" delete deletebackuprequests.velero.io "${BK}-delete" --ignore-not-found --wait=false || true
                fi
                echo "velero objects for $NS cleaned up (ns $VNS)"
              fi

              # Scratch namespaces (both, in case the run aborted at any point).
              for n in "$DST" "$SRC"; do
                kubectl delete namespace "$n" --ignore-not-found --wait=false || true
              done
              # Anything left labelled as ours (defensive; the two names above are the only
              # namespaces this test ever creates).
              kubectl delete namespace -l "conformance.home.k3s/created-by=${NS}" --ignore-not-found --wait=false || true

              # Released PVs from either namespace's PVCs would otherwise leak storage.
              for pv in $(kubectl get pv -o jsonpath='{range .items[*]}{.metadata.name}{" "}{.spec.claimRef.namespace}{"\n"}{end}' 2>/dev/null | grep -E " (${SRC}|${DST})$" | cut -d' ' -f1 || true); do
                kubectl delete pv "$pv" --ignore-not-found --wait=false || true
                echo "released leftover PersistentVolume: $pv"
              done

              echo "dr-restore teardown complete (best-effort; every object was created by this test)"