fix(wpt): support native touch pointer actions

moli-benchmark/moli_benchmark/wpt_cross/native_input.py

@@ -113,6 +113,7 @@ def key_description(value: str, *, shift: bool = False) -> Key:
 @dataclass
 class Pointer:
     kind: str = "mouse"
+    touch_id: int | None = None
     x: float = 0
     y: float = 0
     buttons: list[int] = field(default_factory=list)
@@ -141,8 +142,8 @@ def validate_actions(sources: Any) -> None:
             raise ValueError("source actions must be a list")
         if source["type"] == "pointer":
             parameters = source.get("parameters", {})
-            if not isinstance(parameters, dict) or parameters.get("pointerType", "mouse") not in {"mouse", "pen"}:
-                raise ValueError("native WPT input supports mouse and pen pointer sources")
+            if not isinstance(parameters, dict) or parameters.get("pointerType", "mouse") not in {"mouse", "pen", "touch"}:
+                raise ValueError("native WPT input supports mouse, pen and touch pointer sources")
         for action in actions:
             if not isinstance(action, dict) or action.get("type") not in allowed[source["type"]]:
                 raise ValueError("unsupported input action")
@@ -155,6 +156,12 @@ def validate_actions(sources: Any) -> None:
                 type(action.get("button")) is not int or not 0 <= action["button"] <= 4
             ):
                 raise ValueError("unsupported pointer button")
+            if (
+                kind in {"pointerDown", "pointerUp"}
+                and source.get("parameters", {}).get("pointerType") == "touch"
+                and action["button"] != 0
+            ):
+                raise ValueError("native touch input requires the primary button")
             if kind in {"pointerMove", "scroll"}:
                 fields = ("x", "y", "deltaX", "deltaY") if kind == "scroll" else ("x", "y")
                 if any(type(action.get(name)) is not int for name in fields):
@@ -403,10 +410,39 @@ class NativeInput:
             **extra,
         })
 
+    async def touch(self, pointer: Pointer, event: str, **extra: Any) -> None:
+        if pointer.touch_id is None:
+            raise ValueError("touch pointer has no contact identifier")
+        if event == "touchStart":
+            if pointer.buttons:
+                return
+            pointer.buttons.append(0)
+        elif event == "touchEnd":
+            if not pointer.buttons:
+                return
+        elif not pointer.buttons:
+            # Moving an unpressed touch source changes its next contact position;
+            # touch screens do not produce hover events.
+            return
+        await self.command("Input.dispatchTouchEvent", {
+            "type": event, "modifiers": self.modifiers(),
+            # Chromium updates only the supplied contacts. Sending the whole
+            # active set would race with other sources' concurrent timed moves.
+            "touchPoints": [{
+                "id": pointer.touch_id, "x": pointer.x, "y": pointer.y,
+                "force": 0 if event == "touchEnd" else 1, **extra,
+            }],
+        })
+        if event == "touchEnd":
+            pointer.buttons.clear()
+
     async def reset_actions(self) -> None:
         for source in reversed(list(self.keyboards)):
             await self.release_keyboard(source)
         for pointer in self.pointers.values():
+            if pointer.kind == "touch":
+                await self.touch(pointer, "touchEnd")
+                continue
             for button in reversed(list(pointer.buttons)):
                 await self.mouse(pointer, "mouseReleased", button)
         self.pointers.clear()
@@ -446,12 +482,15 @@ class NativeInput:
             sources = request["actions"]
             validate_actions(sources)
             await self.reset_actions()
+            next_touch_id = 0
             for source in sources:
                 if source["type"] == "pointer":
                     pointer_type = source.get("parameters", {}).get("pointerType", "mouse")
-                    if pointer_type not in {"mouse", "pen"}:
-                        raise ValueError(f"native WPT pointer type is not supported: {pointer_type}")
-                    self.pointers[source["id"]] = Pointer(kind=pointer_type)
+                    pointer = Pointer(kind=pointer_type)
+                    if pointer_type == "touch":
+                        pointer.touch_id = next_touch_id
+                        next_touch_id += 1
+                    self.pointers[source["id"]] = pointer
                 elif source["type"] not in {"key", "none", "wheel"}:
                     raise ValueError(f"unsupported input source: {source['type']}")
             ticks = max((len(source["actions"]) for source in sources), default=0)
