test_sync.py

import asyncio
import json
import os
import shutil
import sys
import time
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from chromix.sync import SyncEndpoint, SyncGroup, prepare_sync

OPERATIONS = [{'kind': 'move', 'x': 0.25, 'y': 0.5, 'modifiers': [], 'buttons': [], 'button': 'none'},
 {'kind': 'text', 'text': '你好 🌍', 'cursorBefore': False},
 {'kind': 'keyDown', 'key': 'Enter', 'code': 'Enter', 'shortcut': 'Enter', 'modifiers': [], 'repeat': False, 'char': False, 'character': 13},
 {'kind': 'keyUp', 'key': 'Enter', 'code': 'Enter', 'shortcut': 'Enter', 'modifiers': [], 'repeat': False, 'char': False, 'character': 13}]


class FakeEndpoint:
    def __init__(self, slow=0, fail=False):
        self.queue, self.packets, self.calls = [], [], []
        self.role, self.epoch, self.seq = "", "", 0
        self.slow, self.fail = slow, fail

    async def connect(self):
        pass

    def close(self):
        self.role = ""

    async def request(self, method, **params):
        self.calls.append(method)
        if method == "begin":
            self.role, self.epoch, self.seq = params["role"], params["epoch"], params["seq"]
            self.queue.clear()
        if method == "stop":
            self.role = ""
        if method == "ping" and not self.role:
            raise RuntimeError("session expired")
        if method == "poll":
            operations = self.queue[:8]
            del self.queue[:8]
            return dict(operations=operations)
        if method == "apply":
            if self.slow:
                await asyncio.sleep(self.slow)
            if self.fail:
                raise RuntimeError("target lost")
            packet = params["packet"]
            assert self.role == "replay"
            assert packet["epoch"] == self.epoch
            assert packet["seq"] == self.seq + 1
            self.seq = packet["seq"]
            self.packets.append(packet)
            return dict(seq=packet["seq"])
        return {}


async def until(predicate):
    deadline = time.monotonic() + 3
    while time.monotonic() < deadline:
        if predicate():
            return
        await asyncio.sleep(.01)
    raise AssertionError("condition timed out")


