- add test (requires manual test with pyvenv) for #10

- fix #10

cdp_patches/input/browsers.py

@@ -27,14 +27,14 @@
 try:
     from selenium.webdriver import Chrome as SeleniumChrome
 except ImportError:
-    SeleniumChrome: Type["SeleniumChrome"] = "SeleniumChrome"  # type: ignore[no-redef]
+    SeleniumChrome: Type["SeleniumChrome"] = type("SeleniumChrome", (object, ), {})   # type: ignore[no-redef]
 
 try:
     from selenium_driverless.sync.webdriver import Chrome as DriverlessSyncChrome
     from selenium_driverless.webdriver import Chrome as DriverlessAsyncChrome
 except ImportError:
-    DriverlessAsyncChrome: Type["DriverlessAsyncChrome"] = "DriverlessAsyncChrome"  # type: ignore[no-redef]
-    DriverlessSyncChrome: Type["DriverlessSyncChrome"] = "DriverlessSyncChrome"  # type: ignore[no-redef]
+    DriverlessAsyncChrome: Type["DriverlessAsyncChrome"] = type("DriverlessAsyncChrome", (object, ), {})  # type: ignore[no-redef]
+    DriverlessSyncChrome: Type["DriverlessSyncChrome"] = type("DriverlessSyncChrome", (object, ), {})  # type: ignore[no-redef]
 
 all_browsers = Union[AsyncContext, AsyncBrowser, SyncContext, SyncBrowser, BotrightContext, SeleniumChrome, DriverlessAsyncChrome, DriverlessSyncChrome]
 sync_browsers = Union[SeleniumChrome, SyncContext, SyncBrowser, DriverlessSyncChrome]

tests/conftest.py

@@ -1,16 +1,37 @@
-from typing import AsyncGenerator, Generator, List
+from typing import AsyncGenerator, Generator
+from .utils import find_chrome_executable, random_port
+import subprocess
+import tempfile
+import signal
+import os
 
 import pytest
 import pytest_asyncio
-from playwright.async_api import Page as AsyncPage
-from playwright.async_api import async_playwright
-from playwright.sync_api import Page as SyncPage
-from playwright.sync_api import sync_playwright
-from selenium import webdriver as selenium_webdriver
-from selenium.webdriver.chrome.service import Service as SeleniumChromeService
-from selenium_driverless import webdriver as async_webdriver
-from selenium_driverless.sync import webdriver as sync_webdriver
-from webdriver_manager.chrome import ChromeDriverManager
+from typing import List, Type
+
+# support testing without having playwright installed
+try:
+    from playwright.async_api import Page as AsyncPage
+    from playwright.async_api import async_playwright
+    from playwright.sync_api import Page as SyncPage
+    from playwright.sync_api import sync_playwright
+    from selenium import webdriver as selenium_webdriver
+    from selenium.webdriver.chrome.service import Service as SeleniumChromeService
+    from selenium_driverless import webdriver as async_webdriver
+    from selenium_driverless.sync import webdriver as sync_webdriver
+    from webdriver_manager.chrome import ChromeDriverManager
+except ImportError:
+    AsyncBrowser: Type["AsyncBrowser"] = "AsyncBrowser"  # type: ignore[no-redef]
+    AsyncContext: Type["AsyncContext"] = "AsyncContext"  # type: ignore[no-redef]
+    SyncBrowser: Type["SyncBrowser"] = "SyncBrowser"  # type: ignore[no-redef]
+    SyncContext: Type["SyncContext"] = "SyncContext"  # type: ignore[no-redef]
+    SyncPage: Type["SyncPage"] = "SyncPage"  # type: ignore[no-redef]
+    AsyncPage: Type["AsyncPage"] = "AsyncPage"  # type: ignore[no-redef]
+
+    ChromeWebdriver: Type["ChromeWebdriver"] = "ChromeWebdriver"  # type: ignore[no-redef]
+    sync_webdriver = async_webdriver = selenium_webdriver = type("webdriver", (object,), {
+        "Chrome": ChromeWebdriver
+    })
 
 from cdp_patches.input import AsyncInput, SyncInput
 
@@ -160,3 +181,25 @@ async def async_driver() -> AsyncGenerator[async_webdriver.Chrome, None]:
     async with async_webdriver.Chrome(options) as driver:
         driver.async_input = await AsyncInput(browser=driver)
         yield driver
+
+
+@pytest.fixture
+def chrome_proc() -> Generator[subprocess.Popen, None, None]:
+    with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tempdir:
+        path = find_chrome_executable()
+        proc = subprocess.Popen([path, f"--remote-debugging-port={random_port()}",
+                                 f"--user-data-dir={tempdir}", "--no-first-run"])
+        try:
+            yield proc
+        finally:
+            if os.name == 'posix':
+                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
+            else:
+                proc.terminate()
+            try:
+                proc.wait(10)
+            except subprocess.TimeoutExpired:
+                if os.name == 'posix':
+                    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+                else:
+                    proc.kill()

tests/server.py

@@ -7,15 +7,15 @@
 import threading
 from contextlib import closing
 from http import HTTPStatus
+from pathlib import Path
 from typing import Any, Callable, Dict, Generator, Generic, Optional, Set, Tuple, TypeVar, cast
 from urllib.parse import urlparse
 
-from playwright._impl._path_utils import get_file_dirname
 from twisted.internet import reactor as _twisted_reactor
 from twisted.internet.selectreactor import SelectReactor
 from twisted.web import http
 
-_dirname = get_file_dirname()
+_dirname = Path(__file__).parent.absolute()
 reactor = cast(SelectReactor, _twisted_reactor)
 
 

tests/test_other.py

@@ -0,0 +1,27 @@
+import pytest
+import subprocess
+from cdp_patches.input import SyncInput, AsyncInput
+
+
+def test_raises_from_pid():
+    with pytest.raises(TimeoutError):
+        sync_input = SyncInput(pid=-1, window_timeout=1)
+        sync_input.click("left", 100, 100)
+
+
+@pytest.mark.asyncio
+async def test_async_raises_from_pid():
+    with pytest.raises(TimeoutError):
+        sync_input = await AsyncInput(pid=-1, window_timeout=1)
+        await sync_input.click("left", 100, 100)
+
+
+@pytest.mark.asyncio
+async def test_async_from_pid(chrome_proc: subprocess.Popen):
+    sync_input = await AsyncInput(pid=chrome_proc.pid, window_timeout=1)
+    await sync_input.click("left", 100, 100)
+
+
+def test_sync_from_pid(chrome_proc: subprocess.Popen):
+    sync_input = SyncInput(pid=chrome_proc.pid, window_timeout=1)
+    sync_input.click("left", 100, 100)

tests/utils.py

@@ -0,0 +1,61 @@
+import os
+import sys
+import socket
+from contextlib import closing
+
+IS_POSIX = sys.platform.startswith(("darwin", "cygwin", "linux", "linux2"))
+
+
+def find_chrome_executable() -> str:
+    """
+    Finds the Chrome, Chrome beta, Chrome canary, Chromium executable
+
+    Returns
+    -------
+    executable_path :  str
+        the full file path to found executable
+
+    """
+    candidates = set()
+    if IS_POSIX:
+        for item in os.environ.get("PATH", "").split(os.pathsep):
+            for subitem in (
+                    "google-chrome",
+                    "chromium",
+                    "chromium-browser",
+                    "chrome",
+                    "google-chrome-stable",
+            ):
+                candidates.add(os.sep.join((item, subitem)))
+        if "darwin" in sys.platform:
+            candidates.update(
+                [
+                    "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
+                    "/Applications/Chromium.app/Contents/MacOS/Chromium",
+                ]
+            )
+    else:
+        for item in map(
+                os.environ.get,
+                ("PROGRAMFILES", "PROGRAMFILES(X86)", "LOCALAPPDATA", "PROGRAMW6432"),
+        ):
+            if item is not None:
+                for subitem in (
+                        "Google/Chrome/Application",
+                        "Google/Chrome Beta/Application",
+                        "Google/Chrome Canary/Application",
+                ):
+                    candidates.add(os.sep.join((item, subitem, "chrome.exe")))
+    for candidate in candidates:
+        if os.path.exists(candidate) and os.access(candidate, os.X_OK):
+            return os.path.normpath(candidate)
+    raise FileNotFoundError("Couldn't find installed Chrome or Chromium executable")
+
+
+def random_port(host: str = None) -> int:
+    if not host:
+        host = ''
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+        s.bind((host, 0))
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        return s.getsockname()[1]