Merge from workspace to draft/toga-web-testing

web-testbed/run_tests.py

@@ -0,0 +1,69 @@
+import os
+import signal
+import subprocess
+import sys
+import time
+from shutil import which
+
+SERVER_CMD = ["briefcase", "run", "web", "--no-browser"]
+TEST_CMD = ["pytest", "tests"]
+STARTUP_WAIT_SECS = float(os.getenv("SERVER_STARTUP_SECS", "5.0"))
+
+IS_WINDOWS = os.name == "nt"
+CREATE_NEW_PROCESS_GROUP = 0x00000200 if IS_WINDOWS else 0
+
+
+def start_server():
+    if which(SERVER_CMD[0]) is None:
+        print(f"Error: '{SERVER_CMD[0]}' not found on PATH.", file=sys.stderr)
+        sys.exit(127)
+    kwargs = {}
+    if IS_WINDOWS:
+        kwargs["creationflags"] = CREATE_NEW_PROCESS_GROUP
+    kwargs["stdout"] = subprocess.PIPE
+    kwargs["stderr"] = subprocess.STDOUT
+    return subprocess.Popen(SERVER_CMD, **kwargs)
+
+
+def stop_server(proc, timeout=10):
+    if proc.poll() is not None:
+        return
+    try:
+        if IS_WINDOWS:
+            # Try to be gentle first
+            try:
+                proc.send_signal(signal.CTRL_BREAK_EVENT)
+            except Exception:
+                proc.terminate()
+        else:
+            proc.send_signal(signal.SIGINT)
+        proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        try:
+            proc.terminate()
+            proc.wait(timeout=3)
+        except Exception:
+            proc.kill()
+
+
+def main():
+    print("> Starting web server:", " ".join(SERVER_CMD))
+    server = start_server()
+
+    try:
+        time.sleep(STARTUP_WAIT_SECS)
+        print("> Running tests:", " ".join(TEST_CMD))
+        result = subprocess.run(TEST_CMD)
+        exit_code = result.returncode
+    except KeyboardInterrupt:
+        print("\n> Interrupted by user.", file=sys.stderr)
+        exit_code = 130
+    finally:
+        print("> Shutting down web server…")
+        stop_server(server)
+
+    sys.exit(exit_code)
+
+
+if __name__ == "__main__":
+    main()

web-testbed/src/testbed/app.py

@@ -1,122 +1,22 @@
-import os
-import types
-from unittest.mock import Mock
-
 import toga
 from toga.style import Pack
 from toga.style.pack import COLUMN
 
-try:
-    import js
-except ModuleNotFoundError:
-    js = None
-try:
-    from pyodide.ffi import create_proxy, to_js
-except ModuleNotFoundError:
-    pyodide = None
-
-
-def _truthy(v) -> bool:
-    return str(v).strip().lower() in {"1", "true", "yes", "on"}
-
-
-def _web_testing_enabled() -> bool:
-    if _truthy(os.getenv("TOGA_WEB_TESTING")):
-        return True
-
-    if js is not None:
-        try:
-            if _truthy(getattr(js.window, "TOGA_WEB_TESTING", "")):
-                return True
-            qs = str(getattr(js.window, "location", None).search or "")
-            # enable if ?toga_web_testing=1 in url
-            if "toga_web_testing" in qs.lower():
-                return True
-        except Exception:
-            pass
-
-    return False
+from .web_test_harness import WebTestHarness
 
 
 class HelloWorld(toga.App):
     def startup(self):
         main_box = toga.Box(style=Pack(direction=COLUMN))
         self.label = toga.Label(id="myLabel", text="Test App - Toga Web Testing")
 
-        if _web_testing_enabled() and js is not None and create_proxy is not None:
-            self.my_objs = {}
-            js.window.test_cmd = create_proxy(self.cmd_test)
+        self.web_test = WebTestHarness(self)
 
         main_box.add(self.label)
         self.main_window = toga.MainWindow(title=self.formal_name)
         self.main_window.content = main_box
         self.main_window.show()
 
