fix(purl): preserve error stream records

Author: lelia

socketdev/purl/__init__.py

@@ -96,17 +96,25 @@ def post(
         path += params
         response = self.api.do_request(path=path, payload=purls, method="POST")
         if response.status_code == 200:
-            purl = []
+            artifact_rows = []
+            stream_records = []
             result = response.text
             result = result.strip('"').strip()
             for line in result.split("\n"):
                 if line and line != '"':
                     try:
                         item = json.loads(line)
-                        purl.append(item)
+                        if isinstance(item, dict) and item.get("_type") in {
+                            "purlError",
+                            "summary",
+                        }:
+                            stream_records.append(item)
+                        else:
+                            artifact_rows.append(item)
                     except json.JSONDecodeError:
                         continue
-            purl_deduped = Dedupe.dedupe(purl, batched=True)
+            purl_deduped = Dedupe.dedupe(artifact_rows, batched=True)
+            purl_deduped.extend(stream_records)
             if strict:
                 self._raise_on_missing(components, purl_deduped)
             return purl_deduped
@@ -120,8 +128,8 @@ def _raise_on_missing(components: list, results: list) -> None:
         """Raise APIPartialResponse if any requested component purl is absent from results.
 
         Only components exposing a ``purl`` string are checked; the batch API echoes the
-        request identifier back as ``inputPurl`` (falling back to ``purl``), so we compare
-        against both.
+        request identifier back as ``inputPurl`` (falling back to ``purl``), including
+        under ``value`` for typed ``purlError`` stream records.
         """
         requested = [
             c["purl"]
@@ -138,6 +146,11 @@ def _raise_on_missing(components: list, results: list) -> None:
                 value = row.get(field)
                 if isinstance(value, str):
                     returned.add(value)
+            record_value = row.get("value")
+            if isinstance(record_value, dict):
+                input_purl = record_value.get("inputPurl")
+                if isinstance(input_purl, str):
+                    returned.add(input_purl)
         missing = [purl for purl in requested if purl not in returned]
         if missing:
             raise APIPartialResponse(

tests/unit/test_all_endpoints_unit.py

@@ -471,6 +471,28 @@ def test_purl_post_synthetic_pending_scan_row(self):
         self.assertIsNone(alert["severity"])
         self.assertIsNone(alert["action"])
 
+    def test_purl_post_preserves_purl_error_record(self):
+        """purlError stream records bypass artifact deduplication."""
+        error_row = {
+            "_type": "purlError",
+            "value": {
+                "error": "package_not_found",
+                "inputPurl": "pkg:npm/missing@1.0.0",
+            },
+        }
+        self._mock_purl_ndjson(json.dumps(error_row))
+
+        result = self.sdk.purl.post(
+            components=[{"purl": "pkg:npm/missing@1.0.0"}],
+            org_slug="test-org",
+            purl_errors=True,
+            strict=True,
+        )
+
+        self.assertEqual(result, [error_row])
+        url = self.mock_requests.request.call_args[0][1]
+        self.assertIn("purlErrors=true", url)
+
     def test_purl_post_strict_raises_on_missing(self):
         """strict=True raises APIPartialResponse listing purls dropped from the response."""
         from socketdev.exceptions import APIPartialResponse