Extract shared mapping-list parser helper

Co-authored-by: Shri Sukhani <shrisukhani@users.noreply.github.com>

Author: cursoragent

hyperbrowser/client/managers/extension_utils.py

@@ -3,6 +3,7 @@
 
 from hyperbrowser.exceptions import HyperbrowserError
 from hyperbrowser.models.extension import ExtensionResponse
+from .list_parsing_utils import parse_mapping_list_items
 
 _MAX_DISPLAYED_MISSING_KEYS = 20
 _MAX_DISPLAYED_MISSING_KEY_LENGTH = 120
@@ -108,48 +109,9 @@ def parse_extension_list_response_data(response_data: Any) -> List[ExtensionResp
             "Failed to iterate 'extensions' list from response",
             original_error=exc,
         ) from exc
-    parsed_extensions: List[ExtensionResponse] = []
-    for index, extension in enumerate(extension_items):
-        if not isinstance(extension, Mapping):
-            raise HyperbrowserError(
-                "Expected extension object at index "
-                f"{index} but got {_get_type_name(extension)}"
-            )
-        try:
-            extension_keys = list(extension.keys())
-        except HyperbrowserError:
-            raise
-        except Exception as exc:
-            raise HyperbrowserError(
-                f"Failed to read extension object at index {index}",
-                original_error=exc,
-            ) from exc
-        for key in extension_keys:
-            if type(key) is str:
-                continue
-            raise HyperbrowserError(
-                f"Expected extension object keys to be strings at index {index}"
-            )
-        extension_payload: dict[str, object] = {}
-        for key in extension_keys:
-            try:
-                extension_payload[key] = extension[key]
-            except HyperbrowserError:
-                raise
-            except Exception as exc:
-                key_display = _format_key_display(key)
-                raise HyperbrowserError(
-                    "Failed to read extension object value for key "
-                    f"'{key_display}' at index {index}",
-                    original_error=exc,
-                ) from exc
-        try:
-            parsed_extensions.append(ExtensionResponse(**extension_payload))
-        except HyperbrowserError:
-            raise
-        except Exception as exc:
-            raise HyperbrowserError(
-                f"Failed to parse extension at index {index}",
-                original_error=exc,
-            ) from exc
-    return parsed_extensions
+    return parse_mapping_list_items(
+        extension_items,
+        item_label="extension",
+        parse_item=lambda extension_payload: ExtensionResponse(**extension_payload),
+        key_display=_format_key_display,
+    )

hyperbrowser/client/managers/list_parsing_utils.py

@@ -0,0 +1,57 @@
+from collections.abc import Mapping
+from typing import Any, Callable, List, TypeVar
+
+from hyperbrowser.exceptions import HyperbrowserError
+
+T = TypeVar("T")
+
+
+def parse_mapping_list_items(
+    items: List[Any],
+    *,
+    item_label: str,
+    parse_item: Callable[[dict[str, object]], T],
+    key_display: Callable[[str], str],
+) -> List[T]:
+    parsed_items: List[T] = []
+    for index, item in enumerate(items):
+        if not isinstance(item, Mapping):
+            raise HyperbrowserError(
+                f"Expected {item_label} object at index {index} but got {type(item).__name__}"
+            )
+        try:
+            item_keys = list(item.keys())
+        except HyperbrowserError:
+            raise
+        except Exception as exc:
+            raise HyperbrowserError(
+                f"Failed to read {item_label} object at index {index}",
+                original_error=exc,
+            ) from exc
+        for key in item_keys:
+            if type(key) is str:
+                continue
+            raise HyperbrowserError(
+                f"Expected {item_label} object keys to be strings at index {index}"
+            )
+        item_payload: dict[str, object] = {}
+        for key in item_keys:
+            try:
+                item_payload[key] = item[key]
+            except HyperbrowserError:
+                raise
+            except Exception as exc:
+                raise HyperbrowserError(
+                    f"Failed to read {item_label} object value for key '{key_display(key)}' at index {index}",
+                    original_error=exc,
+                ) from exc
+        try:
+            parsed_items.append(parse_item(item_payload))
+        except HyperbrowserError:
+            raise
+        except Exception as exc:
+            raise HyperbrowserError(
+                f"Failed to parse {item_label} at index {index}",
+                original_error=exc,
+            ) from exc
+    return parsed_items

hyperbrowser/client/managers/session_utils.py

@@ -1,8 +1,8 @@
-from collections.abc import Mapping
 from typing import Any, List, Type, TypeVar
 
 from hyperbrowser.exceptions import HyperbrowserError
 from hyperbrowser.models.session import SessionRecording
+from .list_parsing_utils import parse_mapping_list_items
 from .response_utils import parse_response_model
 
 T = TypeVar("T")
@@ -59,48 +59,9 @@ def parse_session_recordings_response_data(
             "Failed to iterate session recording response list",
             original_error=exc,
         ) from exc