-    def cmd_test(self, code):
-        env = {"self": self, "toga": toga, "my_objs": self.my_objs, "Mock": Mock}
-        local = {}
-        try:
-            exec(code, env, local)
-            result = local.get("result", env.get("result"))
-            envelope = self._serialise_payload(result)
-            return to_js(envelope, dict_converter=js.Object.fromEntries)
-        except Exception as e:
-            return to_js(
-                {"type": "error", "value": str(e)}, dict_converter=js.Object.fromEntries
-            )
-
-    def _serialise_payload(self, x):
-        # primitives
-        if x is None:
-            return {"type": "none", "value": None}
-        if isinstance(x, bool):
-            return {"type": "bool", "value": x}
-        if isinstance(x, int):
-            return {"type": "int", "value": x}
-        if isinstance(x, float):
-            return {"type": "float", "value": x}
-        if isinstance(x, str):
-            return {"type": "str", "value": x}
-
-        # containers
-        if isinstance(x, list):
-            return {"type": "list", "items": [self._serialise_payload(i) for i in x]}
-        if isinstance(x, tuple):
-            return {"type": "tuple", "items": [self._serialise_payload(i) for i in x]}
-        if isinstance(x, dict):
-            items = []
-            for k, v in x.items():
-                if k is None:
-                    key_env = {"type": "none", "value": None}
-                elif isinstance(k, bool):
-                    key_env = {"type": "bool", "value": k}
-                elif isinstance(k, int):
-                    key_env = {"type": "int", "value": k}
-                elif isinstance(k, float):
-                    key_env = {"type": "float", "value": k}
-                elif isinstance(k, str):
-                    key_env = {"type": "str", "value": k}
-                else:
-                    key_env = {"type": "str", "value": str(k)}
-                items.append([key_env, self._serialise_payload(v)])
-            return {"type": "dict", "items": items}
-
-        # references by id
-        obj_id = self._key_for(x)
-        is_callable = callable(x) or isinstance(
-            x, (types.FunctionType, types.MethodType)
-        )
-        return {"type": "callable" if is_callable else "object", "id": obj_id}
-
-    def _key_for(self, x):
-        for k, v in self.my_objs.items():
-            if v is x:
-                return k
-        # If not registered, register it
-        k = str(id(x))
-        self.my_objs[k] = x
-        return k
-
 
 def main():
     return HelloWorld()

web-testbed/src/testbed/web_test_harness.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import os
+import types
+from unittest.mock import Mock
+
+import toga
+
+try:
+    import js
+except ModuleNotFoundError:
+    js = None
+
+try:
+    from pyodide.ffi import create_proxy, to_js
+except ModuleNotFoundError:
+    pyodide = None
+    create_proxy = None
+    to_js = None
+
+
+def _truthy(v) -> bool:
+    return str(v).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def web_testing_enabled() -> bool:
+    if _truthy(os.getenv("TOGA_WEB_TESTING")):
+        return True
+
+    if js is not None:
+        try:
+            if _truthy(getattr(js.window, "TOGA_WEB_TESTING", "")):
+                return True
+            qs = str(getattr(js.window, "location", None).search or "")
+            if "toga_web_testing" in qs.lower():
+                return True
+        except Exception:
+            pass
+    return False
+
+
+class WebTestHarness:
+    def __init__(self, app, *, expose_name: str = "test_cmd"):
+        self.app = app
+        self.my_objs = {}
+        self.app.my_objs = self.my_objs
+
+        self._js_available = (
+            js is not None and create_proxy is not None and to_js is not None
+        )
+        if self._js_available and web_testing_enabled():
+            js.window.test_cmd = create_proxy(self.cmd_test)
+
+    def cmd_test(self, code):
+        try:
+            env = globals().copy()
+            env.update(locals())
+
+            env["self"] = self.app
+            env["toga"] = toga
+            env["my_objs"] = self.my_objs
+            env["Mock"] = Mock
+
+            exec(code, env, env)
+            result = env.get("result")
+            envelope = self._serialise_payload(result)
+            return to_js(envelope, dict_converter=js.Object.fromEntries)
+        except Exception as e:
+            return to_js(
+                {"type": "error", "value": str(e)}, dict_converter=js.Object.fromEntries
+            )
+
+    def _serialise_payload(self, x):
+        # primitives
+        if x is None:
+            return {"type": "none", "value": None}
+        if isinstance(x, bool):
+            return {"type": "bool", "value": x}
+        if isinstance(x, int):
+            return {"type": "int", "value": x}
+        if isinstance(x, float):
+            return {"type": "float", "value": x}
+        if isinstance(x, str):
+            return {"type": "str", "value": x}
+
+        # containers
+        if isinstance(x, list):
+            return {"type": "list", "items": [self._serialise_payload(i) for i in x]}
+        if isinstance(x, tuple):
+            return {"type": "tuple", "items": [self._serialise_payload(i) for i in x]}
+        if isinstance(x, dict):
+            items = []
+            for k, v in x.items():
+                if k is None:
+                    key_env = {"type": "none", "value": None}
+                elif isinstance(k, bool):
+                    key_env = {"type": "bool", "value": k}
+                elif isinstance(k, int):
+                    key_env = {"type": "int", "value": k}
+                elif isinstance(k, float):
+                    key_env = {"type": "float", "value": k}
+                elif isinstance(k, str):
+                    key_env = {"type": "str", "value": k}
+                else:
+                    key_env = {"type": "str", "value": str(k)}
+                items.append([key_env, self._serialise_payload(v)])
+            return {"type": "dict", "items": items}
+
+        # references by id
+        obj_id = self._key_for(x)
+        is_callable = callable(x) or isinstance(
+            x, (types.FunctionType, types.MethodType)
+        )
+        return {"type": "callable" if is_callable else "object", "id": obj_id}
+
+    def _key_for(self, x):
+        for k, v in self.my_objs.items():
+            if v is x:
+                return k
+        # If not registered, register it
+        k = str(id(x))
+        self.my_objs[k] = x
+        return k

