Centralize session upload input normalization

cursoragent · shrisukhani · cursoragent · commit 812d28d76938 · 2026-02-14T09:21:14.000Z
Co-authored-by: Shri Sukhani &lt;shrisukhani@users.noreply.github.com&gt;
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
@@ -97,7 +97,8 @@ This runs lint, format checks, compile checks, tests, and package build.
   - `tests/test_docs_python3_commands.py` (`README`/`CONTRIBUTING`/examples python3 command consistency enforcement),
   - `tests/test_example_sync_async_parity.py` (sync/async example parity enforcement),
   - `tests/test_example_run_instructions.py` (example run-instruction consistency enforcement),
-  - `tests/test_computer_action_endpoint_helper_usage.py` (computer-action endpoint-normalization helper usage enforcement).
+  - `tests/test_computer_action_endpoint_helper_usage.py` (computer-action endpoint-normalization helper usage enforcement),
+  - `tests/test_session_upload_helper_usage.py` (session upload-input normalization helper usage enforcement).
 
 ## Code quality conventions
 
diff --git a/hyperbrowser/client/managers/async_manager/session.py b/hyperbrowser/client/managers/async_manager/session.py
@@ -1,11 +1,9 @@
-import os
 from os import PathLike
 from typing import IO, List, Optional, Union, overload
 import warnings
 from hyperbrowser.exceptions import HyperbrowserError
-from hyperbrowser.type_utils import is_plain_string, is_string_subclass_instance
-from ...file_utils import ensure_existing_file_path
 from ..serialization_utils import serialize_model_dump_to_dict
