test(e2e): cover decrypt_fields consume policy lifecycle

.github/workflows/e2e-replay.yaml

@@ -52,11 +52,15 @@ jobs:
         env:
           SOURCE_RUN: ${{ inputs.source_run }}
           REPLAY_MODE: ${{ inputs.mode }}
+          REPLAY_SCENARIO: ${{ inputs.scenario }}
         run: |
           if [ -n "$SOURCE_RUN" ]; then
             [[ "$SOURCE_RUN" =~ ^[0-9]+$ ]] && [ "$REPLAY_MODE" = replay ]
           fi
           make setup-e2e-replay
+          if [ "$REPLAY_MODE" = record ] || [ -n "$SOURCE_RUN" ]; then
+            export KONGCTL_REPLAY_REFRESH_SCENARIO="$REPLAY_SCENARIO"
+          fi
           python3 -m unittest discover -s scripts -p e2e_replay_test.py
       - name: Build experiment binaries
         run: make build-e2e-replay

scripts/e2e_replay.py

@@ -426,7 +426,7 @@ def request_key(endpoint, method, target, data):
             "query": [list(pair) for pair in sorted(parse_qsl(url.query, keep_blank_values=True))], "body": body}
 
 
-def validate_cassette(cassette, directory, scenario=SCENARIO):
+def validate_cassette(cassette, directory, scenario=SCENARIO, *, allow_stale=False):
     check_eligibility(directory)
     fixtures = fixture_strings(directory)
     fields = {"schema_version", "scenario", "inputs_sha256", "source", "interactions"}
@@ -436,7 +436,7 @@ def validate_cassette(cassette, directory, scenario=SCENARIO):
         raise ValueError("invalid cassette fields")
     if type(cassette["schema_version"]) is not int or cassette["schema_version"] not in (1, 2) or cassette["scenario"] != scenario:
         raise ValueError("unsupported cassette schema or scenario")
-    if cassette["inputs_sha256"] != scenario_digest(directory):
+    if not allow_stale and cassette["inputs_sha256"] != scenario_digest(directory):
         raise ValueError(f"stale cassette: {scenario}; re-record and review, or remove replay eligibility")
     source = cassette["source"]
     if not isinstance(source, dict) or not isinstance(source.get("commit"), str) or not re.fullmatch(r"[0-9a-f]{40}", source["commit"]):

scripts/e2e_replay_test.py

@@ -5,6 +5,7 @@
 import importlib.util
 import json
 import os
+import re
 from pathlib import Path
 import ssl
 import tempfile
@@ -22,7 +23,38 @@ def interaction(method="GET", body=None):
             "response": {"status": 200, "body": {"data": []}}}
 
 
+def cassette_refresh_scenario(env):
+    """Only the manual candidate workflow may defer one old input fingerprint."""
+    scenario = env.get("KONGCTL_REPLAY_REFRESH_SCENARIO")
+    if not scenario:
+        return None
+    mode, source = env.get("REPLAY_MODE"), env.get("SOURCE_RUN", "")
+    if (env.get("GITHUB_EVENT_NAME") != "workflow_dispatch"
+            or env.get("GITHUB_WORKFLOW") != "E2E replay experiment"
+            or scenario not in MODULE.SCENARIOS
+            or scenario != env.get("REPLAY_SCENARIO")
+            or not (mode == "record" and not source
+                    or mode == "replay" and re.fullmatch(r"[0-9]+", source))):
+        raise ValueError("cassette refresh requires the selected manual recording or source-run replay")
+    return scenario
+
+
 class ReplayTest(unittest.TestCase):