web-testbed/tests/conftest.py

@@ -1,30 +1,14 @@
 # from pytest import fixture, register_assert_rewrite, skip
-# import toga
-
 import pytest
 
-from .tests_backend.playwright_page import BackgroundPage
-from .tests_backend.proxies.app_proxy import AppProxy
-from .tests_backend.proxies.base_proxy import BaseProxy
-from .tests_backend.widgets.base import SimpleProbe
-
-
-@pytest.fixture(scope="session")
-def page():
-    p = BackgroundPage()
-    return p
-
+import toga
 
-# Inject Playwright page object into
-@pytest.fixture(scope="session", autouse=True)
-def _wire_page(page):
-    BaseProxy.page_provider = staticmethod(lambda: page)
-    SimpleProbe.page_provider = staticmethod(lambda: page)
+pytest_plugins = ["tests.tests_backend.web_test_patch"]
 
 
 @pytest.fixture(scope="session")
 def app():
-    return AppProxy()
+    return toga.App.app()
 
 
 @pytest.fixture(scope="session")

web-testbed/tests/tests_backend/playwright_page.py

@@ -29,28 +29,23 @@ def _run(self):
         self._loop.close()
 
     async def _bootstrap(self):
-        try:
-            self._play = await async_playwright().start()
-            self._browser = await self._play.chromium.launch(headless=True)
-            self._context = await self._browser.new_context()
-            await self._context.add_init_script("window.TOGA_WEB_TESTING = true;")
-
-            self._page = await self._context.new_page()
-
-            await self._page.goto(
-                "http://localhost:8080", wait_until="load", timeout=30_000
-            )
-
-            await self._page.wait_for_function(
-                "() => typeof window.test_cmd === 'function'"
-            )
-
-            self._alock = asyncio.Lock()
-        except Exception:
-            raise
-        finally:
-            self._alock = asyncio.Lock()
-            self._ready.set()
+        self._play = await async_playwright().start()
+        self._browser = await self._play.chromium.launch(headless=True)
+        self._context = await self._browser.new_context()
+        await self._context.add_init_script("window.TOGA_WEB_TESTING = true;")
+
+        self._page = await self._context.new_page()
+
+        await self._page.goto(
+            "http://localhost:8080", wait_until="load", timeout=30_000
+        )
+
+        await self._page.wait_for_function(
+            "() => typeof window.test_cmd === 'function'"
+        )
+
+        self._alock = asyncio.Lock()
+        self._ready.set()
 
     async def _eval(self, js, *args):
         async with self._alock: