fix(kernels): check HTTP status when downloading outputs

src/kaggle/api/kaggle_api_extended.py

@@ -7394,6 +7394,10 @@ def kernels_output(
             outfile = os.path.join(target_dir, item.file_name)
             outfiles.append(outfile)
             download_response = requests.get(item.url, stream=True)
+            # A failed fetch (e.g. an expired signed URL) still has a body, so
+            # writing it would save the error page as the output file and still
+            # report success. Fail loudly instead, like every other download.
+            download_response.raise_for_status()
             if force or self.download_needed(download_response, outfile, quiet):
                 os.makedirs(os.path.split(outfile)[0], exist_ok=True)
                 with open(outfile, "wb") as out:

tests/unit/test_kernels_logs.py

@@ -3,6 +3,7 @@
 from unittest.mock import patch, MagicMock, call
 import io
 import json
+import os
 import tempfile
 import sys
 import builtins
@@ -98,6 +99,38 @@ def test_kernels_output_page_token_downloads_specific_page(self, mock_client, mo
         self.assertEqual(request.page_token, "page-2")
         self.assertEqual(request.page_size, 50)
 
+    @patch("kaggle.api.kaggle_api_extended.requests.get")
+    @patch.object(KaggleApi, "build_kaggle_client")
+    def test_kernels_output_failed_download_is_not_written(self, mock_client, mock_get):
+        """Test a failed file download raises instead of saving the error body."""
+        import requests as _requests
+
+        response = MagicMock()
+        response.files = [MagicMock(file_name="submission.csv", url="https://example.com/submission.csv")]
+        response.next_page_token = ""
+        response.log = None
+
+        mock_kaggle = MagicMock()
+        mock_kaggle.kernels.kernels_api_client.list_kernel_session_output.return_value = response
+        mock_client.return_value.__enter__ = MagicMock(return_value=mock_kaggle)
+        mock_client.return_value.__exit__ = MagicMock(return_value=False)
+
+        error_body = b"<Error><Code>AccessDenied</Code></Error>"
+        failed = MagicMock(status_code=403, content=error_body, headers={"Content-Length": str(len(error_body))})
+        failed.raise_for_status.side_effect = _requests.exceptions.HTTPError("403 Client Error", response=failed)
+        mock_get.return_value = failed
+
+        captured = io.StringIO()
+        sys.stdout = captured
+        try:
+            with tempfile.TemporaryDirectory() as temp_dir:
+                with self.assertRaises(_requests.exceptions.HTTPError):
+                    self.api.kernels_output("owner/kernel-slug", temp_dir, quiet=False)
+                self.assertFalse(os.path.exists(os.path.join(temp_dir, "submission.csv")))
+        finally:
+            sys.stdout = sys.__stdout__
+        self.assertNotIn("Output file downloaded", captured.getvalue())
+
     @patch.object(KaggleApi, "kernels_output")
     def test_kernels_output_cli_passes_page_size(self, mock_output):
         """Test CLI wrapper passes page_size to kernels_output."""