+    def test_cassette_refresh_is_scoped_to_manual_candidate_workflow(self):
+        env = {"GITHUB_EVENT_NAME": "workflow_dispatch", "GITHUB_WORKFLOW": "E2E replay experiment",
+               "REPLAY_MODE": "record", "REPLAY_SCENARIO": MODULE.SCENARIO,
+               "KONGCTL_REPLAY_REFRESH_SCENARIO": MODULE.SCENARIO}
+        self.assertIsNone(cassette_refresh_scenario({}))
+        self.assertEqual(MODULE.SCENARIO, cassette_refresh_scenario(env))
+        self.assertEqual(MODULE.SCENARIO, cassette_refresh_scenario(
+            {**env, "REPLAY_MODE": "replay", "SOURCE_RUN": "123"}))
+        for changes in [{"GITHUB_EVENT_NAME": "pull_request"}, {"GITHUB_WORKFLOW": "CI Test"},
+                        {"REPLAY_SCENARIO": "portal/sync"}, {"KONGCTL_REPLAY_REFRESH_SCENARIO": "unknown"},
+                        {"REPLAY_MODE": "replay"}, {"SOURCE_RUN": "123"},
+                        {"REPLAY_MODE": "replay", "SOURCE_RUN": "not-a-run"}]:
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                cassette_refresh_scenario({**env, **changes})
+
     def test_assertion_fields_are_data_but_environment_controls_stay_restricted(self):
         import yaml
         with tempfile.TemporaryDirectory() as temporary:
@@ -346,11 +378,13 @@ def test_recording_rejects_echoed_credentials(self):
 
     def test_repository_cassettes_are_current_even_without_replay_routing(self):
         root = MODULE.ROOT / "test/e2e/scenarios"
+        refresh = cassette_refresh_scenario(os.environ)
         for path in sorted(root.glob("**/replay/cassette.json")):
             directory = path.parent.parent
+            scenario = directory.relative_to(root).as_posix()
             with self.subTest(path=path):
                 MODULE.validate_cassette(MODULE.load_cassette(path), directory,
-                                        directory.relative_to(root).as_posix())
+                                        scenario, allow_stale=scenario == refresh)
 
     def test_external_dependencies_require_explicit_review(self):
         with tempfile.TemporaryDirectory() as directory:
@@ -645,6 +679,16 @@ def test_cassette_schema_and_staleness(self):
             (root / "scenario.yaml").write_text("baseInputsPath: testdata\nsteps: [{name: changed}]")
             with self.assertRaisesRegex(ValueError, "stale cassette"):
                 MODULE.validate_cassette(cassette, root)
+            MODULE.validate_cassette(cassette, root, allow_stale=True)
+            invalid = copy.deepcopy(cassette)
+            invalid["interactions"][0]["response"]["body"] = {"token": "private"}
+            with self.assertRaisesRegex(ValueError, "sensitive field"):
+                MODULE.validate_cassette(invalid, root, allow_stale=True)
+            with self.assertRaisesRegex(ValueError, "must be a live recording"):
+                MODULE.validate_cassette(bootstrap, root, allow_stale=True)
+            (root / "scenario.yaml").write_text("baseInputsPath: testdata\nenv: {}\nsteps: []")
+            with self.assertRaisesRegex(ValueError, "environment"):
+                MODULE.validate_cassette(cassette, root, allow_stale=True)
 
     def test_real_https_connect_and_certificate_validation(self):
         engine = MODULE.Replay({"interactions": [interaction()]})

test/e2e/replay-policy.md

@@ -16,7 +16,7 @@ The enabled subset is `control-plane/get`, `control-plane/apply`,
 | apply | [34427742802][apply] |
 | plan/apply-workflow | [34428189915][plan] |
 | sync | [Recording][sync], [isolated phases][sync-replay] |
-| event-gateway/consume-policy | [Recording and isolated replays][consume-record] |
+| event-gateway/consume-policy | [Recording][consume-record], [isolated replays][consume-replay] |
 | portal/sync | [Recording][portal-record], [isolated replays][portal-replay] |
 | portal/visibility | [Recording][visibility-record], [isolated replays][visibility-replay] |
 | portal/api_docs_with_children | [Recording][docs-record], [isolated replays][docs-replay] |
@@ -26,7 +26,8 @@ The enabled subset is `control-plane/get`, `control-plane/apply`,
 [plan]: https://github.com/Kong/kongctl/actions/runs/34428189915
 [sync]: https://github.com/Kong/kongctl/actions/runs/34520312018
 [sync-replay]: https://github.com/Kong/kongctl/actions/runs/34521883600
-[consume-record]: https://github.com/Kong/kongctl/actions/runs/35048315678
+[consume-record]: https://github.com/Kong/kongctl/actions/runs/35115143565
+[consume-replay]: https://github.com/Kong/kongctl/actions/runs/35116004092
 [portal-record]: https://github.com/Kong/kongctl/actions/runs/34517553668
 [portal-replay]: https://github.com/Kong/kongctl/actions/runs/34521886789
 [visibility-record]: https://github.com/Kong/kongctl/actions/runs/34609380360
@@ -80,6 +81,16 @@ the isolated replay job passes. Commit it under the scenario's
 `replay/cassette.json` and add its directory to the sorted policy list. The
 recorder never commits or overwrites reviewed cassettes automatically.
 
+To refresh an enabled scenario, first update its inputs and push the branch,
+then dispatch `mode=record` for that scenario. The manual workflow's build
+tests defer only the selected old cassette's input-fingerprint comparison.
+Its schema, provenance, sanitization and scenario eligibility are still
+checked, as are all other repository cassettes. The same exception applies
+when validating a `source_run` candidate. Normal PR tests, ordinary replay,
+and validation of the newly recorded candidate still require current input
+fingerprints. This lets recording replace stale data without relaxing the
+PR gate or removing the scenario from replay routing.
+
 New recordings use cassette schema v2, which also preserves an allowlisted
 response media type (`application/json` or `application/problem+json`; absent
 is allowed only for an empty body). This is significant: the SDK uses the
@@ -189,14 +200,14 @@ exclude cassette directories or weaken the recorder's sensitive-data checks.
 setup; CI performs it before network isolation. No kongctl dependency changes.
 Dependency installation belongs to job setup, not scenario execution savings.
 
-`event-gateway/consume-policy` passed its complete live recording and three
-isolated replays, each matching all 150 exchanges in strict order. No parallel
-phase annotations or matching exceptions were needed. Scenario execution took
-4.217–4.220 seconds in replay versus a recent 28.61-second live median; this
-is observational, not a measured workflow speedup. Live scenario timing
-already includes its reset; do not add reset savings again. Its scenario-local
-replay README records provenance, same-source live evidence, coverage and
-measurement limitations.
+`event-gateway/consume-policy` passed its expanded live scenario and three
+isolated replays, each matching all 211 exchanges. Two bounded phases cover
+independent schema-validation-parent/static-key creation and cleanup.
+The decrypt_fields child lifecycle and all assertions remain intact. Replay
+scenario execution took 6.855–6.971 seconds; wrappers took 7.565–7.744 seconds.
+Its scenario-local replay README records provenance, phase boundaries and
+measurement limitations. The cassette retains the live source provenance;
+the isolated job's validated artifact adds only the reviewed annotations.
 
 `portal/sync` is enabled after its complete live scenario and three isolated
 replays passed (293 HTTP interactions each), followed by three isolated

test/e2e/scenarios/event-gateway/consume-policy/overlays/008-decrypt-fields-prerequisites/config.yaml

@@ -0,0 +1,50 @@
+_defaults:
+  kongctl:
+    namespace: event-gateways-vc-consume-policy-comprehensive-test
+
+event_gateways:
+  - ref: egw-cp-for-vc-consume-policy-test
+    name: egw-cp-for-vc-consume-policy-test
+    description: "EGW for Virtual Cluster Consume Policy Test"
+    backend_clusters:
+      - ref: default-backend-cluster
+        name: default-backend-cluster
+        description: "Backend Cluster for Test"
+        bootstrap_servers:
+          - "egw-backend-1.example.com:9092"
+          - "egw-backend-2.example.com:9092"
+        authentication:
+          type: anonymous
+        tls:
+          enabled: true
+          insecure_skip_verify: false
+          tls_versions:
+            - tls12
+            - tls13
+    virtual_clusters:
+      - ref: default-virtual-cluster-for-consume-policy-test
+        name: default-virtual-cluster-name-for-consume-policy-test
+        description: "Virtual Cluster for Consume Policy Test"
+        destination:
+          id: !ref default-backend-cluster#id
+        authentication:
+         - type: anonymous
+        acl_mode: enforce_on_gateway
+        dns_label: vc-default
+    static_keys:
+      - ref: static-key-for-consume-policy
+        name: static-key-for-consume-policy-name
+        description: Static Key for Consume Policy Test
+        value: "YXNkZmdoamthc2RmZ2hqa2FzZGZnaGprYXNkZmdoams="
+event_gateway_virtual_cluster_consume_policies:
+  - ref: default-consume-policy
+    name: default-consume-policy-name-for-test
+    description: Default Consume Policy description
+    event_gateway: egw-cp-for-vc-consume-policy-test
+    virtual_cluster: default-virtual-cluster-for-consume-policy-test
+    type: schema_validation
+    enabled: true
+    config:
+      key_validation_action: skip
+      value_validation_action: mark
+      type: json

