codex-app-server/cloud/tests/test_playwright_remote.py

"""Tests for the Playwright remote-control service (fake only).

Covers: holder transitions + input gating, the Unix-socket HTTP
endpoints (state/control/input/stream), and the main-thread
input execution path. No real Chromium: the CDP/screencast path
is exercised through fakes, real-browser proof lives in the
production verification task.
"""

import json
import os
import socket
import sys
import tempfile
import threading
import unittest
import urllib.request

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from playwright_remote import (RemoteState, execute_input,  # noqa: E402
                               start_remote_server)


class _StubBrowser:
    def __init__(self):
        self.calls = []

    def cached_info(self):
        return {"url": "https://example.com", "title": "T",
                "started": True}

    def click_at(self, x, y):
        self.calls.append(("click_at", x, y))
        return {"ok": True}

    def mouse_move(self, x, y):
        self.calls.append(("mouse_move", x, y))
        return {"ok": True}

    def key_press(self, key):
        self.calls.append(("key_press", key))
        return {"ok": True}

    def type_text(self, text):
        self.calls.append(("type_text", text))
        return {"ok": True}


class RemoteStateTest(unittest.TestCase):
    def test_holder_and_input_gating(self):
        state = RemoteState(_StubBrowser())
        self.assertEqual(state.get_holder(), "agent")
        with self.assertRaises(PermissionError):
            state.enqueue_input({"type": "mouse_click", "x": 1, "y": 2})
        state.set_holder("user")
        state.enqueue_input({"type": "mouse_click", "x": 10, "y": 20})
        item = state.inputs.get_nowait()
        execute_input(state.browser, item)
        self.assertIn(("click_at", 10, 20), state.browser.calls)
        with self.assertRaises(ValueError):
            state.set_holder("nobody")
        with self.assertRaises(ValueError):
            execute_input(state.browser, {"type": "teleport"})

    def test_state_dict(self):
        state = RemoteState(_StubBrowser())
        state.set_frame("abc")
        d = state.state_dict()
        self.assertEqual(d["holder"], "agent")
        self.assertTrue(d["streaming"])
        self.assertEqual(d["url"], "https://example.com")


class HttpServiceTest(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.browser = _StubBrowser()
        self.state = RemoteState(self.browser)
        self.sock = os.path.join(self._tmp.name, "remote.sock")
        self.server = start_remote_server(self.state, self.sock)

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self._tmp.cleanup()

    def _unix_get(self, path):
        import http.client

        class UnixConn(http.client.HTTPConnection):
            def connect(self):
                self.sock = socket.socket(socket.AF_UNIX,
                                          socket.SOCK_STREAM)
                self.sock.connect(self._sock_path)

        conn = UnixConn("localhost", timeout=5)
        conn._sock_path = self.sock
        conn.request("GET", path)
        resp = conn.getresponse()
        return resp.status, json.loads(resp.read().decode("utf-8"))

    def _unix_post(self, path, obj):
        import http.client

        class UnixConn(http.client.HTTPConnection):
            def connect(self):
                self.sock = socket.socket(socket.AF_UNIX,
                                          socket.SOCK_STREAM)
                self.sock.connect(self._sock_path)

        conn = UnixConn("localhost", timeout=5)
        conn._sock_path = self.sock
        body = json.dumps(obj).encode("utf-8")
        conn.request("POST", path, body=body,
                     headers={"Content-Type": "application/json"})
        resp = conn.getresponse()
        return resp.status, json.loads(resp.read().decode("utf-8"))

    def test_state_control_input_flow(self):
        status, body = self._unix_get("/state")
        self.assertEqual(status, 200)
        self.assertEqual(body["holder"], "agent")
        status, body = self._unix_post(
            "/input", {"type": "mouse_click", "x": 1, "y": 2})
        self.assertEqual(status, 409)
        status, body = self._unix_post("/control", {"holder": "user"})
        self.assertEqual((status, body["holder"]), (200, "user"))
        status, body = self._unix_post(
            "/input", {"type": "key", "key": "Enter"})
        self.assertEqual((status, body["queued"]), (200, True))
        self.assertEqual(self.state.inputs.get_nowait()["key"], "Enter")
        status, body = self._unix_post("/control", {"holder": "agent"})
        self.assertEqual(body["holder"], "agent")

    def test_stream_emits_latest_frame(self):
        self.state.set_frame("FRAMEBYTES")
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.settimeout(5)
        sock.connect(self.sock)
        sock.sendall(b"GET /stream HTTP/1.1\r\nHost: x\r\n\r\n")
        buf = b""
        while b"\n\n" not in buf:
            buf += sock.recv(4096)
        head, _, rest = buf.partition(b"\n\n")
        self.assertIn(b"text/event-stream", head)
        first = rest
        while b"\n\n" not in first:
            first += sock.recv(4096)
        payload = json.loads(first.split(b"data: ", 1)[1].split(b"\n\n")[0])
        self.assertEqual(payload["image"], "FRAMEBYTES")
        sock.close()


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