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

"""Browser proxy endpoints: /tasks/{id}/browser/{state,input,control}.

The control plane forwards to the MCP remote service over the Unix
socket in the task worktree. Here the socket server is the real
playwright_remote module with a stub browser; no Chromium, no
worker, no model.
"""

import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import unittest
from pathlib import Path

try:
    from cloud.auth import AuthStore
    from cloud.playwright_remote import RemoteState, start_remote_server
    from cloud.server import create_server
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.auth import AuthStore
    from cloud.playwright_remote import RemoteState, start_remote_server
    from cloud.server import create_server


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


class BrowserProxyTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        self.auth = AuthStore(self.db_path)
        self.owner_id, self.owner_key = self.auth.create_user("owner")
        self.outsider_id, self.outsider_key = self.auth.create_user("o")
        self.server = create_server(self.db_path, "127.0.0.1", 0)
        self.worktrees = os.path.join(self.tmp.name, "host-worktrees")
        os.makedirs(self.worktrees)
        self.server.host_worktrees_dir = self.worktrees
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True)
        self.thread.start()
        status, self.project = self._req(
            "POST", "/projects", {"name": "demo"}, self.owner_key)
        self.assertEqual(status, 201)
        self._task("task-b1")
        wt = os.path.join(self.worktrees, "task-b1")
        os.makedirs(wt)
        self.remote = RemoteState(_StubBrowser())
        self.sock_server = start_remote_server(
            self.remote, os.path.join(wt, ".browser-remote.sock"))
        self._task("task-nosock")
        os.makedirs(os.path.join(self.worktrees, "task-nosock"))

    def tearDown(self):
        self.sock_server.shutdown()
        self.sock_server.server_close()
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)
        try:
            self.auth.conn.close()
        except Exception:
            pass
        self.tmp.cleanup()

    def _task(self, task_id):
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "INSERT INTO tasks(id, project_id, prompt, status)"
                " VALUES(?, ?, ?, ?)",
                (task_id, self.project["id"], "do it", "running"))
            conn.commit()
        finally:
            conn.close()

    def _req(self, method, path, body=None, key=None):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        hdrs = {}
        payload = None
        if body is not None:
            payload = json.dumps(body).encode("utf-8")
            hdrs["Content-Type"] = "application/json"
        if key is not None:
            hdrs["X-API-Key"] = key
        try:
            conn.request(method, path, body=payload, headers=hdrs)
            resp = conn.getresponse()
            data = resp.read()
        finally:
            conn.close()
        try:
            parsed = json.loads(data.decode("utf-8")) if data else None
        except ValueError:
            parsed = None
        return resp.status, parsed

    def test_state_input_control_flow(self):
        status, body = self._req(
            "GET", "/tasks/task-b1/browser/state", key=self.owner_key)
        self.assertEqual(status, 200, body)
        self.assertEqual(body["holder"], "agent")
        self.assertEqual(body["url"], "https://example.com")
        status, body = self._req(
            "POST", "/tasks/task-b1/browser/input",
            {"type": "mouse_click", "x": 1, "y": 2}, self.owner_key)
        self.assertEqual(status, 409, body)
        status, body = self._req(
            "POST", "/tasks/task-b1/browser/control",
            {"holder": "user"}, self.owner_key)
        self.assertEqual((status, body["holder"]), (200, "user"))
        status, body = self._req(
            "POST", "/tasks/task-b1/browser/input",
            {"type": "key", "key": "Enter"}, self.owner_key)
        self.assertEqual((status, body["queued"]), (200, True))
        status, body = self._req(
            "POST", "/tasks/task-b1/browser/control",
            {"holder": "agent"}, self.owner_key)
        self.assertEqual(body["holder"], "agent")

    def test_forbidden_and_no_session(self):
        status, _ = self._req(
            "GET", "/tasks/task-b1/browser/state", key=self.outsider_key)
        self.assertEqual(status, 403)
        status, body = self._req(
            "GET", "/tasks/task-nosock/browser/state", key=self.owner_key)
        self.assertEqual(status, 503, body)


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