+from ..session_upload_utils import normalize_upload_file_input
 from ..session_utils import (
     parse_session_recordings_response_data,
     parse_session_response_model,
@@ -166,21 +164,12 @@ async def get_downloads_url(self, id: str) -> GetSessionDownloadsUrlResponse:
     async def upload_file(
         self, id: str, file_input: Union[str, PathLike[str], IO]
     ) -> UploadFileResponse:
-        if is_plain_string(file_input) or isinstance(file_input, PathLike):
-            try:
-                raw_file_path = os.fspath(file_input)
-            except HyperbrowserError:
-                raise
-            except Exception as exc:
-                raise HyperbrowserError(
-                    "file_input path is invalid",
-                    original_error=exc,
-                ) from exc
-            file_path = ensure_existing_file_path(
-                raw_file_path,
-                missing_file_message=f"Upload file not found at path: {raw_file_path}",
-                not_file_message=f"Upload file path must point to a file: {raw_file_path}",
-            )
+        file_path, file_obj = normalize_upload_file_input(
+            file_input,
+            missing_file_message=f"Upload file not found at path: {file_input}",
+            not_file_message=f"Upload file path must point to a file: {file_input}",
+        )
+        if file_path is not None:
             try:
                 with open(file_path, "rb") as file_obj:
                     files = {"file": file_obj}
@@ -193,39 +182,12 @@ async def upload_file(
                     f"Failed to open upload file at path: {file_path}",
                     original_error=exc,
                 ) from exc
-        elif is_string_subclass_instance(file_input):
-            raise HyperbrowserError("file_input path must be a plain string path")
         else:
-            try:
-                read_method = getattr(file_input, "read", None)
-            except HyperbrowserError:
-                raise
-            except Exception as exc:
-                raise HyperbrowserError(
-                    "file_input file-like object state is invalid",
-                    original_error=exc,
-                ) from exc
-            if callable(read_method):
-                try:
-                    is_closed = bool(getattr(file_input, "closed", False))
-                except HyperbrowserError:
-                    raise
-                except Exception as exc:
-                    raise HyperbrowserError(
-                        "file_input file-like object state is invalid",
-                        original_error=exc,
-                    ) from exc
-                if is_closed:
-                    raise HyperbrowserError("file_input file-like object must be open")
-                files = {"file": file_input}
-                response = await self._client.transport.post(
-                    self._client._build_url(f"/session/{id}/uploads"),
-                    files=files,
-                )
-            else:
-                raise HyperbrowserError(
-                    "file_input must be a file path or file-like object"
-                )
+            files = {"file": file_obj}
+            response = await self._client.transport.post(
+                self._client._build_url(f"/session/{id}/uploads"),
+                files=files,
+            )
 
         return parse_session_response_model(
             response.data,
diff --git a/hyperbrowser/client/managers/session_upload_utils.py b/hyperbrowser/client/managers/session_upload_utils.py
@@ -0,0 +1,61 @@
+import os
+from os import PathLike
+from typing import IO, Optional, Tuple, Union
+
+from hyperbrowser.exceptions import HyperbrowserError
+from hyperbrowser.type_utils import is_plain_string, is_string_subclass_instance
+
+from ..file_utils import ensure_existing_file_path
+
+
+def normalize_upload_file_input(
+    file_input: Union[str, PathLike[str], IO],
+    *,
+    missing_file_message: str,
+    not_file_message: str,
+) -> Tuple[Optional[str], Optional[IO]]:
+    if is_plain_string(file_input) or isinstance(file_input, PathLike):
+        try:
+            raw_file_path = os.fspath(file_input)
+        except HyperbrowserError:
+            raise
+        except Exception as exc:
+            raise HyperbrowserError(
+                "file_input path is invalid",
+                original_error=exc,
+            ) from exc
+        file_path = ensure_existing_file_path(
+            raw_file_path,
+            missing_file_message=missing_file_message,
+            not_file_message=not_file_message,
+        )
+        return file_path, None
+
+    if is_string_subclass_instance(file_input):
+        raise HyperbrowserError("file_input path must be a plain string path")
+
+    try:
+        read_method = getattr(file_input, "read", None)
+    except HyperbrowserError:
+        raise
+    except Exception as exc:
+        raise HyperbrowserError(
+            "file_input file-like object state is invalid",
+            original_error=exc,
+        ) from exc
+    if not callable(read_method):
+        raise HyperbrowserError("file_input must be a file path or file-like object")
+
+    try:
+        is_closed = bool(getattr(file_input, "closed", False))
+    except HyperbrowserError:
+        raise
+    except Exception as exc:
+        raise HyperbrowserError(
+            "file_input file-like object state is invalid",
+            original_error=exc,
+        ) from exc
+    if is_closed:
+        raise HyperbrowserError("file_input file-like object must be open")
+
+    return None, file_input
diff --git a/hyperbrowser/client/managers/sync_manager/session.py b/hyperbrowser/client/managers/sync_manager/session.py
@@ -1,11 +1,9 @@
-import os
 from os import PathLike
 from typing import IO, List, Optional, Union, overload
 import warnings
 from hyperbrowser.exceptions import HyperbrowserError
-from hyperbrowser.type_utils import is_plain_string, is_string_subclass_instance
-from ...file_utils import ensure_existing_file_path
 from ..serialization_utils import serialize_model_dump_to_dict
+from ..session_upload_utils import normalize_upload_file_input
 from ..session_utils import (
     parse_session_recordings_response_data,
     parse_session_response_model,
@@ -158,21 +156,12 @@ def get_downloads_url(self, id: str) -> GetSessionDownloadsUrlResponse:
     def upload_file(
         self, id: str, file_input: Union[str, PathLike[str], IO]
     ) -> UploadFileResponse:
-        if is_plain_string(file_input) or isinstance(file_input, PathLike):
-            try:
-                raw_file_path = os.fspath(file_input)
-            except HyperbrowserError:
-                raise
-            except Exception as exc:
-                raise HyperbrowserError(
-                    "file_input path is invalid",
-                    original_error=exc,
-                ) from exc
-            file_path = ensure_existing_file_path(
-                raw_file_path,
-                missing_file_message=f"Upload file not found at path: {raw_file_path}",
-                not_file_message=f"Upload file path must point to a file: {raw_file_path}",
-            )
+        file_path, file_obj = normalize_upload_file_input(
+            file_input,
+            missing_file_message=f"Upload file not found at path: {file_input}",
+            not_file_message=f"Upload file path must point to a file: {file_input}",
+        )
+        if file_path is not None:
             try:
                 with open(file_path, "rb") as file_obj:
                     files = {"file": file_obj}
@@ -185,39 +174,12 @@ def upload_file(
                     f"Failed to open upload file at path: {file_path}",
                     original_error=exc,
                 ) from exc
-        elif is_string_subclass_instance(file_input):
-            raise HyperbrowserError("file_input path must be a plain string path")
         else:
-            try:
-                read_method = getattr(file_input, "read", None)
-            except HyperbrowserError:
-                raise
-            except Exception as exc:
-                raise HyperbrowserError(
-                    "file_input file-like object state is invalid",
-                    original_error=exc,
-                ) from exc
-            if callable(read_method):
-                try:
-                    is_closed = bool(getattr(file_input, "closed", False))
-                except HyperbrowserError:
-                    raise
-                except Exception as exc:
-                    raise HyperbrowserError(
-                        "file_input file-like object state is invalid",
-                        original_error=exc,
-                    ) from exc
-                if is_closed:
-                    raise HyperbrowserError("file_input file-like object must be open")
-                files = {"file": file_input}
-                response = self._client.transport.post(
-                    self._client._build_url(f"/session/{id}/uploads"),
-                    files=files,
-                )
-            else:
-                raise HyperbrowserError(
-                    "file_input must be a file path or file-like object"
-                )
+            files = {"file": file_obj}
+            response = self._client.transport.post(
+                self._client._build_url(f"/session/{id}/uploads"),
+                files=files,
+            )
 
         return parse_session_response_model(
             response.data,
diff --git a/tests/test_architecture_marker_usage.py b/tests/test_architecture_marker_usage.py
@@ -27,6 +27,7 @@
     "tests/test_example_sync_async_parity.py",
     "tests/test_example_run_instructions.py",
     "tests/test_computer_action_endpoint_helper_usage.py",
+    "tests/test_session_upload_helper_usage.py",
 )
 
 
diff --git a/tests/test_session_upload_helper_usage.py b/tests/test_session_upload_helper_usage.py
@@ -0,0 +1,18 @@
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.architecture
+
+
+SESSION_MANAGER_MODULES = (
+    "hyperbrowser/client/managers/sync_manager/session.py",
+    "hyperbrowser/client/managers/async_manager/session.py",
+)
+
+
+def test_session_managers_use_shared_upload_input_normalizer():
+    for module_path in SESSION_MANAGER_MODULES:
+        module_text = Path(module_path).read_text(encoding="utf-8")
+        assert "normalize_upload_file_input(" in module_text
+        assert "os.fspath(" not in module_text
diff --git a/tests/test_session_upload_utils.py b/tests/test_session_upload_utils.py
@@ -0,0 +1,107 @@
+import io
+from os import PathLike
+from pathlib import Path
+
+import pytest
+
+from hyperbrowser.client.managers.session_upload_utils import (
+    normalize_upload_file_input,
+)
+from hyperbrowser.exceptions import HyperbrowserError
+
+
+def test_normalize_upload_file_input_returns_path_for_plain_string(tmp_path: Path):
+    file_path = tmp_path / "file.txt"
+    file_path.write_text("content")
+
+    normalized_path, file_obj = normalize_upload_file_input(
+        str(file_path),
+        missing_file_message=f"Upload file not found at path: {file_path}",
+        not_file_message=f"Upload file path must point to a file: {file_path}",
+    )
+
+    assert normalized_path == str(file_path)
+    assert file_obj is None
+
+
+def test_normalize_upload_file_input_returns_path_for_pathlike(tmp_path: Path):
+    file_path = tmp_path / "file.txt"
+    file_path.write_text("content")
+
+    normalized_path, file_obj = normalize_upload_file_input(
+        file_path,
+        missing_file_message=f"Upload file not found at path: {file_path}",
+        not_file_message=f"Upload file path must point to a file: {file_path}",
+    )
+
+    assert normalized_path == str(file_path)
+    assert file_obj is None
+
+
+def test_normalize_upload_file_input_rejects_string_subclass():
+    class _PathString(str):
+        pass
+
+    with pytest.raises(
+        HyperbrowserError, match="file_input path must be a plain string path"
+    ) as exc_info:
+        normalize_upload_file_input(
+            _PathString("/tmp/file.txt"),  # type: ignore[arg-type]
+            missing_file_message="Upload file not found at path: /tmp/file.txt",
+            not_file_message="Upload file path must point to a file: /tmp/file.txt",
+        )
+
+    assert exc_info.value.original_error is None
+
+
+def test_normalize_upload_file_input_wraps_invalid_pathlike_state_errors():
+    class _BrokenPathLike(PathLike[str]):
+        def __fspath__(self) -> str:
+            raise RuntimeError("broken fspath")
+
+    with pytest.raises(
+        HyperbrowserError, match="file_input path is invalid"
+    ) as exc_info:
+        normalize_upload_file_input(
+            _BrokenPathLike(),
+            missing_file_message="Upload file not found",
+            not_file_message="Upload file path must point to a file",
+        )
+
+    assert isinstance(exc_info.value.original_error, RuntimeError)
+
+
+def test_normalize_upload_file_input_returns_open_file_like_object():
+    file_obj = io.BytesIO(b"content")
+
+    normalized_path, normalized_file_obj = normalize_upload_file_input(
+        file_obj,
+        missing_file_message="Upload file not found",
+        not_file_message="Upload file path must point to a file",
+    )
+
+    assert normalized_path is None
+    assert normalized_file_obj is file_obj
+
+
+def test_normalize_upload_file_input_rejects_closed_file_like_object():
+    file_obj = io.BytesIO(b"content")
+    file_obj.close()
+
+    with pytest.raises(HyperbrowserError, match="file-like object must be open"):
+        normalize_upload_file_input(
+            file_obj,
+            missing_file_message="Upload file not found",
+            not_file_message="Upload file path must point to a file",
+        )
+
+
+def test_normalize_upload_file_input_rejects_non_callable_read_attribute():
+    fake_file = type("FakeFile", (), {"read": "not-callable"})()
+
+    with pytest.raises(HyperbrowserError, match="file_input must be a file path"):
+        normalize_upload_file_input(
+            fake_file,  # type: ignore[arg-type]
+            missing_file_message="Upload file not found",
+            not_file_message="Upload file path must point to a file",
+        )

Original file line number	Diff line number	Diff line change
`@@ -27,6 +27,7 @@`
`27`	`27`	`"tests/test_example_sync_async_parity.py",`
`28`	`28`	`"tests/test_example_run_instructions.py",`
`29`	`29`	`"tests/test_computer_action_endpoint_helper_usage.py",`
	`30`	`+ "tests/test_session_upload_helper_usage.py",`
`30`	`31`	`)`
`31`	`32`
`32`	`33`