velero: extract shared replay_collector_blobs helper

The "parse each collector JSON blob, replay it, raise on bad JSON" loop
was still duplicated between plugin/pytest.py's dd_agent_check and
kube_discovery.py's replay_collector_output, right down to the identical
error string — the neighboring flags_to_argv/find_collector_blobs
extraction stopped one step short of this one. Extract it into
replay_collector_blobs() in _env.py, tighten the bare raise Exception
into raise ValueError(...) from e so the original JSONDecodeError
traceback survives, and add a regression test that feeds a malformed
JSON blob through run_discovery_check_kubernetes.

Environment: Datadog workspace

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: vitkyrka

datadog_checks_dev/datadog_checks/dev/_env.py

@@ -132,6 +132,16 @@ def find_collector_blobs(output: str) -> list[str]:
     return re.findall(JSON_COLLECTOR_PATTERN, output, re.DOTALL)
 
 
+def replay_collector_blobs(matches: list[str], stub_aggregator: Any, stub_agent: Any) -> None:
+    """Parse and replay each collector JSON blob found by ``find_collector_blobs``."""
+    for raw_json in matches:
+        try:
+            collector = json.loads(raw_json)
+        except Exception as e:
+            raise ValueError(f'Error loading json: {e}\nCollector Json Output:\n{raw_json}') from e
+        replay_check_run(collector, stub_aggregator, stub_agent)
+
+
 def replay_check_run(agent_collector, stub_aggregator, stub_agent):
     errors = []
     for collector in agent_collector:

datadog_checks_dev/datadog_checks/dev/kube_discovery.py

@@ -34,7 +34,7 @@
     flags_to_argv,
     format_config,
     get_state,
-    replay_check_run,
+    replay_collector_blobs,
     save_state,
     set_up_env,
 )
@@ -235,12 +235,7 @@ def get_kube_discovery_state() -> dict[str, Any]:
 def replay_collector_output(output: str, aggregator: Any, datadog_agent: Any) -> int:
     """Replay collector JSON blobs from ``agent check --json`` output."""
     matches = find_collector_blobs(output)
-    for raw_json in matches:
-        try:
-            collector = json.loads(raw_json)
-        except Exception as e:
-            raise Exception(f'Error loading json: {e}\nCollector Json Output:\n{raw_json}')
-        replay_check_run(collector, aggregator, datadog_agent)
+    replay_collector_blobs(matches, aggregator, datadog_agent)
 
     return len(matches)
 

datadog_checks_dev/datadog_checks/dev/plugin/pytest.py

@@ -25,7 +25,7 @@
     format_config,
     get_env_vars,
     get_state,
-    replay_check_run,
+    replay_collector_blobs,
     save_state,
     serialize_data,
     set_up_env,
@@ -219,12 +219,7 @@ def run_check(config=None, **kwargs):
             message_parts.append(result.stdout + result.stderr)
             raise ValueError('{}\nCould not find valid check output'.format('\n'.join(message_parts)))
 
-        for raw_json in matches:
-            try:
-                collector = json.loads(raw_json)
-            except Exception as e:
-                raise Exception("Error loading json: {}\nCollector Json Output:\n{}".format(e, raw_json))
-            replay_check_run(collector, aggregator, datadog_agent)
+        replay_collector_blobs(matches, aggregator, datadog_agent)
 
         return aggregator
 

datadog_checks_dev/tests/test_kube_discovery.py

@@ -264,6 +264,16 @@ def test_no_json_output_raises(self, e2e_mode):
                 with pytest.raises(ValueError, match='Could not find valid check output'):
                     run_discovery_check_kubernetes(mock.Mock(), mock.Mock())
 
+    def test_malformed_json_output_raises(self, e2e_mode):
+        save_kube_discovery_state('/tmp/kubeconfig')
+
+        stdout = 'preamble\n[ not valid json\n]\n'
+
+        with mock.patch('datadog_checks.dev.kube_discovery.run_command', return_value=result(stdout=stdout)):
+            with mock.patch('datadog_checks.dev.kube_discovery.find_check_root', return_value='/root/test_check'):
+                with pytest.raises(ValueError, match='Error loading json'):
+                    run_discovery_check_kubernetes(mock.Mock(), mock.Mock())
+
 
 class TestAssertAllDiscoveryCandidatesStableKubernetes:
     def test_skips_outside_e2e(self, monkeypatch):