class SyncTests(unittest.IsolatedAsyncioTestCase):
    async def test_order_pause_and_resume(self):
        leader, a, b = FakeEndpoint(), FakeEndpoint(), FakeEndpoint()
        group = SyncGroup(leader=leader, followers=[a, b])
        try:
            await group.start()
            leader.queue.extend(OPERATIONS)
            await until(lambda: len(a.packets) == len(b.packets) == 4)
            self.assertEqual([p["operation"] for p in a.packets], OPERATIONS)
            self.assertEqual(a.packets, b.packets)
            epoch = a.epoch
            await group.pause()
            leader.queue.append(OPERATIONS[0])
            await group.resume()
            leader.queue.append(OPERATIONS[1])
            await until(lambda: len(a.packets) == 5)
            self.assertNotEqual(a.epoch, epoch)
            self.assertEqual(a.packets[-1]["seq"], 1)
            self.assertEqual(a.packets[-1]["operation"], OPERATIONS[1])
        finally:
            await group.close()

    async def test_join_switch_remove(self):
        leader, a, b = FakeEndpoint(), FakeEndpoint(), FakeEndpoint()
        group = SyncGroup(leader=leader, followers=[a])
        try:
            await group.start()
            leader.queue.extend(OPERATIONS)
            await until(lambda: len(a.packets) == 4)
            await group.add_follower(b)
            leader.queue.append(OPERATIONS[0])
            await until(lambda: len(b.packets) == 1)
            self.assertEqual(b.packets[0]["seq"], 5)
            await group.set_leader(a)
            a.queue.append(OPERATIONS[1])
            await until(lambda: len(leader.packets) == 1)
            self.assertEqual(leader.packets[0]["seq"], 1)
            await group.remove_follower(b)
            a.queue.append(OPERATIONS[0])
            await until(lambda: len(leader.packets) == 2)
            self.assertEqual(b.role, "")
        finally:
            await group.close()

    async def test_failure_isolation(self):
        leader, good = FakeEndpoint(), FakeEndpoint()
        bad, slow = FakeEndpoint(fail=True), FakeEndpoint(slow=.3)
        group = SyncGroup(leader=leader, followers=[slow, good, bad])
        errors = []
        group.on("member_error", errors.append)
        try:
            await group.start()
            leader.queue.extend(OPERATIONS)
            await until(lambda: len(good.packets) == 4)
            self.assertEqual(len(slow.packets), 0)
            self.assertEqual(len(errors), 1)
            self.assertIs(errors[0]["browser"], bad)
        finally:
            await group.close()

    async def test_ownership(self):
        leader, a = FakeEndpoint(), FakeEndpoint()
        first = SyncGroup(leader=leader, followers=[a])
        second = SyncGroup(leader=leader)
        try:
            await first.start()
            with self.assertRaisesRegex(RuntimeError, "another SyncGroup"):
                await second.start()
            await second.close()
            self.assertEqual(leader.role, "capture")
            with self.assertRaisesRegex(ValueError, "Duplicate"):
                await first.add_follower(a)
        finally:
            await first.close()

    async def test_keepalive_and_one_producer(self):
        leader, follower = FakeEndpoint(), FakeEndpoint()
        group = SyncGroup(leader=leader, followers=[follower])
        try:
            await group.start()
            for _ in range(5):
                await group.pause()
                await group.resume()
            await until(lambda: "ping" in follower.calls)
            count = leader.calls.count("poll")
            await asyncio.sleep(.1)
            self.assertLess(leader.calls.count("poll") - count, 25)
        finally:
            await group.close()

    @unittest.skipUnless(sys.platform == "win32" and shutil.which("node"), "Windows + Node required")
    async def test_native_pipe_unicode_fragmentation(self):
        ep = SyncEndpoint()
        script = r'''
const net = require('net');
const operations = JSON.parse(process.argv[2]);
net.createServer(socket => {
  let buffer = Buffer.alloc(0);
  socket.on('data', chunk => {
    buffer = Buffer.concat([buffer, chunk]);
    while (buffer.length >= 4 && buffer.length >= buffer.readUInt32LE(0) + 4) {
      const size = buffer.readUInt32LE(0), req = JSON.parse(buffer.subarray(4, size+4));
      buffer = buffer.subarray(size+4);
      const body = Buffer.from(JSON.stringify(req.method === 'hello' ? {version:1,capabilities:[]} : {operations}));
      const frame = Buffer.alloc(body.length+4); frame.writeUInt32LE(body.length); body.copy(frame,4);
      socket.write(frame.subarray(0,2)); setTimeout(() => socket.write(frame.subarray(2)),2);
    }
  });
}).listen('\\\\.\\pipe\\uxr-sync-' + process.argv[1], () => console.log('ready'));
'''
        process = await asyncio.create_subprocess_exec(shutil.which("node"), "-e", script,
            ep.id, json.dumps(OPERATIONS), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
        try:
            self.assertEqual(await asyncio.wait_for(process.stdout.readline(), 5), b"ready\n")
            await ep.connect()
            results = await asyncio.gather(ep.request("poll"), ep.request("poll"))
            for result in results:
                self.assertEqual(result["operations"], OPERATIONS)
        finally:
            ep.close()
            process.terminate()
            await process.wait()

    @unittest.skipUnless(sys.platform == "win32", "Windows launch opt-in")
    async def test_launch_opt_in_preserves_browser_objects(self):
        from types import SimpleNamespace
        from unittest.mock import patch
        from chromix import api
        calls = []
        class Browser:
            pages = []
            def on(self, *_):
                pass
            async def close(self):
                pass
            async def new_context(self, **_):
                return Browser()
            async def new_page(self, **_):
                return Browser()
        async def launch(**options):
            calls.append(options)
            return Browser()
        async def stop():
            pass
        async def start():
            return SimpleNamespace(chromium=SimpleNamespace(launch=launch, launch_persistent_context=launch), stop=stop)
        modules = {"playwright": SimpleNamespace(), "playwright.async_api":
                   SimpleNamespace(async_playwright=lambda: SimpleNamespace(start=start))}
        with patch.dict(sys.modules, modules), patch.object(api, "_prepare", side_effect=lambda *a, **kw:
                (Path("fake-chrome.exe"), [], {}, None)), patch.object(api, "apply_font_env"), \
                patch.object(api, "_wrap_geometry"):
            for fn in (api.launch_async, api.launch_context_async, api.launch_persistent_context_async):
                options = dict(sync=True, headless=False, args=["--fingerprint=off"])
                if "persistent" in fn.__name__:
                    options["user_data_dir"] = "unused-profile"
                browser = await fn(**options)
                self.assertIsInstance(browser, Browser)
                self.assertIsInstance(browser.sync, SyncEndpoint)
                self.assertIn(browser.sync.args[0], calls[-1]["args"])
                self.assertNotIn("sync", calls[-1])
                await browser.close()

    def test_options(self):
        self.assertIsNone(prepare_sync(False, True))
        with self.assertRaises(ValueError):
            SyncEndpoint(id="../bad")
        with self.assertRaises((ValueError, RuntimeError)):
            prepare_sync(True, True)

    async def test_oversized_frame_and_timeout_close_connection(self):
        class Writer:
            closed = False
            def write(self, data):
                pass
            async def drain(self):
                pass
            def close(self):
                self.closed = True
        for oversized in (True, False):
            ep = SyncEndpoint(timeout=.05)
            reader, writer = asyncio.StreamReader(), Writer()
            ep._reader, ep._writer = reader, writer
            if oversized:
                reader.feed_data((1024 * 1024 + 1).to_bytes(4, "little"))
            with self.assertRaises(ValueError if oversized else asyncio.TimeoutError):
                await ep.request("hello")
            self.assertTrue(writer.closed)
            self.assertIsNone(ep._writer)


def require_input_desktop():
    """Read-only preflight; do not move focus or inject input on a developer PC."""
    import ctypes
    from ctypes import wintypes
    if sys.platform != "win32":
        raise RuntimeError("native synchronization acceptance requires Windows")
    user32 = ctypes.WinDLL("user32", use_last_error=True)
    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
    kernel32.GetCurrentThreadId.restype = wintypes.DWORD
    user32.GetThreadDesktop.argtypes = [wintypes.DWORD]
    user32.GetThreadDesktop.restype = wintypes.HANDLE
    user32.GetUserObjectInformationW.argtypes = [wintypes.HANDLE, ctypes.c_int,
        ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)]
    desktop = user32.GetThreadDesktop(kernel32.GetCurrentThreadId())
    receiving_input, needed = wintypes.BOOL(), wintypes.DWORD()
    if not desktop or not user32.GetUserObjectInformationW(
            desktop, 6, ctypes.byref(receiving_input), ctypes.sizeof(receiving_input), ctypes.byref(needed)):
        raise ctypes.WinError(ctypes.get_last_error())
    if not receiving_input.value:
        raise RuntimeError("runner needs an unlocked input desktop for native capture/replay")


