fix profiling test

tests/test_profiling.py

@@ -32,31 +32,29 @@ def setup_library(self):
 
     def test_library(self):
         """ All profiling libraries payload have start and end fields"""
-        requests = list(interfaces.library.get_profiling_data())
-        self._check_requests(requests)
+        interfaces.library.validate_profiling(self._validate_data)
 
     def setup_agent(self):
         self._common_setup()
 
     def test_agent(self):
         """ All profiling agent payload have recording-start and recording-end fields"""
-        requests = list(interfaces.agent.get_profiling_data())
-        self._check_requests(requests)
-
-    def _check_requests(self, requests):
-        assert len(requests) > 0, "No profiling requests"
-
-        # Requests are multipart, and content is a list
-        requests = [r["request"]["content"] for r in requests]
-        requests = [r for r in requests if isinstance(r, list)]
-        # Flatten list
-        requests = [r for sublist in requests for r in sublist]
-
-        requests = [r for r in requests if 'name="event"' in r["headers"].get("Content-Disposition", "")]
-        assert len(requests) > 0, "No profiling event requests"
-        for req in requests:
-            content = req["content"]
-            assert "start" in content, "No start field"
-            assert "end" in content, "No end field"
-            assert re.fullmatch(TIMESTAMP_PATTERN, content["start"])
-            assert re.fullmatch(TIMESTAMP_PATTERN, content["end"])
+        interfaces.agent.validate_profiling(self._validate_data)
+
+    @staticmethod
+    def _validate_data(data):
+        content = data["request"]["content"]
+
+        for part in content:
+            headers = {k.lower(): v for k, v in part["headers"].items()}
+            if 'name="event"' in headers.get("content-disposition", ""):
+                part_content = part["content"]
+
+                assert "start" in part_content, "No start field"
+                assert "end" in part_content, "No end field"
+                assert re.fullmatch(TIMESTAMP_PATTERN, part_content["start"])
+                assert re.fullmatch(TIMESTAMP_PATTERN, part_content["end"])
+
+                return True
+
+        raise ValueError("No profiling event requests")

utils/interfaces/_agent.py

@@ -62,6 +62,9 @@ def assert_use_domain(self, expected_domain):
     def get_profiling_data(self):
         yield from self.get_data(path_filters="/api/v2/profile")
 
+    def validate_profiling(self, validator, success_by_default=False):
+        self.validate(validator, path_filters="/api/v2/profile", success_by_default=success_by_default)
+
     def validate_appsec(self, request, validator):
         for data, payload, chunk, span, appsec_data in self.get_appsec_data(request=request):
             if validator(data, payload, chunk, span, appsec_data):

utils/interfaces/_library/core.py

@@ -372,6 +372,9 @@ def assert_no_skipped_seq_ids(self):
     def get_profiling_data(self):
         yield from self.get_data(path_filters="/profiling/v1/input")
 
+    def validate_profiling(self, validator, success_by_default=False):
+        self.validate(validator, path_filters="/profiling/v1/input", success_by_default=success_by_default)
+
     def assert_trace_exists(self, request, span_type=None):
         for _, _, span in self.get_spans(request=request):
             if span_type is None or span.get("type") == span_type: