diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index f994223..2230e88 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -11,6 +11,16 @@ on: description: Exact 40-character commit SHA approved for this operation. required: true type: string + notification_only: + description: Use the reviewed notification overlay and preserve the selected 00781 TCP configuration. + default: false + required: true + type: boolean + notification_candidate_revision: + description: Exact fresh revision name for notification_only. No latest aliases or reuse. + default: "" + required: false + type: string approve_image_deploy: description: Create a new Cloud Run revision without traffic. default: false @@ -167,6 +177,9 @@ jobs: steps: - name: Check whether Cloud Run automation is enabled id: config + env: + NOTIFICATION_ONLY: ${{ inputs.notification_only }} + NOTIFICATION_CANDIDATE_REVISION: ${{ inputs.notification_candidate_revision }} run: | set -euo pipefail @@ -211,6 +224,27 @@ jobs: cleanup_enabled=true fi + if [ "${NOTIFICATION_ONLY:-false}" = "true" ]; then + for approval in APPROVE_ENV_SECRET_SYNC APPROVE_SCHEDULER_IAM_SYNC APPROVE_TRAFFIC_SHIFT APPROVE_CLEANUP; do + if [ "${!approval:-false}" != "false" ]; then + echo "Notification-only cannot combine environment, IAM, traffic or cleanup operations." >&2 + exit 1 + fi + done + if [[ ! "${NOTIFICATION_CANDIDATE_REVISION:-}" =~ ^charles-schwab-quant-service-[a-z0-9-]+$ ]] || [ "${#NOTIFICATION_CANDIDATE_REVISION}" -gt 63 ]; then + echo "Exact notification candidate revision required." >&2 + exit 1 + fi + env_sync_enabled=false + scheduler_iam_sync_enabled=false + traffic_shift_enabled=false + cleanup_enabled=false + elif [ "${NOTIFICATION_ONLY:-false}" != "false" ]; then + echo "Invalid notification-only selector." >&2 + exit 1 + fi + write_github_output "notification_only=${NOTIFICATION_ONLY:-false}" + write_github_output \ "deploy_enabled=${deploy_enabled}" \ "env_sync_enabled=${env_sync_enabled}" \ @@ -259,7 +293,7 @@ jobs: python-version: "3.12" - name: Install strategy status dependencies - if: steps.config.outputs.enabled == 'true' + if: steps.config.outputs.enabled == 'true' && inputs.notification_only != true run: | set -euo pipefail python -m pip install --upgrade pip uv @@ -394,7 +428,7 @@ jobs: version: ">= 416.0.0" - name: Capture no-traffic deployment baseline - if: steps.config.outputs.deploy_enabled == 'true' + if: steps.config.outputs.deploy_enabled == 'true' && inputs.notification_only != true env: DEPLOY_READBACK_FILE: ${{ runner.temp }}/cloud-run-no-traffic-baseline.json run: | @@ -412,7 +446,7 @@ jobs: --output="${DEPLOY_READBACK_FILE}" - name: Verify deployed runtime target admission before traffic shift - if: steps.config.outputs.deploy_enabled == 'true' || steps.config.outputs.env_sync_enabled == 'true' || steps.config.outputs.traffic_shift_enabled == 'true' + if: inputs.notification_only != true && (steps.config.outputs.deploy_enabled == 'true' || steps.config.outputs.env_sync_enabled == 'true' || steps.config.outputs.traffic_shift_enabled == 'true') run: | set -euo pipefail uv run --no-sync python scripts/verify_deployed_runtime_target_admission.py \ @@ -423,7 +457,7 @@ jobs: - name: Build, push, and deploy Cloud Run image id: deploy - if: steps.config.outputs.deploy_enabled == 'true' + if: steps.config.outputs.deploy_enabled == 'true' && inputs.notification_only != true env: SYNC_PLAN_JSON: ${{ steps.strategy_requirements.outputs.sync_plan_json }} run: | @@ -473,7 +507,7 @@ jobs: --quiet - name: Verify no-traffic deployment readback - if: steps.config.outputs.deploy_enabled == 'true' + if: steps.config.outputs.deploy_enabled == 'true' && inputs.notification_only != true env: DEPLOY_READBACK_FILE: ${{ runner.temp }}/cloud-run-no-traffic-baseline.json EXPECTED_IMAGE_DIGEST: ${{ steps.deploy.outputs.image_digest }} @@ -493,6 +527,93 @@ jobs: --expected-sha="${EXPECTED_SHA}" \ --expected-image-digest="${EXPECTED_IMAGE_DIGEST}" + - name: Checkout reviewed notification source + if: steps.config.outputs.notification_only == 'true' && steps.config.outputs.deploy_enabled == 'true' + uses: actions/checkout@v6 + with: + ref: e159e6cc07f9a9eea417b810674f934eb5291e6f + fetch-depth: 0 + path: notification-source + + - name: Capture selected native notification baseline + if: steps.config.outputs.notification_only == 'true' && steps.config.outputs.deploy_enabled == 'true' + run: | + set -euo pipefail + umask 077 + python3 scripts/verify_cloud_run_no_traffic_deploy.py capture-native \ + --project="${GCP_PROJECT_ID}" --region="${CLOUD_RUN_REGION}" --service="${CLOUD_RUN_SERVICE}" \ + --baseline-revision=charles-schwab-quant-service-00781-lel \ + --output="${RUNNER_TEMP}/native-notification-before.json" + + - name: Build exact notification overlay without dependency installation + if: steps.config.outputs.notification_only == 'true' && steps.config.outputs.deploy_enabled == 'true' + id: notification_image + env: + NOTIFICATION_CANDIDATE_REVISION: ${{ inputs.notification_candidate_revision }} + run: | + set -euo pipefail + umask 077 + # Validate the frozen serving baseline before spending build resources. + # The placeholder is used only by the pure planner, never submitted. + python3 - <<'PY_CHECK' + import json, os + from pathlib import Path + from scripts.build_schwab_notification_revision_plan import build_native_plan, NOTIFICATION_SOURCE + before = json.loads((Path(os.environ["RUNNER_TEMP"]) / "native-notification-before.json").read_text()) + expected = dict(project=os.environ["GCP_PROJECT_ID"], region=os.environ["CLOUD_RUN_REGION"], service=os.environ["CLOUD_RUN_SERVICE"], + baseline_revision="charles-schwab-quant-service-00781-lel", candidate_revision=os.environ["NOTIFICATION_CANDIDATE_REVISION"], + candidate_image="registry.invalid/prebuild-only@sha256:" + "0" * 64, + resource_version=before["service"]["metadata"]["resourceVersion"], candidate_sha=NOTIFICATION_SOURCE, candidate_run=os.environ["GITHUB_RUN_ID"]) + import copy + build_native_plan(before, copy.deepcopy(before), expected) + print("Selected baseline pre-build validation passed; image and runtime adoption remain unverified.") + PY_CHECK + context="${RUNNER_TEMP}/notification-overlay" + python3 scripts/build_schwab_notification_revision_plan.py prepare-overlay \ + --repository=notification-source --output="${context}" + artifact_registry_hostname="${GCP_ARTIFACT_REGISTRY_HOSTNAME:-${CLOUD_RUN_REGION}-docker.pkg.dev}" + image_repo="${artifact_registry_hostname}/${GCP_PROJECT_ID}/${GCP_ARTIFACT_REGISTRY_REPOSITORY}/charlesschwabplatform/${CLOUD_RUN_SERVICE}" + image="${image_repo}:notification-e159-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" + gcloud auth configure-docker "${artifact_registry_hostname}" --quiet + docker build -t "${image}" "${context}" + # The existing RS-08 image probe runs Python with --network none, not + # the application entrypoint. Use the OLD source lock and exact target. + NOTIFICATION_IMAGE="${image}" python3 - <<'PY' + import json, os + from pathlib import Path + from scripts.verify_image_ues_release_binding import main + before = json.loads((Path(os.environ["RUNNER_TEMP"]) / "native-notification-before.json").read_text()) + runtime_literals = [e["value"] for e in before["revision"]["spec"]["containers"][0]["env"] + if e["name"] in {"RUNTIME_TARGET_JSON", "QSL_RUNTIME_TARGET_JSON"} and "value" in e] + if len(runtime_literals) != 1: + raise SystemExit("Selected runtime target unavailable; stopped") + for name in ("SYNC_PLAN_JSON", "CLOUD_RUN_SERVICE_TARGETS_JSON"): + os.environ.pop(name, None) + os.environ["RUNTIME_TARGET_JSON"] = runtime_literals[0] + raise SystemExit(main(["--image", os.environ["NOTIFICATION_IMAGE"], "--lockfile", "notification-source/uv.lock"])) + PY + docker push "${image}" + digest="$(gcloud artifacts docker images describe "${image}" --project="${GCP_PROJECT_ID}" --format='value(image_summary.digest)')" + [[ "${digest}" =~ ^sha256:[0-9a-f]{64}$ ]] || { echo "Invalid notification digest" >&2; exit 1; } + printf 'immutable_image=%s@%s\n' "${image_repo}" "${digest}" >> "${GITHUB_OUTPUT}" + + - name: Publish one native notification revision and wait for its TCP startup + if: steps.config.outputs.notification_only == 'true' && steps.config.outputs.deploy_enabled == 'true' + env: + NOTIFICATION_IMAGE: ${{ steps.notification_image.outputs.immutable_image }} + NOTIFICATION_CANDIDATE_REVISION: ${{ inputs.notification_candidate_revision }} + run: | + set -euo pipefail + # Reuse setup-gcloud's authentication libraries; no dependency sync. + sdk_root="$(gcloud info --format='value(installation.sdk_root)')" + PYTHONPATH="${sdk_root}/lib:${sdk_root}/lib/third_party${PYTHONPATH:+:${PYTHONPATH}}" \ + python3 scripts/build_schwab_notification_revision_plan.py native-publish \ + --before="${RUNNER_TEMP}/native-notification-before.json" \ + --project="${GCP_PROJECT_ID}" --region="${CLOUD_RUN_REGION}" --service="${CLOUD_RUN_SERVICE}" \ + --baseline-revision=charles-schwab-quant-service-00781-lel \ + --candidate-revision="${NOTIFICATION_CANDIDATE_REVISION}" \ + --candidate-image="${NOTIFICATION_IMAGE}" --candidate-run="${GITHUB_RUN_ID}" + - name: Sync Cloud Run environment if: steps.config.outputs.env_sync_enabled == 'true' env: diff --git a/scripts/build_schwab_notification_revision_plan.py b/scripts/build_schwab_notification_revision_plan.py index b7d01c1..bfbfa58 100644 --- a/scripts/build_schwab_notification_revision_plan.py +++ b/scripts/build_schwab_notification_revision_plan.py @@ -347,8 +347,249 @@ class _Parser(argparse.ArgumentParser): def error(self, message: str) -> None: _stop("INVALID_ARGUMENTS") +# Native publishing is separate from the closed synthetic review model above. +# It never imports business code, changes a runtime target, or sends an app probe. +NOTIFICATION_SOURCE = "e159e6cc07f9a9eea417b810674f934eb5291e6f" +OLD_UES_PIN = "4a3943883cd6b5bbfe32a559e56a91b40a81b7ce" +_DEPENDENCY_HASHES = { + "uv.lock": "5bead5ad0b79876f74dd6b02ab29538e5cc5ace2420c15f6222e77c7f458a66f", + "pyproject.toml": "2a943c7217b46912e6da0ac1bc382e4a2c9fff644aadf0da5077ba1d911fe780", + "qsl.toml": "979b31053ff222a496e6b39718b0f1b412fa317ad2db009807deb4efedaf0807", +} +_OVERLAY_EXISTING = frozenset({ + "application/account_new_risk_gate_support.py", "application/execution_service.py", + "application/rebalance_service.py", "application/runtime_report_summary.py", + "application/runtime_strategy_adapters.py", "main.py", +}) +_OVERLAY_REQUIRED = _OVERLAY_EXISTING | { + "notifications/display_names.json", "notifications/digest_snapshot/provenance.json", + "scripts/execution_report_heartbeat.py", "scripts/emit_unified_digest_candidates.py", +} +_OUTPUT_LABELS = a.NATIVE_OUTPUT_LABELS +_OUTPUT_ANNOTATIONS = a.NATIVE_OUTPUT_ANNOTATIONS + + +def build_native_plan(before: dict, current: dict, expected: dict) -> tuple[dict, dict]: + """Construct one native v1 Service guarded by its ORIGINAL resourceVersion. + + A supported native snapshot is required; a closed-model projection or a + desired template is never a source for the new revision's configuration. + """ + try: + a.native_snapshot_inventory(before) + a.native_snapshot_inventory(current) + _required(expected, _EXPECTED | {"candidate_sha": str, "candidate_run": str}) + _expected({k: expected[k] for k in _EXPECTED}, TCP_00781_STARTUP_POLICY) + if expected["candidate_sha"] != NOTIFICATION_SOURCE or not re.fullmatch(r"[0-9]+", expected["candidate_run"]): + _stop("NATIVE_SOURCE_OR_RUN_MISMATCH") + for snap in (before, current): + for part, name in (("service", expected["service"]), ("revision", expected["baseline_revision"])): + meta = snap[part]["metadata"] + if (meta["name"] != name or not (meta["namespace"] == expected["project"] or re.fullmatch(r"[1-9][0-9]*", meta["namespace"])) + or not _nonempty(meta["uid"]) or not _nonempty(meta["resourceVersion"])): + _stop("NATIVE_IDENTITY_MISMATCH") + if snap["service"]["metadata"]["namespace"] != snap["revision"]["metadata"]["namespace"]: + _stop("NATIVE_NAMESPACE_MISMATCH") + if snap["service"]["metadata"]["resourceVersion"] != expected["resource_version"]: + _stop("RESOURCE_VERSION_MISMATCH") + revision = snap["revision"] + main_container = revision["spec"]["containers"][0] + if (main_container["image"] != _TCP_00781_IMAGE + or revision["metadata"].get("labels", {}).get("commit-sha") != _TCP_00781_SOURCE): + _stop("NATIVE_BASELINE_PROVENANCE_MISMATCH") + if main_container["startupProbe"] != _TCP_00781_PROBE: + _stop("TCP_BASELINE_PROBE_MISMATCH") + # Absence is a capability/decision gap, not permission to enable it. + if revision["metadata"].get("annotations", {}).get("run.googleapis.com/health-check-disabled") != "false": + _stop("NATIVE_STARTUP_CHECK_NOT_EXPLICITLY_ENABLED") + _traffic(a.native_routes(snap["service"]["spec"]["traffic"]), expected) + if a.native_routes(snap["service"]["status"]["traffic"], status=True) != a.native_routes(snap["service"]["spec"]["traffic"]): + _stop("NATIVE_TRAFFIC_NOT_OBSERVED") + if sum(row["percent"] for row in a.native_routes(snap["service"]["spec"]["traffic"]) + if row["revisionName"] == expected["baseline_revision"]) != 100: + _stop("NATIVE_BASELINE_NOT_100_PERCENT") + targets = [entry["value"] for entry in main_container["env"] + if entry["name"] in {"RUNTIME_TARGET_JSON", "QSL_RUNTIME_TARGET_JSON"} and "value" in entry] + if len(targets) != 1: + _stop("NATIVE_TARGET_BINDING_UNAVAILABLE") + target = json.loads(targets[0], object_pairs_hook=_pairs, parse_constant=_invalid_constant) + if (target.get("strategy_release", {}).get("strategy_revision") != OLD_UES_PIN + or target.get("runtime_risk_limits", {}).get("binding", {}).get("ues_revision") != OLD_UES_PIN): + _stop("NATIVE_OLD_UES_BINDING_MISMATCH") + for container in revision["spec"]["containers"]: + if not _image(container["image"]): _stop("INVALID_IMMUTABLE_IMAGE") + if container.get("command") or container.get("args"): _stop("NATIVE_COMMAND_OVERRIDE_UNSUPPORTED") + for entry in container.get("env", []): + if "value" in entry and type(entry["value"]) is not str: _stop("INVALID_ENV_LITERAL") + if before is current: _stop("INDEPENDENT_SNAPSHOTS_REQUIRED") + if before != current: _stop("CONCURRENT_SNAPSHOT_CHANGED") + revision_meta = current["revision"]["metadata"] + labels = {k: v for k, v in revision_meta.get("labels", {}).items() if k not in _OUTPUT_LABELS} + labels.update({"commit-sha": NOTIFICATION_SOURCE, "github-run-id": expected["candidate_run"]}) + if "run-id" in labels: labels["run-id"] = expected["candidate_run"] + # Old build provenance describes the base, not this overlay source. + annotations = {k: v for k, v in revision_meta.get("annotations", {}).items() + if k not in _OUTPUT_ANNOTATIONS and not k.startswith("run.googleapis.com/build-")} + spec = copy.deepcopy(current["revision"]["spec"]) + spec["containers"][0]["image"] = expected["candidate_image"] + if expected["candidate_image"] == _TCP_00781_IMAGE: _stop("IMAGE_UNCHANGED") + service_meta = current["service"]["metadata"] + metadata = {k: copy.deepcopy(service_meta[k]) for k in ("name", "namespace", "resourceVersion", "labels", "annotations") if k in service_meta} + request = {"apiVersion": "serving.knative.dev/v1", "kind": "Service", "metadata": metadata, + "spec": {"template": {"metadata": {"name": expected["candidate_revision"], "labels": labels, + "annotations": annotations}, "spec": spec}, + "traffic": copy.deepcopy(current["service"]["spec"]["traffic"])}} + return request, {"mode": "native_original_rv", "deployment_ready": False, + "source": NOTIFICATION_SOURCE, "source_is_tool_main": False, + "protected_baseline_sha256": _hash(current), "request_sha256": _hash(request), + "allowed_changes": ["template.name", "template.spec.containers[0].image", "template.source_provenance"], + "service_desired_template": "REPLACED_FROM_EXPLICIT_BASELINE_NOT_FROM_DESIRED", + "traffic_and_all_tags": "PRESERVED", "effective_notification_flags": "UNVERIFIED", + "runtime_marker_CAS": "UNVERIFIED", "startup": "NOT_EXECUTED_TCP_ONLY"} + except PlanError: + raise + except a.PreflightError as exc: + raise PlanError(str(exc)) from None + except (ValueError, TypeError, KeyError, AttributeError, RecursionError): + _stop("INVALID_NATIVE_INPUT") + + +def replace_native_service(request: dict, expected: dict, *, session=None) -> None: + """One native PUT. No gcloud replace, automatic retry, response dump or rebase.""" + owned = session is None + if request.get("metadata", {}).get("resourceVersion") != expected["resource_version"]: + _stop("RESOURCE_VERSION_MISMATCH") + _expected({k: expected[k] for k in _EXPECTED}, TCP_00781_STARTUP_POLICY) + response = None + try: + if owned: + import google.auth + from google.auth.transport.requests import AuthorizedSession + credentials, _project = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"]) + session = AuthorizedSession(credentials, max_refresh_attempts=0) + url = (f"https://{expected['region']}-run.googleapis.com/apis/serving.knative.dev/v1/" + f"namespaces/{expected['project']}/services/{expected['service']}") + response = session.request("PUT", url, json=request, timeout=60, allow_redirects=False, stream=True) + status = response.status_code + except Exception: + _stop("NATIVE_WRITE_UNCERTAIN_DO_NOT_RETRY") + finally: + if response is not None: response.close() + if owned and session is not None: session.close() + if status in {409, 412}: _stop("NATIVE_RESOURCE_VERSION_CONFLICT") + if status != 200: _stop("NATIVE_WRITE_UNCERTAIN_DO_NOT_RETRY") + + +def publish_native(before: dict, expected: dict, *, read=None, capture=None, replace=None, pause=None, attempts: int = 24) -> dict: + """Read fresh, write once, then wait only on exact candidate control-plane GETs.""" + import time + read = read or a._run_json + capture = capture or a.capture_native_snapshot + replace = replace or replace_native_service + pause = pause or time.sleep + if type(attempts) is not int or not 1 <= attempts <= 24: _stop("INVALID_POLL_BOUND") + current = capture(expected, read=read) + request, review = build_native_plan(before, current, expected) + candidates = read(["gcloud", "run", "revisions", "list", f"--project={expected['project']}", + f"--region={expected['region']}", f"--filter=metadata.name={expected['candidate_revision']}", + "--format=json(metadata.name)"]) + if candidates != []: _stop("NATIVE_CANDIDATE_ALREADY_EXISTS_OR_AMBIGUOUS") + replace(request, expected) + candidate_identity = dict(expected, baseline_revision=expected["candidate_revision"]) + for index in range(attempts): + after = capture(candidate_identity, read=read) + try: + a.verify_native_readback(before, request, after["service"], after["revision"]) + except a.PreflightError as exc: + if str(exc) != "NATIVE_STARTUP_PENDING": raise PlanError(str(exc)) from None + if index == attempts - 1: _stop("NATIVE_STARTUP_NOT_OBSERVED_DO_NOT_REDEPLOY") + pause(15) + else: + return review | {"startup": "CONTROL_PLANE_TCP_READY_ONLY", "application_delivery": "NOT_VERIFIED"} + _stop("NATIVE_STARTUP_NOT_OBSERVED_DO_NOT_REDEPLOY") + + +def _git_bytes(repository: Path, *args: str) -> bytes: + import subprocess + result = subprocess.run(["git", "-C", str(repository), *args], capture_output=True, check=False) + if result.returncode: _stop("NATIVE_SOURCE_GIT_UNAVAILABLE") + return result.stdout + + +def prepare_notification_overlay(repository: Path, destination: Path, *, git=None) -> dict: + """Only exact reviewed Git blobs enter the context; no workspace/auth files.""" + git = git or (lambda *args: _git_bytes(repository, *args)) + if destination.exists(): _stop("OVERLAY_DESTINATION_EXISTS") + if git("rev-parse", "HEAD").decode().strip() != NOTIFICATION_SOURCE: _stop("OVERLAY_SOURCE_MISMATCH") + if git("merge-base", _TCP_00781_SOURCE, NOTIFICATION_SOURCE).decode().strip() != _TCP_00781_SOURCE: _stop("OVERLAY_BASE_MISMATCH") + for name, digest in _DEPENDENCY_HASHES.items(): + blob = git("show", NOTIFICATION_SOURCE + ":" + name) + if hashlib.sha256(blob).hexdigest() != digest or blob != git("show", _TCP_00781_SOURCE + ":" + name): + _stop("OVERLAY_DEPENDENCY_CHANGED") + rows = git("diff", "--name-status", "--no-renames", _TCP_00781_SOURCE, NOTIFICATION_SOURCE).decode().splitlines() + paths = [] + for row in rows: + status, name = row.split("\t") + runtime = (name in _OVERLAY_EXISTING or name in _OVERLAY_REQUIRED + or re.fullmatch(r"notifications/(?:[a-z_]+/)*[a-z_]+\.py", name)) + documentation = re.fullmatch(r"(?:tests|docs)/[A-Za-z0-9_./-]+", name) or name in {"README.md", "CHANGELOG.md", "MANIFEST.in", ".github/workflows/execution-report-heartbeat.yml"} + if not runtime: + if documentation: continue + _stop("OVERLAY_UNREVIEWED_PATH") + if status not in {"A", "M"}: _stop("OVERLAY_DELETE_OR_RENAME") + if ".." in Path(name).parts: _stop("OVERLAY_INVALID_PATH") + record = git("ls-tree", NOTIFICATION_SOURCE, "--", name).decode().split() + if len(record) != 4 or record[0] not in {"100644", "100755"} or record[1] != "blob": + _stop("OVERLAY_NOT_REGULAR_FILE") + paths.append(name) + if not _OVERLAY_REQUIRED <= set(paths): _stop("OVERLAY_REQUIRED_ASSET_MISSING") + blobs = {name: git("show", NOTIFICATION_SOURCE + ":" + name) for name in paths} + destination.mkdir(parents=True, mode=0o700) + for name, blob in blobs.items(): + output = destination / name + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(blob) + # Docker's JSON COPY form: no shell expansion, no RUN, no dependency install. + recipe = "FROM " + _TCP_00781_IMAGE + "\n" + "".join( + "COPY " + json.dumps([name, "/app/" + name]) + "\n" for name in sorted(blobs)) + (destination / "Dockerfile").write_text(recipe, encoding="utf-8") + return {"source": NOTIFICATION_SOURCE, "base": _TCP_00781_SOURCE, "files_sha256": { + name: hashlib.sha256(blob).hexdigest() for name, blob in sorted(blobs.items())}, + "build": "NOT_EXECUTED", "installed_dependencies": "PRESERVED_BY_BASE_IMAGE_NOT_RUNTIME_VERIFIED"} + + +def _native_main(argv: list[str]) -> int: + parser = _Parser() + modes = parser.add_subparsers(dest="mode", required=True) + prepare = modes.add_parser("prepare-overlay") + prepare.add_argument("--repository", required=True, type=Path) + prepare.add_argument("--output", required=True, type=Path) + publish = modes.add_parser("native-publish") + publish.add_argument("--before", required=True, type=Path) + for flag in ("project", "region", "service", "baseline-revision", "candidate-revision", "candidate-image", "candidate-run"): + publish.add_argument("--" + flag, required=True) + try: + args = parser.parse_args(argv) + if args.mode == "prepare-overlay": + result = prepare_notification_overlay(args.repository, args.output) + else: + before = load_request(args.before) + expected = {k: getattr(args, k) for k in _EXPECTED if k != "resource_version"} + expected["resource_version"] = before["service"]["metadata"]["resourceVersion"] + expected.update(candidate_sha=NOTIFICATION_SOURCE, candidate_run=args.candidate_run) + result = publish_native(before, expected) + print(_canonical(result)) + except (PlanError, a.PreflightError) as exc: + print("Native operation stopped: " + str(exc), file=sys.stderr); return 1 + except (ValueError, TypeError, KeyError, OSError, AttributeError): + print("Native operation stopped: INVALID_INPUT", file=sys.stderr); return 1 + return 0 + def main(argv: list[str] | None = None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + if argv and argv[0] in {"prepare-overlay", "native-publish"}: + return _native_main(argv) parser = _Parser(description="Review a closed local synthetic snapshot; emits hashes only.") parser.add_argument("--input", required=True, type=Path) try: diff --git a/scripts/verify_cloud_run_no_traffic_deploy.py b/scripts/verify_cloud_run_no_traffic_deploy.py index 1025c83..671901a 100644 --- a/scripts/verify_cloud_run_no_traffic_deploy.py +++ b/scripts/verify_cloud_run_no_traffic_deploy.py @@ -6,6 +6,7 @@ import argparse import hashlib import json +import os import re import subprocess import sys @@ -658,6 +659,290 @@ def _preflight_local(args: argparse.Namespace, baseline: object, candidate: obje _stop("LIVE_UES_PIN_NOT_ADMITTED") +# Native capture deliberately keeps the Service's desired template out of the +# response. Every supported Revision config field is inventoried by NAME before +# a value projection; unknown fields or env literals stop that projection. +_NATIVE_MAP = object() +_NATIVE_METADATA = { + "name": str, "namespace": str, "uid": str, "resourceVersion": str, + "generation": int, "creationTimestamp": str, "selfLink": str, + "labels": _NATIVE_MAP, "annotations": _NATIVE_MAP, + "ownerReferences": [{"apiVersion": str, "kind": str, "name": str, "uid": str, + "controller": bool, "blockOwnerDeletion": bool}], +} +_NATIVE_SPEC = { + "serviceAccountName": str, "containerConcurrency": int, "timeoutSeconds": int, + "containers": [_CONTAINER | {"workingDir": str}], + "volumes": _RUNTIME["volumes"], +} +_NATIVE_ANNOTATIONS = frozenset({ + "run.googleapis.com/client-name", "run.googleapis.com/client-version", "run.googleapis.com/operation-id", + "run.googleapis.com/ingress", "run.googleapis.com/ingress-status", "run.googleapis.com/urls", + "run.googleapis.com/launch-stage", "run.googleapis.com/execution-environment", + "run.googleapis.com/network-interfaces", "run.googleapis.com/vpc-access-connector", + "run.googleapis.com/vpc-access-egress", "run.googleapis.com/startup-cpu-boost", + "run.googleapis.com/cpu-throttling", "run.googleapis.com/container-dependencies", + "run.googleapis.com/health-check-disabled", "run.googleapis.com/secrets", + "run.googleapis.com/build-source-location", "run.googleapis.com/build-id", + "run.googleapis.com/build-image-uri", "run.googleapis.com/base-images", + "autoscaling.knative.dev/minScale", "autoscaling.knative.dev/maxScale", + "serving.knative.dev/creator", "serving.knative.dev/lastModifier", +}) +NATIVE_OUTPUT_LABELS = frozenset({"serving.knative.dev/service", "serving.knative.dev/configuration", + "serving.knative.dev/configurationGeneration", "serving.knative.dev/route", + "cloud.googleapis.com/location", "run.googleapis.com/startupProbeType"}) +NATIVE_OUTPUT_ANNOTATIONS = frozenset({"serving.knative.dev/creator", "serving.knative.dev/lastModifier", + "run.googleapis.com/operation-id", "run.googleapis.com/urls", "run.googleapis.com/ingress-status", + "run.googleapis.com/client-name", "run.googleapis.com/client-version"}) +_NATIVE_CONDITIONS = [{"type": str, "status": str, "reason": str, "lastTransitionTime": str}] + + +def _native_fields(schema: object, prefix: str) -> list[str]: + if schema is _ENV: + return [prefix + ".name", prefix + ".valueFrom.secretKeyRef"] + if isinstance(schema, dict): + return [field for key, child in schema.items() + for field in _native_fields(child, prefix + "." + key)] + if isinstance(schema, list): + return _native_fields(schema[0], prefix + "[]") + return [prefix] + + +def _native_at(value: object, path: str) -> list: + values = [value] + for part in path.split("."): + key, repeated = part.removesuffix("[]"), part.endswith("[]") + next_values = [] + for item in values: + if isinstance(item, dict) and key in item: + child = item[key] + next_values.extend(child if repeated and isinstance(child, list) else [child]) + values = next_values + return values + + +def _native_inventory(read, command: list[str], schema: object, path: str) -> None: + if schema is _ENV: + schema = {"name": str, "value": str, "valueFrom": {"secretKeyRef": {"name": str, "key": str}}} + if isinstance(schema, list): + _native_inventory(read, command, schema[0], path + "[]") + elif isinstance(schema, dict): + # Separate projections avoid a parent list(keys) replacing its children + # in gcloud's JSON formatter. Only field names leave gcloud here. + names = read(command + [f"--format=json({path}.list(show=keys))"]) + for item in _native_at(names, path): + if not isinstance(item, str) or set(filter(None, item.split(","))) - schema.keys(): + _stop("NATIVE_UNSUPPORTED_CONFIGURATION_FIELD") + for key, child in schema.items(): + _native_inventory(read, command, child, path + "." + key) + + +def _native_metadata(value: object) -> None: + _keys(value, _NATIVE_METADATA, {"name", "namespace", "uid", "resourceVersion"}) + for key, child in value.items(): + if key in {"labels", "annotations"}: + if key == "annotations" and isinstance(child, dict) and child.keys() - _NATIVE_ANNOTATIONS: + _stop("NATIVE_UNSUPPORTED_ANNOTATION") + if not isinstance(child, dict) or any(type(k) is not str or type(v) is not str for k, v in child.items()): + _stop("NATIVE_INVALID_METADATA") + else: + _shape(child, _NATIVE_METADATA[key]) + + + +def native_routes(rows: object, *, status: bool = False) -> list: + if not isinstance(rows, list): _stop("NATIVE_INVALID_TRAFFIC") + result = [] + for row in rows: + _keys(row, {"revisionName", "percent", "tag", "latestRevision"} | ({"url"} if status else set()), {"revisionName"}) + if "latestRevision" in row and row["latestRevision"] is not False: + _stop("NATIVE_LATEST_TRAFFIC_UNSUPPORTED") + if type(row["revisionName"]) is not str or type(row.get("percent", 0)) is not int: + _stop("NATIVE_INVALID_TRAFFIC") + normalized = {"revisionName": row["revisionName"], "percent": row.get("percent", 0)} + if "tag" in row: + if type(row["tag"]) is not str: _stop("NATIVE_INVALID_TRAFFIC") + normalized["tag"] = row["tag"] + if "url" in row and type(row["url"]) is not str: _stop("NATIVE_INVALID_TRAFFIC") + result.append(normalized) + return sorted(result, key=_canonical) + + +def native_snapshot_inventory(snapshot: object) -> None: + """Validate ALL env names before any copying, hashing or literal access.""" + _keys(snapshot, {"service", "revision"}, {"service", "revision"}) + for key in ("service", "revision"): + obj = snapshot[key] + _keys(obj, {"metadata", "spec", "status"}, {"metadata", "spec", "status"}) + _native_metadata(obj["metadata"]) + _keys(snapshot["service"]["spec"], {"traffic"}, {"traffic"}) + native_routes(snapshot["service"]["spec"]["traffic"]) + _keys(snapshot["service"]["status"], {"traffic", "observedGeneration", "latestCreatedRevisionName"}, {"traffic"}) + if "observedGeneration" in snapshot["service"]["status"]: + _shape(snapshot["service"]["status"]["observedGeneration"], int) + if "latestCreatedRevisionName" in snapshot["service"]["status"]: + _shape(snapshot["service"]["status"]["latestCreatedRevisionName"], str) + native_routes(snapshot["service"]["status"]["traffic"], status=True) + _shape(snapshot["revision"]["status"], {"imageDigest": str, "observedGeneration": int, "conditions": _NATIVE_CONDITIONS}) + _shape(snapshot["revision"]["spec"], _NATIVE_SPEC) + _keys(snapshot["revision"]["spec"], _NATIVE_SPEC, {"containers", "serviceAccountName"}) + containers = snapshot["revision"]["spec"]["containers"] + if not containers: + _stop("MISSING_CONTAINER") + for container in containers: + _keys(container, _NATIVE_SPEC["containers"][0], {"image", "env", "ports", "startupProbe"} if container is containers[0] else {"image"}) + names = [entry["name"] for entry in container.get("env", [])] + if len(names) != len(set(names)): + _stop("DUPLICATE_ENV_NAME") + + +def capture_native_snapshot(identity: dict, *, read=None) -> dict: + """Read supported native fields; never export desired or unknown env values. + + Nonempty command/args overrides and unsupported config shapes are explicit + capability gaps, rather than defaults or silently discarded configuration. + """ + read = read or _run_json + for key in ("project", "region", "service", "baseline_revision"): + if type(identity.get(key)) is not str or not re.fullmatch(r"[a-z][a-z0-9-]*", identity[key]): + _stop("INVALID_EXPECTED_IDENTITY") + if not identity["baseline_revision"].startswith(identity["service"] + "-"): + _stop("INVALID_EXPECTED_IDENTITY") + shared = [f"--project={identity['project']}", f"--region={identity['region']}"] + service_cmd = ["gcloud", "run", "services", "describe", identity["service"], *shared] + revision_cmd = ["gcloud", "run", "revisions", "describe", identity["baseline_revision"], *shared] + commands = {"service": service_cmd, "revision": revision_cmd} + stamps, annotation_names = {}, {} + # The earliest UID/RV covers every name inventory, not only later values. + for kind, cmd in commands.items(): + stamp = read(cmd + ["--format=json(metadata.uid,metadata.resourceVersion)"]) + meta = stamp.get("metadata") if isinstance(stamp, dict) else None + _keys(meta, {"uid", "resourceVersion"}, {"uid", "resourceVersion"}) + if any(type(meta[k]) is not str or not meta[k] for k in meta): + _stop("NATIVE_INVALID_INITIAL_STAMP") + stamps[kind] = meta + _native_inventory(read, service_cmd, {"template": _NATIVE_MAP, "traffic": _NATIVE_MAP}, "spec") + _native_inventory(read, revision_cmd, _NATIVE_SPEC, "spec") + for kind, cmd in commands.items(): + annotation_names[kind] = set() + _native_inventory(read, cmd, _NATIVE_METADATA, "metadata") + # A last-applied blob or arbitrary annotation can embed secret env + # values. Reject its NAME before requesting any metadata values. + keys = read(cmd + ["--format=json(metadata.annotations.list(show=keys))"]) + for names in _native_at(keys, "metadata.annotations"): + if not isinstance(names, str) or set(filter(None, names.split(","))) - _NATIVE_ANNOTATIONS: + _stop("NATIVE_UNSUPPORTED_ANNOTATION") + annotation_names[kind].update(filter(None, names.split(","))) + # A literal name without a secret reference must be explicitly allowlisted. + names = read(revision_cmd + ["--format=json(metadata.uid,metadata.resourceVersion,spec.containers[].env[].name,spec.containers[].env[].valueFrom.secretKeyRef)"]) + if names.get("metadata") != stamps["revision"]: + _stop("NATIVE_OBJECT_CHANGED_DURING_CAPTURE") + literal_names = [] + for env in _native_at(names, "spec.containers[].env[]"): + _keys(env, {"name", "valueFrom"}, {"name"}) + if "valueFrom" not in env: + if env["name"] not in LITERAL_ENV: + _stop("UNKNOWN_ENV_LITERAL") + literal_names.append(env["name"]) + else: + _shape(env, _ENV) + if len(literal_names) != len(set(literal_names)): + _stop("NATIVE_AMBIGUOUS_LITERAL_CONTAINER") + # Overrides may contain credentials. Project lengths only, even if a + # different object appears later; a final RV rejection cannot unread text. + empty_overrides = {} + for field in ("command", "args"): + lengths = read(revision_cmd + [f"--format=json(spec.containers[].image,spec.containers[].{field}.len())"]) + for index, row in enumerate(_native_at(lengths, "spec.containers[]")): + if field in row: + if type(row[field]) is not int or row[field] != 0: + _stop("NATIVE_COMMAND_OVERRIDE_UNSUPPORTED") + empty_overrides.setdefault(index, []).append(field) + for path in ("spec.traffic", "status.traffic"): + _native_inventory(read, service_cmd, [{"revisionName": str, "percent": int, "tag": str, "latestRevision": bool, **({"url": str} if path.startswith("status") else {})}], path) + spec_fields = [field for field in _native_fields(_NATIVE_SPEC, "spec") + if field not in {"spec.containers[].command[]", "spec.containers[].args[]"}] + def metadata_fields(kind): + fields = [field for key, schema in _NATIVE_METADATA.items() if key != "annotations" + for field in _native_fields(schema, "metadata." + key)] + return fields + ["metadata.annotations." + json.dumps(key) for key in sorted(annotation_names[kind])] + service = read(service_cmd + ["--format=json(" + ",".join([ + *metadata_fields("service"), "spec.traffic", "status.traffic", "status.observedGeneration", + "status.latestCreatedRevisionName", + ]) + ")"]) + revision = read(revision_cmd + ["--format=json(" + ",".join([ + *metadata_fields("revision"), *spec_fields, "status.imageDigest", "status.observedGeneration", + *_native_fields(_NATIVE_CONDITIONS, "status.conditions"), + ]) + ")"]) + for kind, obj in (("service", service), ("revision", revision)): + if not isinstance(obj, dict) or {k: obj.get("metadata", {}).get(k) for k in ("uid", "resourceVersion")} != stamps[kind]: + _stop("NATIVE_OBJECT_CHANGED_DURING_CAPTURE") + # Preserve explicitly observed empty override fields without requesting + # their text. Nonempty overrides were rejected; UID/RV is still bracketed. + for index, fields in empty_overrides.items(): + if index >= len(revision.get("spec", {}).get("containers", [])): + _stop("NATIVE_OBJECT_CHANGED_DURING_CAPTURE") + for field in fields: + revision["spec"]["containers"][index][field] = [] + for name in literal_names: + fmt = 'json(spec.containers[].env.always().filter("name=' + name + '").map().extract(value))' + values = _cash_values(read(revision_cmd + ["--format=" + fmt])) + if len(values) != 1 or type(values[0]) is not str: + _stop("NATIVE_LITERAL_UNAVAILABLE") + for env in _native_at(revision, "spec.containers[].env[]"): + if env["name"] == name: + env["value"] = values[0] + for kind, cmd in commands.items(): + stamp = read(cmd + ["--format=json(metadata.uid,metadata.resourceVersion)"]) + if not isinstance(stamp, dict) or stamp.get("metadata") != stamps[kind]: + _stop("NATIVE_OBJECT_CHANGED_DURING_CAPTURE") + snapshot = {"service": service, "revision": revision} + native_snapshot_inventory(snapshot) + return snapshot + + +def verify_native_readback(before: dict, request: dict, service: dict, revision: dict) -> None: + """Check the exact candidate, not latest, without any application request.""" + native_snapshot_inventory({"service": service, "revision": revision}) + old_meta, new_meta = before["service"]["metadata"], service["metadata"] + if any(new_meta[k] != old_meta[k] for k in ("name", "namespace", "uid")): + _stop("NATIVE_SERVICE_IDENTITY_CHANGED") + if service["spec"]["traffic"] != before["service"]["spec"]["traffic"] or native_routes(service["status"]["traffic"], status=True) != native_routes(before["service"]["status"]["traffic"], status=True): + _stop("NATIVE_TRAFFIC_CHANGED") + template = request["spec"]["template"] + if (revision["metadata"]["name"] != template["metadata"]["name"] + or revision["metadata"]["namespace"] != old_meta["namespace"] + or not revision["metadata"]["uid"]): + _stop("NATIVE_WRONG_CANDIDATE") + if revision["spec"] != template["spec"]: + _stop("NATIVE_CANDIDATE_CONFIGURATION_CHANGED") + # Compare complete protected metadata; only closed server-output sets differ. + for obj, wanted_meta in ((service, request["metadata"]), (revision, template["metadata"])): + for field, output in (("labels", NATIVE_OUTPUT_LABELS), ("annotations", NATIVE_OUTPUT_ANNOTATIONS)): + actual = {k: v for k, v in obj["metadata"].get(field, {}).items() if k not in output} + wanted = {k: v for k, v in wanted_meta.get(field, {}).items() if k not in output} + if actual != wanted: + _stop("NATIVE_PROTECTED_METADATA_CHANGED") + if revision["status"].get("imageDigest") != template["spec"]["containers"][0]["image"]: + _stop("NATIVE_CANDIDATE_IMAGE_CHANGED") + for obj in (service, revision): + generation = obj["metadata"].get("generation") + if type(generation) is not int or obj["status"].get("observedGeneration") != generation: + _stop("NATIVE_STARTUP_PENDING") + latest = service["status"].get("latestCreatedRevisionName") + if latest != template["metadata"]["name"]: + old_generation = old_meta.get("generation") + if (latest and type(old_generation) is int + and service["metadata"]["generation"] > old_generation): + _stop("NATIVE_OTHER_CANDIDATE_OBSERVED") + _stop("NATIVE_STARTUP_PENDING") + conditions = {x["type"]: x["status"] for x in revision["status"].get("conditions", [])} + if any(conditions.get(k) == "False" for k in ("Ready", "ContainerHealthy")): + _stop("NATIVE_STARTUP_FAILED") + if any(conditions.get(k) != "True" for k in ("Ready", "ContainerHealthy")): + _stop("NATIVE_STARTUP_PENDING") + + def main() -> int: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) @@ -679,9 +964,22 @@ def main() -> int: local.add_argument("--candidate-run") for flag in ("baseline", "candidate", "baseline-repository", "candidate-repository"): local.add_argument(f"--{flag}", required=True, type=Path) + native = subparsers.add_parser("capture-native", help="Read a selected supported native revision, stopping on unknown literals") + for flag in ("project", "region", "service", "baseline-revision"): + native.add_argument(f"--{flag}", required=True) + native.add_argument("--output", required=True, type=Path) args = parser.parse_args() try: - if args.command == "capture": + if args.command == "capture-native": + snapshot = capture_native_snapshot(vars(args)) + try: + descriptor = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as output: + output.write(_canonical(snapshot)) + except OSError: + _stop("NATIVE_CAPTURE_OUTPUT_UNAVAILABLE") + print("Captured selected native configuration; no application health or notification delivery verified.") + elif args.command == "capture": args.output.write_text(_canonical(_snapshot(args)), encoding="utf-8") print("Captured non-secret Cloud Run deployment baseline.") elif args.command == "verify": diff --git a/tests/test_build_schwab_notification_revision_plan.py b/tests/test_build_schwab_notification_revision_plan.py index 2612ccb..b1ffa34 100644 --- a/tests/test_build_schwab_notification_revision_plan.py +++ b/tests/test_build_schwab_notification_revision_plan.py @@ -1120,3 +1120,209 @@ def test_tcp_real_disk_cli_failures_are_fixed_and_redacted(tmp_path, tcp_case, f assert result.stdout == "" assert result.stderr == "Review stopped: " + reason + "\nLOCAL_GUARD_EVENTS=[]\n" assert SENTINEL not in result.stderr + + +@pytest.fixture +def native_case(monkeypatch): + b = planner() + identity = dict(project="demo-project", region="demo-region", service="demo-service", baseline_revision=BASE) + monkeypatch.setattr(b, "_TCP_00781_IDENTITY", identity) + monkeypatch.setattr(b, "_TCP_00781_IMAGE", OLD_IMAGE) + traffic = [{"revisionName": BASE, "percent": 100, "tag": "stable"}, + {"revisionName": DESIRED, "percent": 0, "tag": "keep-zero"}] + meta = {"name": "demo-service", "namespace": "demo-project", "uid": "service-uid", + "resourceVersion": "rv/original==", "generation": 7, + "labels": {"managed-by": "synthetic"}, "annotations": {"run.googleapis.com/ingress": "internal"}} + revision_meta = dict(meta, name=BASE, uid="revision-uid", resourceVersion="revision-rv", generation=1, + labels={"commit-sha": b._TCP_00781_SOURCE, "serving.knative.dev/service": "demo-service"}, + annotations={"run.googleapis.com/health-check-disabled": "false", "autoscaling.knative.dev/maxScale": "1"}) + raw_target = json.dumps({"strategy_release": {"strategy_revision": b.OLD_UES_PIN}, + "runtime_risk_limits": {"binding": {"ues_revision": b.OLD_UES_PIN}}, + "live_continuity": {"signature": "unchanged-synthetic"}}) + spec = {"serviceAccountName": "synthetic@example.invalid", "containerConcurrency": 1, "timeoutSeconds": 300, + "containers": [{"image": OLD_IMAGE, "ports": [{"containerPort": 8080}], + "startupProbe": copy.deepcopy(b._TCP_00781_PROBE), + "env": [{"name": "RUNTIME_TARGET_JSON", "value": raw_target}, + {"name": "CREDENTIAL", "valueFrom": {"secretKeyRef": {"name": "synthetic-ref", "key": "3"}}}], + "resources": {"limits": {"cpu": "1", "memory": "512Mi"}}}, + {"name": "helper", "image": OLD_IMAGE, "env": []}], "volumes": []} + snap = {"service": {"metadata": meta, "spec": {"traffic": traffic}, + "status": {"traffic": copy.deepcopy(traffic), "observedGeneration": 7}}, + "revision": {"metadata": revision_meta, "spec": spec, "status": {"imageDigest": OLD_IMAGE, "observedGeneration": 1}}} + expected = dict(identity, candidate_revision=CANDIDATE, candidate_image=NEW_IMAGE, + resource_version=meta["resourceVersion"], candidate_sha=b.NOTIFICATION_SOURCE, candidate_run="987") + return b, snap, expected + + +def test_native_request_is_detached_and_keeps_config_and_zero_tags(native_case): + b, before, expected = native_case + frozen = copy.deepcopy(before) + request, report = b.build_native_plan(before, copy.deepcopy(before), expected) + assert before == frozen + assert request["metadata"]["resourceVersion"] == "rv/original==" + assert request["spec"]["traffic"] == before["service"]["spec"]["traffic"] + template = request["spec"]["template"] + assert template["metadata"]["name"] == CANDIDATE + assert set(template["metadata"]) == {"name", "labels", "annotations"} + assert template["metadata"]["labels"]["commit-sha"] == b.NOTIFICATION_SOURCE + assert "serving.knative.dev/service" not in template["metadata"]["labels"] + restored = copy.deepcopy(template["spec"]) + restored["containers"][0]["image"] = OLD_IMAGE + assert restored == before["revision"]["spec"] + assert report["deployment_ready"] is False + assert "unchanged-synthetic" not in json.dumps(report) + + +@pytest.mark.parametrize("change", ["service-uid", "service-rv", "revision-uid", "zero-tag", "probe", "sidecar", "target"]) +def test_native_stale_or_changed_snapshots_stop(native_case, change): + b, before, expected = native_case + current = copy.deepcopy(before) + if change == "service-uid": current["service"]["metadata"]["uid"] = "recreated" + elif change == "service-rv": current["service"]["metadata"]["resourceVersion"] = "later" + elif change == "revision-uid": current["revision"]["metadata"]["uid"] = "recreated" + elif change == "zero-tag": current["service"]["spec"]["traffic"][1]["tag"] = "changed" + elif change == "probe": current["revision"]["spec"]["containers"][0]["startupProbe"]["timeoutSeconds"] = 1 + elif change == "sidecar": current["revision"]["spec"]["containers"][1]["image"] = NEW_IMAGE + else: current["revision"]["spec"]["containers"][0]["env"][0]["value"] += " " + with pytest.raises(b.PlanError): b.build_native_plan(before, current, expected) + + +@pytest.mark.parametrize("change", ["desired", "source", "old-image", "disabled-startup", "missing-startup", "wrong-pin", "unknown-env", "latest-route"]) +def test_native_unsupported_or_wrong_baseline_is_not_defaulted(native_case, change): + b, before, expected = native_case + if change == "desired": before["revision"]["metadata"]["name"] = DESIRED + elif change == "source": before["revision"]["metadata"]["labels"]["commit-sha"] = "d" * 40 + elif change == "old-image": before["revision"]["spec"]["containers"][0]["image"] = NEW_IMAGE + elif change == "disabled-startup": before["revision"]["metadata"]["annotations"]["run.googleapis.com/health-check-disabled"] = "true" + elif change == "missing-startup": del before["revision"]["metadata"]["annotations"]["run.googleapis.com/health-check-disabled"] + elif change == "wrong-pin": before["revision"]["spec"]["containers"][0]["env"][0]["value"] = '{"strategy_release":{"strategy_revision":"wrong"}}' + elif change == "unknown-env": before["revision"]["spec"]["containers"][0]["env"].append({"name": "PRIVATE", "value": SENTINEL}) + else: before["service"]["spec"]["traffic"][0]["latestRevision"] = True + with pytest.raises(b.PlanError) as error: b.build_native_plan(before, copy.deepcopy(before), expected) + assert SENTINEL not in str(error.value) + + +@pytest.mark.parametrize("status", [200, 409, 412, 500]) +def test_native_transport_one_request_original_rv_no_body_read(native_case, status): + b, before, expected = native_case + request, _ = b.build_native_plan(before, copy.deepcopy(before), expected) + calls = [] + class Response: + status_code = status + def close(self): pass + def json(self): raise AssertionError("Never read response config or error text") + class Session: + def request(self, method, url, **kw): + calls.append((method, url, kw)); return Response() + if status == 200: b.replace_native_service(request, expected, session=Session()) + else: + with pytest.raises(b.PlanError): b.replace_native_service(request, expected, session=Session()) + assert len(calls) == 1 and calls[0][0] == "PUT" + assert calls[0][2]["json"]["metadata"]["resourceVersion"] == "rv/original==" + assert calls[0][2]["allow_redirects"] is False and calls[0][2]["stream"] is True + + +def test_native_transport_uncertain_error_is_not_retried(native_case): + b, before, expected = native_case + request, _ = b.build_native_plan(before, copy.deepcopy(before), expected) + calls = [] + class Session: + def request(self, *args, **kwargs): + calls.append(1); raise RuntimeError(SENTINEL) + with pytest.raises(b.PlanError, match="NATIVE_WRITE_UNCERTAIN") as error: + b.replace_native_service(request, expected, session=Session()) + assert len(calls) == 1 and SENTINEL not in str(error.value) + + +def ready_native(before, request): + after = copy.deepcopy(before) + after["revision"]["metadata"].update(request["spec"]["template"]["metadata"]) + after["revision"]["metadata"].update(uid="new-uid", resourceVersion="new-rv") + after["revision"]["spec"] = copy.deepcopy(request["spec"]["template"]["spec"]) + after["revision"]["status"] = {"imageDigest": NEW_IMAGE, "observedGeneration": 1, + "conditions": [{"type": k, "status": "True"} for k in ("Ready", "ContainerHealthy")]} + after["service"]["metadata"]["resourceVersion"] = "after-write-rv" + after["service"]["status"]["latestCreatedRevisionName"] = request["spec"]["template"]["metadata"]["name"] + return after + + +def test_native_publisher_reads_exact_candidate_and_waits_only_control_plane(native_case): + b, before, expected = native_case + request, _ = b.build_native_plan(before, copy.deepcopy(before), expected) + after = ready_native(before, request) + pending = copy.deepcopy(after); pending["revision"]["status"]["conditions"] = [] + captures, writes, pauses = [], [], [] + states = iter([copy.deepcopy(before), pending, after]) + def capture(identity, **kwargs): + captures.append(identity["baseline_revision"]); return next(states) + result = b.publish_native(before, expected, read=lambda cmd: [], capture=capture, + replace=lambda req, exp: writes.append(req), pause=pauses.append) + assert captures == [BASE, CANDIDATE, CANDIDATE] + assert len(writes) == 1 and pauses == [15] + assert result["startup"] == "CONTROL_PLANE_TCP_READY_ONLY" + + +@pytest.mark.parametrize("failure", ["conflict", "already-exists", "failed-startup", "pending-bound", "tag-changed"]) +def test_native_publisher_never_redeploys_on_conflict_or_failed_readback(native_case, failure): + b, before, expected = native_case + request, _ = b.build_native_plan(before, copy.deepcopy(before), expected) + after = ready_native(before, request) + if failure == "failed-startup": after["revision"]["status"]["conditions"][0]["status"] = "False" + if failure == "pending-bound": after["revision"]["status"]["conditions"] = [] + if failure == "tag-changed": after["service"]["spec"]["traffic"][1]["tag"] = "changed" + states = iter([copy.deepcopy(before), after]); writes = [] + def replace(req, exp): + writes.append(req) + if failure == "conflict": raise b.PlanError("NATIVE_RESOURCE_VERSION_CONFLICT") + with pytest.raises(b.PlanError): + b.publish_native(before, expected, read=lambda cmd: ([{}] if failure == "already-exists" else []), + capture=lambda *a, **kw: next(states), replace=replace, pause=lambda _: None, attempts=1) + assert len(writes) == (0 if failure == "already-exists" else 1) + + +def fake_overlay_git(b, *, omit=None, extra=None, symlink=False, changed_pin=False): + paths = sorted(b._OVERLAY_REQUIRED - ({omit} if omit else set())) + if extra: paths.append(extra) + def git(*args): + if args == ("rev-parse", "HEAD"): return b.NOTIFICATION_SOURCE.encode() + if args[0] == "merge-base": return b._TCP_00781_SOURCE.encode() + if args[0] == "diff": return "\n".join("M\t" + p for p in paths).encode() + if args[0] == "ls-tree": return (("120000" if symlink else "100644") + " blob " + "a" * 40 + "\t" + args[-1]).encode() + assert args[0] == "show" + source, name = args[1].split(":", 1) + return (("bad" if changed_pin and source == b.NOTIFICATION_SOURCE else "dependency") if name in b._DEPENDENCY_HASHES else "synthetic " + name).encode() + return git + + +def test_overlay_exact_git_blobs_assets_and_no_install_commands(native_case, monkeypatch, tmp_path): + b, _, _ = native_case + monkeypatch.setattr(b, "_DEPENDENCY_HASHES", {n: __import__('hashlib').sha256(b"dependency").hexdigest() for n in b._DEPENDENCY_HASHES}) + out = tmp_path / "context" + result = b.prepare_notification_overlay(tmp_path, out, git=fake_overlay_git(b)) + recipe = (out / "Dockerfile").read_text() + assert recipe.startswith("FROM " + OLD_IMAGE + "\n") + assert all(line.startswith("COPY [") for line in recipe.splitlines()[1:]) + assert set(result["files_sha256"]) == b._OVERLAY_REQUIRED + assert (out / "notifications/display_names.json").exists() + assert (out / "notifications/digest_snapshot/provenance.json").exists() + assert (out / "scripts/execution_report_heartbeat.py").exists() + assert not (out / ".venv").exists() and not (out / "gha-creds-synthetic.json").exists() + + +@pytest.mark.parametrize("case", ["missing-asset", "extra-runtime", "symlink", "pin-changed"]) +def test_overlay_rejects_before_writing_context(native_case, monkeypatch, tmp_path, case): + b, _, _ = native_case + monkeypatch.setattr(b, "_DEPENDENCY_HASHES", {n: __import__('hashlib').sha256(b"dependency").hexdigest() for n in b._DEPENDENCY_HASHES}) + git = fake_overlay_git(b, omit="notifications/display_names.json" if case == "missing-asset" else None, + extra="application/unreviewed.py" if case == "extra-runtime" else None, + symlink=case == "symlink", changed_pin=case == "pin-changed") + out = tmp_path / "context" + with pytest.raises(b.PlanError): b.prepare_notification_overlay(tmp_path, out, git=git) + assert not out.exists() + + +def test_native_numeric_project_namespace_is_preserved_not_rewritten(native_case): + b, before, expected = native_case + for part in ("service", "revision"): before[part]["metadata"]["namespace"] = "123456789" + request, _ = b.build_native_plan(before, copy.deepcopy(before), expected) + assert request["metadata"]["namespace"] == "123456789" diff --git a/tests/test_safe_no_traffic_deploy_workflow.sh b/tests/test_safe_no_traffic_deploy_workflow.sh index 1d03169..2b9fae7 100644 --- a/tests/test_safe_no_traffic_deploy_workflow.sh +++ b/tests/test_safe_no_traffic_deploy_workflow.sh @@ -93,3 +93,51 @@ if names != {"Dockerfile", "tracked.txt"}: raise SystemExit("FAIL: generated authentication file is not excluded by the build boundary") print("PASS: tracked-source container build excludes generated authentication files") PY + +# Execute the actual mode selector with no connected services. Mixed-operation +# approvals must stop before checkout/auth; default HTTP remains available. +python3 - "$workflow" <<'PY' +import json, os, subprocess, sys, tempfile +from pathlib import Path +text = Path(sys.argv[1]).read_text() +# Ruby is already required by this test. Use JSON to avoid a new Python package. +wf = json.loads(subprocess.check_output(["ruby", "-rjson", "-ryaml", "-e", "puts JSON.generate(YAML.load_file(ARGV[0]))", sys.argv[1]])) +steps = wf['jobs']['sync-cloud-run-env']['steps'] +config_step = next(x for x in steps if x.get('id') == 'config') +config = config_step['run'] +# Notification dispatch inputs stay local to their steps, preserving the +# existing sync/drift job-env mirror and actual candidate input forwarding. +job_env = wf['jobs']['sync-cloud-run-env']['env'] +assert 'NOTIFICATION_ONLY' not in job_env and 'NOTIFICATION_CANDIDATE_REVISION' not in job_env +assert config_step['env']['NOTIFICATION_ONLY'] == '${{ inputs.notification_only }}' +assert config_step['env']['NOTIFICATION_CANDIDATE_REVISION'] == '${{ inputs.notification_candidate_revision }}' +for step in [x for x in steps if x.get('id') == 'notification_image' or x.get('name','').startswith('Publish one native')]: + assert step['env']['NOTIFICATION_CANDIDATE_REVISION'] == '${{ inputs.notification_candidate_revision }}' +for conflicting in [None, 'APPROVE_ENV_SECRET_SYNC', 'APPROVE_SCHEDULER_IAM_SYNC', 'APPROVE_TRAFFIC_SHIFT', 'APPROVE_CLEANUP']: + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / 'output' + env = dict(os.environ, GITHUB_OUTPUT=str(output), NOTIFICATION_ONLY='true', + NOTIFICATION_CANDIDATE_REVISION='charles-schwab-quant-service-notification-test', + GITHUB_EVENT_NAME='workflow_dispatch', ENABLE_MAIN_PUSH_CLOUD_RUN_AUTOMATION='false', ENABLE_GITHUB_CLOUD_RUN_DEPLOY='true', + APPROVE_IMAGE_DEPLOY='true', ENABLE_GITHUB_ENV_SYNC='true', CLOUD_RUN_CLEANUP_ENABLED='true') + for key in ['APPROVE_ENV_SECRET_SYNC', 'APPROVE_SCHEDULER_IAM_SYNC', 'APPROVE_TRAFFIC_SHIFT', 'APPROVE_CLEANUP']: + env[key] = str(key == conflicting).lower() + result = subprocess.run(['bash', '-c', config], env=env, capture_output=True) + assert (result.returncode == 0) == (conflicting is None) + if conflicting is None: + values = dict(line.split('=', 1) for line in output.read_text().splitlines()) + assert values['notification_only'] == values['deploy_enabled'] == 'true' + assert all(values[k] == 'false' for k in ['env_sync_enabled','scheduler_iam_sync_enabled','traffic_shift_enabled','cleanup_enabled']) +source = next(x for x in steps if x.get('name') == 'Checkout reviewed notification source') +assert source['with']['ref'] == 'e159e6cc07f9a9eea417b810674f934eb5291e6f' +assert source['with']['path'] == 'notification-source' +build = next(x['run'] for x in steps if x.get('id') == 'notification_image') +assert 'prepare-overlay' in build and '--lockfile", "notification-source/uv.lock"' in build +assert 'git archive HEAD' not in build and 'uv sync' not in build and 'pip install' not in build +publish = next(x['run'] for x in steps if x.get('name','').startswith('Publish one native')) +assert 'native-publish' in publish and '--baseline-revision=charles-schwab-quant-service-00781-lel' in publish +assert all(term not in publish for term in ['gcloud run deploy', 'services replace', '/probe', '/run', 'curl ']) +for name in ['Install strategy status dependencies', 'Capture no-traffic deployment baseline', 'Build, push, and deploy Cloud Run image', 'Verify no-traffic deployment readback']: + assert 'inputs.notification_only != true' in next(x['if'] for x in steps if x.get('name') == name) +print('PASS: native selector rejects four mixed approvals; exact source and separate HTTP/native paths verified') +PY diff --git a/tests/test_verify_cloud_run_no_traffic_deploy.py b/tests/test_verify_cloud_run_no_traffic_deploy.py index 4287813..335dd76 100644 --- a/tests/test_verify_cloud_run_no_traffic_deploy.py +++ b/tests/test_verify_cloud_run_no_traffic_deploy.py @@ -791,3 +791,201 @@ def test_default_http_failure_keeps_original_precedence_over_volume_semantics(ca trap_all_literals(case) with pytest.raises(check.PreflightError, match="^HTTP_STARTUP_GUARD_REQUIRED$"): check._preflight_local(*case) + + +def native_snapshot(): + traffic = [{"revisionName": BASE_REV, "percent": 100, "tag": "stable"}, + {"revisionName": "demo-service-old", "tag": "keep-zero"}] + meta = dict(name="demo-service", namespace="demo-project", uid="service-uid", resourceVersion="service-rv", generation=1) + spec = dict(serviceAccountName="demo@example.invalid", containers=[{ + "image": BASE_IMAGE, "env": [{"name": "SCHWAB_CASH_CURRENCY", "value": "USD"}], + "ports": [{"containerPort": 8080}], "startupProbe": {"tcpSocket": {"port": 8080}}, + }]) + return {"service": {"metadata": meta, "spec": {"traffic": traffic}, + "status": {"traffic": copy.deepcopy(traffic), "observedGeneration": 1, "latestCreatedRevisionName": BASE_REV}}, + "revision": {"metadata": dict(meta, name=BASE_REV, uid="revision-uid", resourceVersion="revision-rv"), + "spec": spec, "status": {"imageDigest": BASE_IMAGE, "observedGeneration": 1, + "conditions": [{"type": k, "status": "True"} for k in ("Ready", "ContainerHealthy")]}}} + + +def test_native_readback_keeps_zero_tags_and_exact_revision(): + before = native_snapshot(); after = copy.deepcopy(before) + template = {"metadata": {"name": CANDIDATE_REV, "labels": {"commit-sha": CANDIDATE_COMMIT}}, + "spec": copy.deepcopy(before["revision"]["spec"])} + template["spec"]["containers"][0]["image"] = CANDIDATE_IMAGE + after["revision"]["metadata"].update(template["metadata"]) + after["service"]["status"]["latestCreatedRevisionName"] = CANDIDATE_REV + after["revision"]["spec"] = template["spec"] + after["revision"]["status"]["imageDigest"] = CANDIDATE_IMAGE + check.verify_native_readback(before, {"metadata": copy.deepcopy(before["service"]["metadata"]), "spec": {"template": template}}, **after) + after["service"]["status"]["traffic"][1]["tag"] = "different" + with pytest.raises(check.PreflightError, match="NATIVE_TRAFFIC_CHANGED"): + check.verify_native_readback(before, {"metadata": copy.deepcopy(before["service"]["metadata"]), "spec": {"template": template}}, **after) + + +@pytest.mark.parametrize("condition", ["pending", "false", "generation", "wrong-image", "wrong-candidate", "sidecar-drift"]) +def test_native_startup_needs_exact_candidate_observed_generation_and_conditions(condition): + before = native_snapshot(); after = copy.deepcopy(before) + request = {"metadata": copy.deepcopy(before["service"]["metadata"]), "spec": {"template": {"metadata": {"name": BASE_REV}, "spec": copy.deepcopy(before["revision"]["spec"])}}} + if condition == "pending": after["revision"]["status"]["conditions"] = [] + elif condition == "false": after["revision"]["status"]["conditions"][0]["status"] = "False" + elif condition == "generation": after["service"]["status"]["observedGeneration"] = 0 + elif condition == "wrong-image": after["revision"]["status"]["imageDigest"] = CANDIDATE_IMAGE + elif condition == "wrong-candidate": after["revision"]["metadata"]["name"] = CANDIDATE_REV + else: after["revision"]["spec"]["containers"][0]["resources"] = {"limits": {"cpu": "2"}} + with pytest.raises(check.PreflightError): check.verify_native_readback(before, request, **after) + + +def test_native_capture_unknown_literal_stops_before_any_value_projection(): + commands = [] + def read(command): + commands.append(command) + fmt = command[-1] + if fmt == "--format=json(metadata.uid,metadata.resourceVersion)": + return {"metadata": {"uid": "u", "resourceVersion": "r"}} + if "env[].name" in fmt: + return {"metadata": {"uid": "u", "resourceVersion": "r"}, + "spec": {"containers": [{"env": [{"name": "PRIVATE_UNKNOWN"}]}]}} + return {} + identity = dict(project="demo-project", region="demo-region", service="demo-service", baseline_revision=BASE_REV) + with pytest.raises(check.PreflightError, match="UNKNOWN_ENV_LITERAL"): + check.capture_native_snapshot(identity, read=read) + assert not any("extract(value)" in cmd[-1] or "json(metadata," in cmd[-1] for cmd in commands) + assert not any("template.spec" in cmd[-1] for cmd in commands) + + +def test_native_capture_unknown_annotation_stops_before_metadata_values(): + commands = [] + def read(command): + commands.append(command) + if command[-1] == "--format=json(metadata.uid,metadata.resourceVersion)": + return {"metadata": {"uid": "u", "resourceVersion": "r"}} + if "annotations.list(show=keys)" in command[-1]: + return {"metadata": {"annotations": "kubectl.kubernetes.io/last-applied-configuration"}} + return {} + identity = dict(project="demo-project", region="demo-region", service="demo-service", baseline_revision=BASE_REV) + with pytest.raises(check.PreflightError, match="NATIVE_UNSUPPORTED_ANNOTATION"): + check.capture_native_snapshot(identity, read=read) + assert not any("json(metadata," in cmd[-1] for cmd in commands) + + +def test_native_capture_unknown_config_field_stops_name_inventory(): + def read(command): + if command[-1] == "--format=json(metadata.uid,metadata.resourceVersion)": + return {"metadata": {"uid": "u", "resourceVersion": "r"}} + return {"spec": "containers,unsupported"} if command[-1] == "--format=json(spec.list(show=keys))" else {} + identity = dict(project="demo-project", region="demo-region", service="demo-service", baseline_revision=BASE_REV) + with pytest.raises(check.PreflightError, match="NATIVE_UNSUPPORTED_CONFIGURATION_FIELD"): + check.capture_native_snapshot(identity, read=read) + + +@pytest.mark.parametrize("drift", [None, "revision", "service"]) +def test_native_capture_value_projection_is_allowlisted_and_uid_rv_bracketed(drift): + expected = native_snapshot(); commands = []; stamp_calls = {"service": 0, "revision": 0} + def read(command): + commands.append(command) + fmt = command[-1] + revision = command[2] == "revisions" + obj = expected["revision" if revision else "service"] + if fmt == "--format=json(metadata.uid,metadata.resourceVersion)": + meta = {k: obj["metadata"][k] for k in ("uid", "resourceVersion")} + part = "revision" if revision else "service" + stamp_calls[part] += 1 + if drift == part and stamp_calls[part] > 1: meta["resourceVersion"] += "changed" + return {"metadata": meta} + if "status.observedGeneration" in fmt: + result = copy.deepcopy(obj) + if revision: del result["spec"]["containers"][0]["env"][0]["value"] + return result + if "env[].name" in fmt: + return {"metadata": {k: obj["metadata"][k] for k in ("uid", "resourceVersion")}, + "spec": {"containers": [{"env": [{"name": "SCHWAB_CASH_CURRENCY"}]}]}} + if "extract(value)" in fmt: + assert 'filter("name=SCHWAB_CASH_CURRENCY")' in fmt + return {"spec": {"containers": [{"env": [["USD"]]}]}} + return {} + identity = dict(project="demo-project", region="demo-region", service="demo-service", baseline_revision=BASE_REV) + if drift: + with pytest.raises(check.PreflightError, match="NATIVE_OBJECT_CHANGED_DURING_CAPTURE"): + check.capture_native_snapshot(identity, read=read) + else: assert check.capture_native_snapshot(identity, read=read) == expected + assert not any("template.spec" in c[-1] or "env[].value," in c[-1] for c in commands) + + +@pytest.mark.parametrize("part", ["service", "revision"]) +def test_native_annotation_race_is_bound_to_earliest_stamp(part): + source = native_snapshot(); commands = [] + def read(command): + commands.append(command); kind = "revision" if command[2] == "revisions" else "service" + obj = source[kind]; fmt = command[-1] + if fmt == "--format=json(metadata.uid,metadata.resourceVersion)": + return {"metadata": {k: obj["metadata"][k] for k in ("uid", "resourceVersion")}} + if "env[].name" in fmt and "status.observedGeneration" not in fmt: + stamp = {k: obj["metadata"][k] for k in ("uid", "resourceVersion")} + if part == "revision": stamp["resourceVersion"] += "changed" + return {"metadata": stamp, "spec": {"containers": [{"env": [{"name": "SCHWAB_CASH_CURRENCY"}]}]}} + if "status.observedGeneration" in fmt: + assert "json(metadata," not in fmt and "metadata.annotations," not in fmt + result = copy.deepcopy(obj) + if kind == part: result["metadata"]["resourceVersion"] += "changed" + if kind == "revision": del result["spec"]["containers"][0]["env"][0]["value"] + return result + return {} + identity = dict(project="demo-project", region="demo-region", service="demo-service", baseline_revision=BASE_REV) + with pytest.raises(check.PreflightError, match="NATIVE_OBJECT_CHANGED_DURING_CAPTURE"): + check.capture_native_snapshot(identity, read=read) + + +@pytest.mark.parametrize("drift", ["service-ingress", "extra-runtime-annotation", "other-latest-candidate"]) +def test_native_readback_complete_metadata_and_observed_service_candidate(drift): + before = native_snapshot(); after = copy.deepcopy(before) + request = {"metadata": copy.deepcopy(before["service"]["metadata"]), "spec": {"template": { + "metadata": {"name": BASE_REV}, "spec": copy.deepcopy(before["revision"]["spec"])}}} + if drift == "service-ingress": after["service"]["metadata"]["annotations"] = {"run.googleapis.com/ingress": "all"} + elif drift == "extra-runtime-annotation": after["revision"]["metadata"]["annotations"] = {"run.googleapis.com/vpc-access-egress": "all-traffic"} + else: + after["service"]["metadata"].update(generation=2, resourceVersion="later-rv") + after["service"]["status"].update(observedGeneration=2, latestCreatedRevisionName="demo-service-other-writer") + with pytest.raises(check.PreflightError): check.verify_native_readback(before, request, **after) + + +def test_native_capture_output_created_private_and_never_overwritten(monkeypatch, tmp_path): + out = tmp_path / "before.json" + monkeypatch.setattr(check, "capture_native_snapshot", lambda _: {"synthetic": True}) + monkeypatch.setattr(sys, "argv", [str(SCRIPT), "capture-native", "--project=demo", "--region=demo", "--service=demo", + "--baseline-revision=demo-base", "--output=" + str(out)]) + assert check.main() == 0 + assert out.stat().st_mode & 0o777 == 0o600 + original = out.read_bytes() + assert check.main() == 1 and out.read_bytes() == original + + +def test_native_observed_generation_boolean_is_not_an_integer(): + before = native_snapshot() + before["service"]["status"]["observedGeneration"] = True + with pytest.raises(check.PreflightError, match="INVALID_STRUCTURE"): + check.native_snapshot_inventory(before) + + +@pytest.mark.parametrize("field", ["command", "args"]) +def test_native_empty_overrides_preserved_without_text_projection(field): + expected = native_snapshot(); expected["revision"]["spec"]["containers"][0][field] = [] + def read(command): + obj = expected["revision" if command[2] == "revisions" else "service"]; fmt = command[-1] + if fmt == "--format=json(metadata.uid,metadata.resourceVersion)": + return {"metadata": {k: obj["metadata"][k] for k in ("uid", "resourceVersion")}} + if f".{field}.len()" in fmt: return {"spec": {"containers": [{"image": BASE_IMAGE, field: 0}]}} + if "status.observedGeneration" in fmt: + assert "spec.containers[].command[]" not in fmt and "spec.containers[].args[]" not in fmt + result = copy.deepcopy(obj) + if command[2] == "revisions": + result["spec"]["containers"][0].pop(field) + result["spec"]["containers"][0]["env"][0].pop("value") + return result + if "env[].name" in fmt: + return {"metadata": {k: obj["metadata"][k] for k in ("uid", "resourceVersion")}, + "spec": {"containers": [{"env": [{"name": "SCHWAB_CASH_CURRENCY"}]}]}} + if "extract(value)" in fmt: return [["USD"]] + return {} + identity = dict(project="demo-project", region="demo-region", service="demo-service", baseline_revision=BASE_REV) + assert check.capture_native_snapshot(identity, read=read) == expected