Centralize file-path existence checks for upload managers

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

Author: cursoragent

hyperbrowser/client/file_utils.py

@@ -0,0 +1,15 @@
+import os
+
+from hyperbrowser.exceptions import HyperbrowserError
+
+
+def ensure_existing_file_path(
+    file_path: str,
+    *,
+    missing_file_message: str,
+    not_file_message: str,
+) -> None:
+    if not os.path.exists(file_path):
+        raise HyperbrowserError(missing_file_message)
+    if not os.path.isfile(file_path):
+        raise HyperbrowserError(not_file_message)

hyperbrowser/client/managers/async_manager/extension.py

@@ -1,7 +1,7 @@
-import os
 from typing import List
 
 from hyperbrowser.exceptions import HyperbrowserError
+from ...file_utils import ensure_existing_file_path
 from hyperbrowser.models.extension import CreateExtensionParams, ExtensionResponse
 
 
@@ -14,13 +14,11 @@ async def create(self, params: CreateExtensionParams) -> ExtensionResponse:
         payload = params.model_dump(exclude_none=True, by_alias=True)
         payload.pop("filePath", None)
 
-        # Check if file exists before trying to open it
-        if not os.path.exists(file_path):
-            raise HyperbrowserError(f"Extension file not found at path: {file_path}")
-        if not os.path.isfile(file_path):
-            raise HyperbrowserError(
-                f"Extension file path must point to a file: {file_path}"
-            )
+        ensure_existing_file_path(
+            file_path,
+            missing_file_message=f"Extension file not found at path: {file_path}",
+            not_file_message=f"Extension file path must point to a file: {file_path}",
+        )
 
         try:
             with open(file_path, "rb") as extension_file:

hyperbrowser/client/managers/async_manager/session.py

@@ -3,6 +3,7 @@
 from typing import IO, List, Optional, Union, overload
 import warnings
 from hyperbrowser.exceptions import HyperbrowserError
+from ...file_utils import ensure_existing_file_path
 from ....models.session import (
     BasicResponse,
     CreateSessionParams,
@@ -115,12 +116,11 @@ async def upload_file(
     ) -> UploadFileResponse:
         if isinstance(file_input, (str, PathLike)):
             file_path = os.fspath(file_input)
-            if not os.path.exists(file_path):
-                raise HyperbrowserError(f"Upload file not found at path: {file_path}")
-            if not os.path.isfile(file_path):
-                raise HyperbrowserError(
-                    f"Upload file path must point to a file: {file_path}"
-                )
+            ensure_existing_file_path(
+                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}",
+            )
             try:
                 with open(file_path, "rb") as file_obj:
                     files = {"file": file_obj}

hyperbrowser/client/managers/sync_manager/extension.py

@@ -1,7 +1,7 @@
-import os
 from typing import List
 
 from hyperbrowser.exceptions import HyperbrowserError
+from ...file_utils import ensure_existing_file_path
 from hyperbrowser.models.extension import CreateExtensionParams, ExtensionResponse
 
 
@@ -14,13 +14,11 @@ def create(self, params: CreateExtensionParams) -> ExtensionResponse:
         payload = params.model_dump(exclude_none=True, by_alias=True)
         payload.pop("filePath", None)
 
-        # Check if file exists before trying to open it
-        if not os.path.exists(file_path):
-            raise HyperbrowserError(f"Extension file not found at path: {file_path}")
-        if not os.path.isfile(file_path):
-            raise HyperbrowserError(
-                f"Extension file path must point to a file: {file_path}"
-            )
+        ensure_existing_file_path(
+            file_path,
+            missing_file_message=f"Extension file not found at path: {file_path}",
+            not_file_message=f"Extension file path must point to a file: {file_path}",
+        )
 
         try:
             with open(file_path, "rb") as extension_file:

hyperbrowser/client/managers/sync_manager/session.py

@@ -3,6 +3,7 @@
 from typing import IO, List, Optional, Union, overload
 import warnings
 from hyperbrowser.exceptions import HyperbrowserError
+from ...file_utils import ensure_existing_file_path
 from ....models.session import (
     BasicResponse,
     CreateSessionParams,
@@ -107,12 +108,11 @@ def upload_file(
     ) -> UploadFileResponse:
         if isinstance(file_input, (str, PathLike)):
             file_path = os.fspath(file_input)
-            if not os.path.exists(file_path):
-                raise HyperbrowserError(f"Upload file not found at path: {file_path}")
-            if not os.path.isfile(file_path):
-                raise HyperbrowserError(
-                    f"Upload file path must point to a file: {file_path}"
-                )
+            ensure_existing_file_path(
+                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}",
+            )
             try:
                 with open(file_path, "rb") as file_obj:
                     files = {"file": file_obj}

tests/test_file_utils.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+
+import pytest
+
+from hyperbrowser.client.file_utils import ensure_existing_file_path
+from hyperbrowser.exceptions import HyperbrowserError
+
+
+def test_ensure_existing_file_path_accepts_existing_file(tmp_path: Path):
+    file_path = tmp_path / "file.txt"
+    file_path.write_text("content")
+
+    ensure_existing_file_path(
+        str(file_path),
+        missing_file_message="missing",
+        not_file_message="not-file",
+    )
+
+
+def test_ensure_existing_file_path_raises_for_missing_file(tmp_path: Path):
+    missing_path = tmp_path / "missing.txt"
+
+    with pytest.raises(HyperbrowserError, match="missing"):
+        ensure_existing_file_path(
+            str(missing_path),
+            missing_file_message="missing",
+            not_file_message="not-file",
+        )
+
+
+def test_ensure_existing_file_path_raises_for_directory(tmp_path: Path):
+    with pytest.raises(HyperbrowserError, match="not-file"):
+        ensure_existing_file_path(
+            str(tmp_path),
+            missing_file_message="missing",
+            not_file_message="not-file",
+        )