Ensure that APIError is thrown on failure to pull image

podman/domain/images_manager.py

@@ -400,6 +400,9 @@ def pull(
             params["compatMode"] = True
             stream = True
 
+        if not params["compatMode"] and not stream:
+            params["quiet"] = True
+
         response = self.client.post("/images/pull", params=params, stream=stream, headers=headers)
         response.raise_for_status(not_found=ImageNotFound)
 
@@ -572,8 +575,30 @@ def _stream_helper(self, response, decode=False):
                         break
                     if reader._fp.chunk_left:
                         data += reader.read(reader._fp.chunk_left)
+                    try:
+                        data_dictionary = json.loads(data)
+                    except json.JSONDecodeError as e:
+                        self._stream_error_helper("Service returned invalid JSON", e.msg)
+                    except UnicodeDecodeError as e:
+                        self._stream_error_helper("Service returned wrongly encoded data", e.msg)
+                    error = data_dictionary.get("error")
+                    if error:
+                        self._stream_error_helper(
+                            "Service returned an error after streaming started", error
+                        )
                     yield data
         else:
             # Response isn't chunked, meaning we probably
             # encountered an error immediately
             yield self._result(response, json=decode)
+
+    def _stream_error_helper(self, message, explanation):
+        """Helper to handle errors after streaming started."""
+        error_response = requests.Response()
+        error_response.status_code = 500
+        error_response.reason = "Internal Server Error"
+        raise APIError(
+            message,
+            response=error_response,
+            explanation=explanation,
+        )

podman/tests/unit/test_imagesmanager.py

@@ -670,6 +670,62 @@ def test_pull_policy(self, mock):
         image = self.client.images.pull("quay.io/fedora:latest", policy="missing")
         self.assertEqual(image.id, image_id)
 
+    @requests_mock.Mocker()
+    def test_pull_no_compat_mode(self, mock):
+        image_id = "sha256:326dd9d7add24646a325e8eaa82125294027db2332e49c5828d96312c5d773ab"
+        mock.post(
+            tests.LIBPOD_URL + "/images/pull?reference=quay.io%2ffedora%3Alatest&quiet=True",
+            json={
+                "error": "",
+                "id": image_id,
+                "images": [image_id],
+                "stream": "",
+            },
+        )
+        mock.get(
+            tests.LIBPOD_URL + "/images"
+            "/sha256%3A326dd9d7add24646a325e8eaa82125294027db2332e49c5828d96312c5d773ab/json",
+            json=FIRST_IMAGE,
+        )
+
+        image = self.client.images.pull("quay.io/fedora", "latest", compatMode=False)
+        self.assertEqual(image.id, image_id)
+
+    @requests_mock.Mocker()
+    def test_pull_no_compat_mode_no_image(self, mock):
+        non_existing_reference = "quay.io/f4ee35641334/f6fda4bb"
+        error = {
+            "cause": "access to the requested resource is not authorized",
+            "message": "unable to copy from source",
+            "response": 401,
+        }
+
+        stream_cases = [
+            dict(stream=False, quiet_postfix="&quiet=True"),
+            dict(stream=True, quiet_postfix=""),
+        ]
+
+        for stream_case in stream_cases:
+            with self.subTest(stream_case=stream_case):
+                mock.post(
+                    tests.LIBPOD_URL
+                    + f"/images/pull?reference={non_existing_reference}%3Alatest"
+                    + f"{stream_case['quiet_postfix']}",
+                    status_code=error["response"],
+                    json=error,
+                )
+
+                with self.assertRaises(APIError) as context:
+                    self.client.images.pull(
+                        non_existing_reference,
+                        "latest",
+                        compatMode=False,
+                        stream=stream_case["stream"],
+                    )
+                self.assertEqual(context.exception.args[0], error["cause"])
+                self.assertEqual(context.exception.explanation, error["message"])
+                self.assertEqual(context.exception.response.status_code, error["response"])
+
     @requests_mock.Mocker()
     def test_list_with_name_parameter(self, mock):
         """Test that name parameter is correctly converted to a reference filter"""