@unittest.skipUnless(os.environ.get("INPUT_SYNC_CHECK_DESKTOP"), "CI desktop preflight opt-in")
class InputDesktopTests(unittest.TestCase):
    def test_runner_receives_desktop_input(self):
        require_input_desktop()


@unittest.skipUnless(os.environ.get("INPUT_SYNC_BROWSER"),
                     "set INPUT_SYNC_BROWSER to a matching rebuilt chrome.exe")
class NativeSyncTests(unittest.IsolatedAsyncioTestCase):
    """Exercise the C++ pipe, adapters and hooks, never a mock endpoint.

    Playwright observes page results and drives IME commit; SendInput exercises
    the Windows/Aura capture path. Run on an unlocked interactive Windows desktop.
    """
    async def test_native_capture_replay_and_window_lifetime(self):
        require_input_desktop()
        import ctypes
        from ctypes import wintypes
        import subprocess
        import tempfile
        from playwright.async_api import async_playwright

        self.assertEqual(sys.platform, "win32", "native acceptance requires Windows")
        executable = Path(os.environ["INPUT_SYNC_BROWSER"]).resolve(strict=True)
        endpoint = SyncEndpoint()
        session, epoch, seq = "native-test", "initial", 0
        native = ctypes.WinDLL("user32", use_last_error=True)

        class Mouse(ctypes.Structure):
            _fields_ = [("dx", wintypes.LONG), ("dy", wintypes.LONG),
                        ("data", wintypes.DWORD), ("flags", wintypes.DWORD),
                        ("time", wintypes.DWORD), ("extra", ctypes.c_size_t)]
        class Key(ctypes.Structure):
            _fields_ = [("vk", wintypes.WORD), ("scan", wintypes.WORD),
                        ("flags", wintypes.DWORD), ("time", wintypes.DWORD),
                        ("extra", ctypes.c_size_t)]
        class Payload(ctypes.Union):
            _fields_ = [("mouse", Mouse), ("key", Key)]
        class Input(ctypes.Structure):
            _fields_ = [("type", wintypes.DWORD), ("payload", Payload)]
        native.SendInput.argtypes = [wintypes.UINT, ctypes.POINTER(Input), ctypes.c_int]
        native.SendInput.restype = wintypes.UINT
        native.SetForegroundWindow.argtypes = [wintypes.HWND]
        native.GetForegroundWindow.restype = wintypes.HWND
        native.GetWindowThreadProcessId.argtypes = [wintypes.HWND, ctypes.POINTER(wintypes.DWORD)]
        native.IsWindowVisible.argtypes = [wintypes.HWND]
        native.GetWindowRect.argtypes = [wintypes.HWND, ctypes.POINTER(wintypes.RECT)]
        callback_type = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM)
        native.EnumWindows.argtypes = [callback_type, wintypes.LPARAM]

        def send_keys(*events):
            inputs = (Input * len(events))()
            for item, (vk, flags) in zip(inputs, events):
                item.type = 1
                item.payload.key = Key(vk, 0, flags, 0, 0)
            self.assertEqual(native.SendInput(len(inputs), inputs, ctypes.sizeof(Input)), len(inputs))

        def click_window(pid):
            handles = []
            @callback_type
            def visit(hwnd, _):
                owner = wintypes.DWORD()
                native.GetWindowThreadProcessId(hwnd, ctypes.byref(owner))
                if owner.value == pid and native.IsWindowVisible(hwnd):
                    handles.append(hwnd)
                return True
            native.EnumWindows(visit, 0)
            self.assertEqual(len(handles), 1, "expected one visible test browser window")
            hwnd = handles[0]
            native.SetForegroundWindow(hwnd)
            self.assertEqual(native.GetForegroundWindow(), hwnd, "Windows desktop must allow test focus")
            rect = wintypes.RECT()
            self.assertTrue(native.GetWindowRect(hwnd, ctypes.byref(rect)))
            native.SetCursorPos((rect.left + rect.right) // 2, (rect.top + rect.bottom) // 2)
            inputs = (Input * 2)()
            inputs[0].payload.mouse.flags = 0x0002  # MOUSEEVENTF_LEFTDOWN
            inputs[1].payload.mouse.flags = 0x0004  # MOUSEEVENTF_LEFTUP
            self.assertEqual(native.SendInput(2, inputs, ctypes.sizeof(Input)), 2)

        async def wait_windows(count):
            for _ in range(100):
                if (await endpoint.request("hello"))["windows"] == count:
                    return
                await asyncio.sleep(.05)
            self.fail(f"native target count did not become {count}")

        async def begin(role):
            nonlocal seq, epoch
            seq = 0
            epoch += "x"
            await endpoint.request("begin", session=session, epoch=epoch, role=role, seq=0)

        async def apply(operation, duplicate=False):
            nonlocal seq
            seq += 1
            packet = dict(version=1, session=session, epoch=epoch, seq=seq, operation=operation)
            self.assertEqual((await endpoint.request("apply", packet=packet))["seq"], seq)
            if duplicate:
                self.assertEqual((await endpoint.request("apply", packet=packet))["seq"], seq)

        async def drain():
            result = []
            while True:
                batch = (await endpoint.request("poll"))["operations"]
                result.extend(batch)
                if not batch:
                    return result

        html = """<style>body{margin:0}textarea{width:100vw;height:80vh}</style>
        <textarea autofocus></textarea><script>
        window.keys=[];window.clicks=0;
        document.addEventListener('keyup',e=>keys.push(e.key));
        document.addEventListener('click',()=>clicks++);
        </script>"""
        with tempfile.TemporaryDirectory(prefix="input-sync-") as profile:
            process = subprocess.Popen([str(executable), *endpoint.args,
                "--user-data-dir=" + profile, "--no-first-run", "--no-default-browser-check",
                "--remote-debugging-port=0", "--no-startup-window", "--keep-alive-for-test",
                "--window-size=900,700", "--force-device-scale-factor=1"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            browser = None
            try:
                port_file = Path(profile) / "DevToolsActivePort"
                for _ in range(200):
                    self.assertIsNone(process.poll(), "browser exited during startup")
                    if port_file.exists():
                        break
                    await asyncio.sleep(.1)
                port = port_file.read_text().splitlines()[0]
                async with async_playwright() as pw:
                    browser = await pw.chromium.connect_over_cdp("http://127.0.0.1:" + port)
                    # No window has ever registered: pipe lifetime starts with the process.
                    await endpoint.connect()
                    connection = endpoint._writer
                    await wait_windows(0)
                    with self.assertRaisesRegex(RuntimeError, "exactly one"):
                        await begin("capture")
                    context = browser.contexts[0]
                    page = await context.new_page()
                    await page.set_content(html)
                    await wait_windows(1)
                    await begin("capture")
                    click_window(process.pid)
                    await page.wait_for_function("document.activeElement.tagName === 'TEXTAREA'")
                    send_keys((0x41, 0), (0x41, 2), (0x42, 0), (0x42, 2))
                    await page.wait_for_function("document.querySelector('textarea').value === 'ab'")
                    await asyncio.sleep(.1)
                    captured = await drain()
                    self.assertTrue(any(op['kind'] == 'down' and op['button'] == 'left' for op in captured))
                    self.assertTrue(any(op['kind'] == 'keyDown' and op['code'] == 'KeyA' for op in captured))
                    for op in captured:
                        self.assertNotIn('flags', op)
                        if op['kind'].startswith('key'):
                            self.assertIsInstance(op['key'], str)
                            self.assertIsInstance(op['code'], str)
                    await endpoint.request("stop")
                    await page.locator('textarea').fill('')
                    await page.evaluate('clicks=0')
                    await begin("replay")
                    for op in captured:
                        await apply(op, duplicate=True)
                    await page.wait_for_function("document.querySelector('textarea').value === 'ab'")
                    await asyncio.sleep(.2)
                    self.assertEqual(await page.locator('textarea').input_value(), 'ab')
                    self.assertEqual(await page.evaluate('clicks'), 1)

                    # Actual renderer text-commit hook, including multi-byte text.
                    await endpoint.request("stop")
                    await begin("capture")
                    cdp = await context.new_cdp_session(page)
                    await cdp.send('Input.insertText', dict(text='你好 🌍'))
                    committed = await drain()
                    texts = [op for op in committed if op['kind'] == 'text']
                    self.assertEqual([op['text'] for op in texts], ['你好 🌍'])
                    await endpoint.request("stop")
                    await page.locator('textarea').fill('')
                    await begin("replay")
                    await apply(texts[0], duplicate=True)
                    await page.wait_for_function("document.querySelector('textarea').value === '你好 🌍'")
                    await asyncio.sleep(.2)
                    self.assertEqual(await page.locator('textarea').input_value(), '你好 🌍')

                    # Stop must release held keys; invalid protocol values fail closed.
                    shift = dict(kind='keyDown', key='Shift', code='ShiftLeft', shortcut='ShiftLeft',
                                 modifiers=['shift'], repeat=False, char=False, character=0)
                    await page.evaluate('keys=[]')
                    await apply(shift)
                    await endpoint.request("stop")
                    await page.wait_for_function("keys.includes('Shift')")
                    await begin('replay')
                    with self.assertRaisesRegex(RuntimeError, 'invalid modifiers'):
                        await apply(dict(shift, modifiers=['unknown']))
                    with self.assertRaisesRegex(RuntimeError, 'not active'):
                        await endpoint.request('ping')

                    # Additional windows stop a session, while the connection survives.
                    await begin('capture')
                    browser_cdp = await browser.new_browser_cdp_session()
                    extra = await browser_cdp.send('Target.createTarget', dict(url='about:blank', newWindow=True))
                    await wait_windows(2)
                    with self.assertRaisesRegex(RuntimeError, 'not active'):
                        await endpoint.request('poll')
                    with self.assertRaisesRegex(RuntimeError, 'exactly one'):
                        await begin('replay')
                    await browser_cdp.send('Target.closeTarget', extra)
                    await wait_windows(1)
                    await begin('capture')
                    await page.close()
                    await wait_windows(0)
                    self.assertIs(endpoint._writer, connection)
                    with self.assertRaisesRegex(RuntimeError, 'exactly one'):
                        await begin('capture')
                    await context.new_page()
                    await wait_windows(1)
                    await begin('capture')
                    self.assertEqual((await endpoint.request('poll'))['operations'], [])
                    self.assertIs(endpoint._writer, connection)
                    await endpoint.request('stop')
                    await browser_cdp.send('Browser.close')
                    await asyncio.get_running_loop().run_in_executor(None, lambda: process.wait(timeout=10))
                    self.assertEqual(process.returncode, 0)
            finally:
                endpoint.close()
                if process.poll() is None:
                    subprocess.run(['taskkill', '/PID', str(process.pid), '/T', '/F'],
                                   capture_output=True, timeout=15)
                    process.wait(timeout=10)


if __name__ == "__main__":
    unittest.main()