@@ -566,10 +605,13 @@ class NativeInput:
         if "pressure" in action:
             extra["force"] = action["pressure"]
         if kind in {"pointerDown", "pointerUp"}:
-            await self.mouse(
-                pointer, "mousePressed" if kind == "pointerDown" else "mouseReleased",
-                action["button"], **extra,
-            )
+            if pointer.kind == "touch":
+                await self.touch(pointer, "touchStart" if kind == "pointerDown" else "touchEnd", **extra)
+            else:
+                await self.mouse(
+                    pointer, "mousePressed" if kind == "pointerDown" else "mouseReleased",
+                    action["button"], **extra,
+                )
         elif kind == "pointerMove":
             origin = action.get("origin", "viewport")
             if origin == "pointer":
@@ -587,6 +629,9 @@ class NativeInput:
                 await asyncio.sleep(max(0, started + duration * step / steps - time.perf_counter()))
                 pointer.x = start_x + (x - start_x) * step / steps
                 pointer.y = start_y + (y - start_y) * step / steps
-                await self.mouse(pointer, "mouseMoved", **extra)
+                if pointer.kind == "touch":
+                    await self.touch(pointer, "touchMove", **extra)
+                else:
+                    await self.mouse(pointer, "mouseMoved", **extra)
         else:
             raise ValueError(f"unsupported pointer action: {kind}")

moli-benchmark/tests/test_wpt_native_input.py

@@ -121,6 +121,128 @@ class NativeInputTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual([event["type"] for event in self.input_events()], ["rawKeyDown", "keyUp"])
         self.assertEqual(self.driver.modifiers(), 0)
 
