Scrub New Relic ingestion keys from test recordings

src/new-relic/azext_new_relic/tests/latest/recording_processors.py

@@ -0,0 +1,38 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import re
+
+from azure.cli.testsdk.scenario_tests import RecordingProcessor
+from azure.cli.testsdk.scenario_tests.utilities import is_text_payload
+
+
+MOCK_INGESTION_KEY = "fake-ingestion-key"
+_INGESTION_KEY_RE = re.compile(
+    r'("ingestionKey"\s*:\s*")[^"]+(")',
+    re.IGNORECASE,
+)
+
+
+class NewRelicSecretScrubber(RecordingProcessor):
+
+    def process_request(self, request):
+        return self._scrub(request)
+
+    def process_response(self, response):
+        return self._scrub(response)
+
+    @staticmethod
+    def _scrub(entity):
+        body = entity.get("body")
+        if not isinstance(body, dict):
+            return entity
+
+        if is_text_payload(entity) and body.get("string"):
+            entity["body"]["string"] = _INGESTION_KEY_RE.sub(
+                rf"\g<1>{MOCK_INGESTION_KEY}\g<2>",
+                entity["body"]["string"],
+            )
+        return entity

src/new-relic/azext_new_relic/tests/latest/recordings/test_new_relic_monitor.yaml

@@ -1327,7 +1327,7 @@ interactions:
     uri: https://management.azure.com/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/cli_test_new_relic_monitor000001/providers/NewRelic.Observability/monitors/test-new-relic-monitor/vmHostPayloads?api-version=2026-06-01
   response:
     body:
-      string: '{"ingestionKey":"feafe32045caa054d172d9b7942a38d89a43NRAL"}'
+      string: '{"ingestionKey":"fake-ingestion-key"}'
     headers:
       cache-control:
       - no-cache

src/new-relic/azext_new_relic/tests/latest/test_new_relic.py

@@ -8,9 +8,17 @@
 from azure.cli.testsdk.scenario_tests import AllowLargeResponse
 from azure.cli.testsdk import ScenarioTest, ResourceGroupPreparer
 
+from .recording_processors import NewRelicSecretScrubber
+
 
 class NewRelicScenario(ScenarioTest):
 
+    def __init__(self, method_name):
+        super().__init__(
+            method_name,
+            recording_processors=[NewRelicSecretScrubber()],
+        )
+
     @AllowLargeResponse(size_kb=10240)
     @ResourceGroupPreparer(name_prefix='cli_test_new_relic_monitor')
     def test_new_relic_monitor(self, resource_group):

src/new-relic/azext_new_relic/tests/latest/test_recording_processors.py

@@ -0,0 +1,33 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from .recording_processors import MOCK_INGESTION_KEY, NewRelicSecretScrubber
+
+
+def test_new_relic_secret_scrubber():
+    processor = NewRelicSecretScrubber()
+    expected = f'{{"ingestionKey":"{MOCK_INGESTION_KEY}"}}'
+
+    for process in (processor.process_request, processor.process_response):
+        entity = {
+            "headers": {"content-type": ["application/json"]},
+            "body": {"string": '{"ingestionKey":"live-key-value"}'},
+        }
+
+        scrubbed = process(entity)
+
+        assert scrubbed["body"]["string"] == expected
+
+
+def test_new_relic_secret_scrubber_ignores_unsupported_body():
+    processor = NewRelicSecretScrubber()
+
+    for process in (processor.process_request, processor.process_response):
+        for entity in (
+            {"headers": {}},
+            {"headers": {}, "body": None},
+            {"headers": {}, "body": "not-a-dictionary"},
+        ):
+            assert process(entity) == entity