test/e2e/scenarios/event-gateway/consume-policy/overlays/009-decrypt-fields-policy/config.yaml

@@ -0,0 +1,67 @@
+_defaults:
+  kongctl:
+    namespace: event-gateways-vc-consume-policy-comprehensive-test
+
+event_gateways:
+  - ref: egw-cp-for-vc-consume-policy-test
+    name: egw-cp-for-vc-consume-policy-test
+    description: "EGW for Virtual Cluster Consume Policy Test"
+    backend_clusters:
+      - ref: default-backend-cluster
+        name: default-backend-cluster
+        description: "Backend Cluster for Test"
+        bootstrap_servers:
+          - "egw-backend-1.example.com:9092"
+          - "egw-backend-2.example.com:9092"
+        authentication:
+          type: anonymous
+        tls:
+          enabled: true
+          insecure_skip_verify: false
+          tls_versions:
+            - tls12
+            - tls13
+    virtual_clusters:
+      - ref: default-virtual-cluster-for-consume-policy-test
+        name: default-virtual-cluster-name-for-consume-policy-test
+        description: "Virtual Cluster for Consume Policy Test"
+        destination:
+          id: !ref default-backend-cluster#id
+        authentication:
+         - type: anonymous
+        acl_mode: enforce_on_gateway
+        dns_label: vc-default
+    static_keys:
+      - ref: static-key-for-consume-policy
+        name: static-key-for-consume-policy-name
+        description: Static Key for Consume Policy Test
+        value: "YXNkZmdoamthc2RmZ2hqa2FzZGZnaGprYXNkZmdoams="
+event_gateway_virtual_cluster_consume_policies:
+  - ref: default-consume-policy
+    name: default-consume-policy-name-for-test
+    description: Default Consume Policy description
+    event_gateway: egw-cp-for-vc-consume-policy-test
+    virtual_cluster: default-virtual-cluster-for-consume-policy-test
+    type: schema_validation
+    enabled: true
+    config:
+      key_validation_action: skip
+      value_validation_action: mark
+      type: json
+  - ref: decrypt-fields-consume-policy
+    name: decrypt-fields-consume-policy-name-for-test
+    description: Field-level decryption consume policy
+    event_gateway: egw-cp-for-vc-consume-policy-test
+    virtual_cluster: default-virtual-cluster-for-consume-policy-test
+    type: decrypt_fields
+    enabled: true
+    parent_policy_id: !ref default-consume-policy#id
+    labels:
+      env: staging
+      team: payments
+    config:
+      failure_mode: error
+      key_sources:
+        - type: static
+      decrypt_fields:
+        paths: 'record.value.content["customer.ssn"]'

test/e2e/scenarios/event-gateway/consume-policy/overlays/010-sync-delete-decrypt-fields-prerequisites/config.yaml

@@ -0,0 +1,35 @@
+_defaults:
+  kongctl:
+    namespace: event-gateways-vc-consume-policy-comprehensive-test
+
+event_gateways:
+  - ref: egw-cp-for-vc-consume-policy-test
+    name: egw-cp-for-vc-consume-policy-test
+    description: "EGW for Virtual Cluster Consume Policy Test"
+    static_keys: []
+    backend_clusters:
+      - ref: default-backend-cluster
+        name: default-backend-cluster
+        description: "Backend Cluster for Test"
+        bootstrap_servers:
+          - "egw-backend-1.example.com:9092"
+          - "egw-backend-2.example.com:9092"
+        authentication:
+          type: anonymous
+        tls:
+          enabled: true
+          insecure_skip_verify: false
+          tls_versions:
+            - tls12
+            - tls13
+    virtual_clusters:
+      - ref: default-virtual-cluster-for-consume-policy-test
+        name: default-virtual-cluster-name-for-consume-policy-test
+        description: "Virtual Cluster for Consume Policy Test"
+        destination:
+          id: !ref default-backend-cluster#id
+        authentication:
+         - type: anonymous
+        acl_mode: enforce_on_gateway
+        dns_label: vc-default
+        consume_policies: []