+    async def test_touch_actions_do_not_hover_or_release_another_source(self):
+        async def point(context_id, request_id, token, origin, x, y):
+            return x, y
+
+        self.driver.point = point
+        await self.driver.perform(1, {"id": 1, "token": "document", "kind": "actions", "actions": [
+            {"id": "first", "type": "pointer", "parameters": {"pointerType": "touch"}, "actions": [
+                {"type": "pointerMove", "x": 10, "y": 20},
+                {"type": "pointerDown", "button": 0},
+                {"type": "pointerDown", "button": 0},
+                {"type": "pointerUp", "button": 0},
+                {"type": "pointerUp", "button": 0},
+                {"type": "pointerMove", "x": 100, "y": 200},
+            ]},
+            {"id": "second", "type": "pointer", "parameters": {"pointerType": "touch"}, "actions": [
+                {"type": "pointerMove", "x": 30, "y": 40},
+                {"type": "pointerDown", "button": 0},
+                {"type": "pause"}, {"type": "pause"},
+                {"type": "pointerMove", "x": 35, "y": 45},
+                {"type": "pointerUp", "button": 0},
+            ]},
+        ]})
+        events = self.input_events()
+        self.assertEqual(
+            [(event["type"], [point["id"] for point in event["touchPoints"]]) for event in events],
+            [("touchStart", [0]), ("touchStart", [1]), ("touchEnd", [0]),
+             ("touchMove", [1]), ("touchEnd", [1])],
+        )
+        self.assertEqual(events[0]["touchPoints"][0]["x"], 10)
+        self.assertEqual(events[-1]["touchPoints"][0]["x"], 35)
+        self.assertTrue(all(not pointer.buttons for pointer in self.driver.pointers.values()))
+
+    async def test_concurrent_timed_touch_moves_update_only_their_own_contact(self):
+        async def point(context_id, request_id, token, origin, x, y):
+            return x, y
+
+        self.driver.point = point
+        await self.driver.perform(1, {"id": 1, "token": "document", "kind": "actions", "actions": [
+            {"id": f"touch-{index}", "type": "pointer", "parameters": {"pointerType": "touch"}, "actions": [
+                {"type": "pointerDown", "button": 0},
+                {"type": "pointerMove", "x": 30 * (index + 1), "y": 40, "duration": 48},
+                {"type": "pointerUp", "button": 0},
+            ]} for index in range(2)
+        ]})
+        active = set()
+        moves = {0: [], 1: []}
+        for event in self.input_events():
+            self.assertEqual(len(event["touchPoints"]), 1)
+            point = event["touchPoints"][0]
+            identifier = point["id"]
+            if event["type"] == "touchStart":
+                self.assertNotIn(identifier, active)
+                active.add(identifier)
+            elif event["type"] == "touchMove":
+                self.assertEqual(active, {0, 1})
+                moves[identifier].append((point["x"], point["y"]))
+            else:
+                self.assertIn(identifier, active)
+                active.remove(identifier)
+        self.assertEqual(active, set())
+        self.assertEqual(moves, {0: [(10, 40 / 3), (20, 80 / 3), (30, 40)],
+                                 1: [(20, 40 / 3), (40, 80 / 3), (60, 40)]})
+
+    async def test_reset_releases_touch_contacts_after_a_lost_start_response(self):
+        await self.driver.key("keyboard", "\ue008", True)
+        first = Pointer(kind="touch", touch_id=11, x=10, y=20)
+        second = Pointer(kind="touch", touch_id=12, x=30, y=40)
+        self.driver.pointers = {"first": first, "second": second}
+        await self.driver.touch(first, "touchStart")
+        command = self.client.command
+
+        async def lost_response(method, params=None, **kwargs):
+            await command(method, params, **kwargs)
+            if method == "Input.dispatchTouchEvent" and params["type"] == "touchStart":
+                raise RawCdpError("start response lost")
+            return SimpleNamespace(response={"result": {}})
+
+        self.client.command = lost_response
+        with self.assertRaisesRegex(RawCdpError, "start response lost"):
+            await self.driver.touch(second, "touchStart")
+        await self.driver.reset_actions()
+        events = [event for event in self.input_events() if event["type"].startswith("touch")]
+        self.assertEqual([(event["type"], event["touchPoints"][0]["id"]) for event in events],
+                         [("touchStart", 11), ("touchStart", 12), ("touchEnd", 11), ("touchEnd", 12)])
+        self.assertEqual([event["modifiers"] for event in events], [8, 8, 0, 0])
+        self.assertEqual(self.driver.pointers, {})
+
+    async def test_invalid_touch_actions_preserve_previously_pressed_contacts(self):
+        pointer = Pointer(kind="touch", touch_id=11)
+        self.driver.pointers["held"] = pointer
+        await self.driver.touch(pointer, "touchStart")
+        commands = list(self.client.commands)
+        for invalid in ({"type": "pointerUp", "button": 1},
+                        {"type": "pointerMove", "x": 10, "y": 20, "width": 10}):
+            with self.subTest(action=invalid), self.assertRaises(ValueError):
+                await self.driver.perform(1, {"id": 1, "token": "document", "kind": "actions", "actions": [
+                    {"id": "touch", "type": "pointer", "parameters": {"pointerType": "touch"}, "actions": [
+                        {"type": "pointerDown", "button": 0}, invalid,
+                    ]},
+                ]})
+            self.assertEqual(self.client.commands, commands)
+            self.assertEqual(pointer.buttons, [0])
+
+    async def test_failed_touch_release_does_not_forget_the_pressed_contact(self):
+        pointer = Pointer(kind="touch", touch_id=11)
+        self.driver.pointers["held"] = pointer
+        await self.driver.touch(pointer, "touchStart")
+        command = self.client.command
+
+        async def rejected_release(method, params=None, **kwargs):
+            await command(method, params, **kwargs)
+            raise RawCdpError("touch release rejected")
+
+        self.client.command = rejected_release
+        with self.assertRaisesRegex(RawCdpError, "touch release rejected"):
+            await self.driver.touch(pointer, "touchEnd")
+        self.assertEqual(pointer.buttons, [0])
+        self.client.command = command
+        await self.driver.reset_actions()
+        self.assertEqual(pointer.buttons, [])
+        self.assertEqual(self.driver.pointers, {})
+
     async def test_pen_move_keeps_pressure_without_reporting_a_button_change(self):
         pointer = Pointer(kind="pen", x=10, y=20)
         await self.driver.mouse(pointer, "mousePressed", 0, force=0.36)