-    parsed_recordings: List[SessionRecording] = []
-    for index, recording in enumerate(recording_items):
-        if not isinstance(recording, Mapping):
-            raise HyperbrowserError(
-                "Expected session recording object at index "
-                f"{index} but got {type(recording).__name__}"
-            )
-        try:
-            recording_keys = list(recording.keys())
-        except HyperbrowserError:
-            raise
-        except Exception as exc:
-            raise HyperbrowserError(
-                f"Failed to read session recording object at index {index}",
-                original_error=exc,
-            ) from exc
-        for key in recording_keys:
-            if type(key) is str:
-                continue
-            raise HyperbrowserError(
-                f"Expected session recording object keys to be strings at index {index}"
-            )
-        recording_payload: dict[str, object] = {}
-        for key in recording_keys:
-            try:
-                recording_payload[key] = recording[key]
-            except HyperbrowserError:
-                raise
-            except Exception as exc:
-                key_display = _format_recording_key_display(key)
-                raise HyperbrowserError(
-                    "Failed to read session recording object value for key "
-                    f"'{key_display}' at index {index}",
-                    original_error=exc,
-                ) from exc
-        try:
-            parsed_recordings.append(SessionRecording(**recording_payload))
-        except HyperbrowserError:
-            raise
-        except Exception as exc:
-            raise HyperbrowserError(
-                f"Failed to parse session recording at index {index}",
-                original_error=exc,
-            ) from exc
-    return parsed_recordings
+    return parse_mapping_list_items(
+        recording_items,
+        item_label="session recording",
+        parse_item=lambda recording_payload: SessionRecording(**recording_payload),
+        key_display=_format_recording_key_display,
+    )

tests/test_list_parsing_utils.py

@@ -0,0 +1,117 @@
+from collections.abc import Mapping
+from typing import Iterator
+
+import pytest
+
+from hyperbrowser.client.managers.list_parsing_utils import parse_mapping_list_items
+from hyperbrowser.exceptions import HyperbrowserError
+
+
+class _ExplodingKeysMapping(Mapping[object, object]):
+    def __getitem__(self, key: object) -> object:
+        _ = key
+        return "value"
+
+    def __iter__(self) -> Iterator[object]:
+        return iter(())
+
+    def __len__(self) -> int:
+        return 0
+
+    def keys(self):  # type: ignore[override]
+        raise RuntimeError("broken keys")
+
+
+class _ExplodingValueMapping(Mapping[object, object]):
+    def __getitem__(self, key: object) -> object:
+        _ = key
+        raise RuntimeError("broken getitem")
+
+    def __iter__(self) -> Iterator[object]:
+        return iter(("field",))
+
+    def __len__(self) -> int:
+        return 1
+
+    def keys(self):  # type: ignore[override]
+        return ("field",)
+
+
+def test_parse_mapping_list_items_parses_each_mapping():
+    parsed = parse_mapping_list_items(
+        [{"id": "a"}, {"id": "b"}],
+        item_label="demo",
+        parse_item=lambda payload: payload["id"],
+        key_display=lambda key: key,
+    )
+
+    assert parsed == ["a", "b"]
+
+
+def test_parse_mapping_list_items_rejects_non_mapping_items():
+    with pytest.raises(
+        HyperbrowserError, match="Expected demo object at index 0 but got list"
+    ):
+        parse_mapping_list_items(
+            [[]],
+            item_label="demo",
+            parse_item=lambda payload: payload,
+            key_display=lambda key: key,
+        )
+
+
+def test_parse_mapping_list_items_wraps_key_read_failures():
+    with pytest.raises(
+        HyperbrowserError, match="Failed to read demo object at index 0"
+    ) as exc_info:
+        parse_mapping_list_items(
+            [_ExplodingKeysMapping()],
+            item_label="demo",
+            parse_item=lambda payload: payload,
+            key_display=lambda key: key,
+        )
+
+    assert isinstance(exc_info.value.original_error, RuntimeError)
+
+
+def test_parse_mapping_list_items_rejects_non_string_keys():
+    with pytest.raises(
+        HyperbrowserError,
+        match="Expected demo object keys to be strings at index 0",
+    ):
+        parse_mapping_list_items(
+            [{1: "value"}],
+            item_label="demo",
+            parse_item=lambda payload: payload,
+            key_display=lambda key: key,
+        )
+
+
+def test_parse_mapping_list_items_wraps_value_read_failures():
+    with pytest.raises(
+        HyperbrowserError,
+        match="Failed to read demo object value for key 'field' at index 0",
+    ) as exc_info:
+        parse_mapping_list_items(
+            [_ExplodingValueMapping()],
+            item_label="demo",
+            parse_item=lambda payload: payload,
+            key_display=lambda key: key,
+        )
+
+    assert isinstance(exc_info.value.original_error, RuntimeError)
+
+
+def test_parse_mapping_list_items_wraps_parse_failures():
+    with pytest.raises(
+        HyperbrowserError,
+        match="Failed to parse demo at index 0",
+    ) as exc_info:
+        parse_mapping_list_items(
+            [{"id": "x"}],
+            item_label="demo",
+            parse_item=lambda payload: 1 / 0,
+            key_display=lambda key: key,
+        )
+
+    assert isinstance(exc_info.value.original_error, ZeroDivisionError)