Refactor/replace mocks with cassettes

src/pyicloud_ipd/session.py

@@ -1,7 +1,8 @@
 import inspect
 import json
 import logging
-from typing import Any, Callable, Mapping, NoReturn
+import typing
+from typing import Any, Callable, Dict, Mapping, NoReturn, Sequence
 
 from requests import Response, Session
 from typing_extensions import override
@@ -73,11 +74,13 @@ def request(self, method: str, url, **kwargs):  # type: ignore
 
         if "timeout" not in kwargs and self.service.http_timeout is not None:
             kwargs["timeout"] = self.service.http_timeout
-
         response = throw_on_503(
             self.observe(handle_connection_error(super().request)(method, url, **kwargs))
         )
 
+        content_type = response.headers.get("Content-Type", "").split(";")[0]
+        json_mimetypes = ["application/json", "text/json"]
+
         request_logger.debug(response.headers)
 
         for header, value in HEADER_DATA.items():
@@ -94,62 +97,61 @@ def request(self, method: str, url, **kwargs):  # type: ignore
         self.cookies.save(ignore_discard=True, ignore_expires=True)  # type: ignore[attr-defined]
         LOGGER.debug("Cookies saved to %s", self.service.cookiejar_path)
 
-        # content_type = response.headers.get("Content-Type", "").split(";")[0]
-        # json_mimetypes = ["application/json", "text/json"]
-
-        # if (
-        #     not response.ok
-        #     and content_type not in json_mimetypes
-        #     and not is_streaming_response(response)
-        # ):
-        #     self._raise_error(str(response.status_code), response.reason)
-
-        # if content_type not in json_mimetypes:
-        #     if self.service.session_data.get("apple_rscd") == "401":
-        #         code: str | None = "401"
-        #         reason: str | None = "Invalid username/password combination."
-        #         self._raise_error(code or "Unknown", reason or "Unknown")
-
-        #     return response
-
-        # try:
-        #     data = response.json() if response.status_code != 204 else {}
-        # except ValueError:
-        #     request_logger.warning("Failed to parse response with JSON mimetype")
-        #     return response
-
-        # request_logger.debug(data)
-
-        # if isinstance(data, dict):
-        #     if data.get("hasError"):
-        #         errors: Sequence[Dict[str, Any]] | None = typing.cast(
-        #             Sequence[Dict[str, Any]] | None, data.get("service_errors")
-        #         )
-        #         # service_errors returns a list of dict
-        #         #    dict includes the keys: code, title, message, supressDismissal
-        #         # Assuming a single error for now
-        #         # May need to revisit to capture and handle multiple errors
-        #         if errors:
-        #             code = errors[0].get("code")
-        #             reason = errors[0].get("message")
-        #         self._raise_error(code or "Unknown", reason or "Unknown")
-        #     elif not data.get("success"):
-        #         reason = data.get("errorMessage")
-        #         reason = reason or data.get("reason")
-        #         reason = reason or data.get("errorReason")
-        #         if not reason and isinstance(data.get("error"), str):
-        #             reason = data.get("error")
-        #         if not reason and data.get("error"):
-        #             reason = "Unknown reason"
-
-        #         code = data.get("errorCode")
-        #         if not code and data.get("serverErrorCode"):
-        #             code = data.get("serverErrorCode")
-        #         if not code and data.get("error"):
-        #             code = data.get("error")
-
-        #         if reason:
-        #             self._raise_error(code or "Unknown", reason)
+        # For error responses, raise exceptions for authentication/server errors
+        # but not for 404s which should be handled gracefully by the caller
+        if (
+            not response.ok
+            and response.status_code != 404
+            and (content_type not in json_mimetypes or response.status_code in [421, 450, 500])
+        ):
+            self._raise_error(str(response.status_code), response.reason)
+
+        if content_type not in json_mimetypes:
+            if self.service.session_data.get("apple_rscd") == "401":
+                code: str | None = "401"
+                reason: str | None = "Invalid username/password combination."
+                self._raise_error(code or "Unknown", reason or "Unknown")
+
+            return response
+
+        try:
+            data = response.json() if response.status_code != 204 else {}
+        except ValueError:
+            request_logger.warning("Failed to parse response with JSON mimetype")
+            return response
+
+        request_logger.debug(data)
+
+        if isinstance(data, dict):
+            if data.get("hasError"):
+                errors: Sequence[Dict[str, Any]] | None = typing.cast(
+                    Sequence[Dict[str, Any]] | None, data.get("service_errors")
+                )
+                # service_errors returns a list of dict
+                #    dict includes the keys: code, title, message, supressDismissal
+                # Assuming a single error for now
+                # May need to revisit to capture and handle multiple errors
+                if errors:
+                    code = errors[0].get("code")
+                    reason = errors[0].get("message")
+                self._raise_error(code or "Unknown", reason or "Unknown")
+            elif not data.get("success"):
+                reason = data.get("errorMessage")
+                reason = reason or data.get("reason")
+                reason = reason or data.get("errorReason")
+                if not reason and isinstance(data.get("error"), str):
+                    reason = data.get("error")
+                if not reason and data.get("error"):
+                    reason = "Unknown reason"
+
+                code = data.get("errorCode")
+                if not code and data.get("serverErrorCode"):
+                    code = data.get("serverErrorCode")
+                if not code and data.get("error"):
+                    code = data.get("error")
+
+                if reason:
+                    self._raise_error(code or "Unknown", reason)
 
         return response
 

tests/helpers/__init__.py

@@ -278,16 +278,10 @@ def readline(self, size: int = -1) -> str:  # type: ignore[override]
     # Create both original and cleaned output
     cleaned_output = "\n".join(cleaned_lines)
 
-    # For compatibility with old tests, adjust exit codes for specific error conditions
-    adjusted_exit_code = exit_code
-    if exit_code == 1 and "Invalid email/password combination" in raw_output:
-        # Authentication failure - old tests expect exit code 2
-        adjusted_exit_code = 2
-
     # For compatibility, provide the cleaned output as the primary output
     # but keep raw_output available if needed
     result = TestResult(
-        exit_code=adjusted_exit_code,
+        exit_code=exit_code,
         output=cleaned_output,
         exception=exception,
         stderr=stderr_capture.getvalue(),
@@ -303,15 +297,21 @@ def readline(self, size: int = -1) -> str:  # type: ignore[override]
 
 
 def run_with_cassette(cassette_path: str, f: Callable[[_T_contra], _T_co], inp: _T_contra) -> _T_co:
-    with vcr.use_cassette(cassette_path):
-        return f(inp)
+    with vcr.use_cassette(cassette_path, allow_playback_repeats=False) as _cassette:
+        result = f(inp)
+        # Check that all interactions were played (optional - can be enabled per test)
+        # assert cassette.all_played, f"Not all cassette interactions were used in {cassette_path}"
+        return result
 
 
 def run_cassette(
     cassette_path: str, params: Sequence[str], input: str | bytes | IO[Any] | None = None
 ) -> TestResult:
-    with vcr.use_cassette(cassette_path):
-        return print_result_exception(run_main_env(DEFAULT_ENV, params, input))
+    with vcr.use_cassette(cassette_path, allow_playback_repeats=False) as _cassette:
+        result = print_result_exception(run_main_env(DEFAULT_ENV, params, input))
+        # Check that all interactions were played (optional - can be enabled per test)
+        # assert _cassette.all_played, f"Not all cassette interactions were used in {cassette_path}"
+        return result
 
 
 def add_cloned_master_cookie_dir(