"""Tests for the Playwright browser MCP (fake browser only).

No real Chromium is launched here: CI has no browsers, so the
Browser class is exercised through a fake sync_playwright
factory, and the MCP server through a fake browser object.
Real-browser proof lives in the production verification task.
"""

import json
import os
import sys
import tempfile
import unittest

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

from playwright_browser import Browser, BrowserError  # noqa: E402
from playwright_mcp import handle_request  # noqa: E402


class _FakeLocator:
    def __init__(self, text):
        self._text = text

    def inner_text(self, timeout=None):
        return self._text

    @property
    def first(self):
        return self

    def click(self, timeout=None):
        pass


class _FakeMouse:
    def __init__(self, page):
        self._page = page

    def click(self, x, y):
        self._page.calls.append(("mouse_click", x, y))

    def move(self, x, y):
        self._page.calls.append(("mouse_move", x, y))


class _FakeKeyboard:
    def __init__(self, page):
        self._page = page

    def press(self, key):
        self._page.calls.append(("key_press", key))

    def type(self, text):
        self._page.calls.append(("key_type", text))


class _FakeCDP:
    def __init__(self):
        self.sent = []
        self.handlers = {}
        self.fail_capture = False

    def on(self, event, handler):
        self.handlers[event] = handler

    def send(self, method, params=None):
        self.sent.append((method, params or {}))
        if method == "Page.captureScreenshot":
            if self.fail_capture:
                raise RuntimeError("capture boom")
            return {"data": "EAGERDATA"}
        return None


class _FakeContext:
    def __init__(self, cdp):
        self.cdp = cdp

    def new_cdp_session(self, page):
        return self.cdp


class _FakePage:
    def __init__(self):
        self.url = "https://example.com/"
        self.calls = []
        self.shot_path = None
        self.mouse = _FakeMouse(self)
        self.keyboard = _FakeKeyboard(self)
        self.cdp = _FakeCDP()
        self.context = _FakeContext(self.cdp)

    def title(self):
        return "Example Domain"

    def goto(self, url, timeout=None, wait_until=None):
        self.calls.append(("goto", url))
        self.url = url

    def locator(self, selector):
        self.calls.append(("locator", selector))
        return _FakeLocator("hello " * 3000)

    def click(self, selector, timeout=None):
        self.calls.append(("click", selector))

    def fill(self, selector, text, timeout=None):
        self.calls.append(("fill", selector, text))

    def evaluate(self, expression):
        self.calls.append(("evaluate", expression))
        if "querySelectorAll" in expression:
            return [{"tag": "button", "role": "", "name": "GO",
                     "path": "button"},
                    {"tag": "a", "role": "link", "name": "Home",
                     "path": "a"}]
        return 42

    def screenshot(self, path=None):
        self.shot_path = path
        with open(path, "wb") as fh:
            fh.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)

    def close(self):
        pass


class _FakeBrowserHandle:
    def __init__(self, page):
        self._page = page

    def new_page(self, viewport=None):
        return self._page

    def close(self):
        pass


class _FakePW:
    def __init__(self, page):
        self._page = page
        self.launch_kwargs = None

    def start(self):
        return self

    @property
    def chromium(self):
        outer = self

        class _Chromium:
            def launch(self, **kwargs):
                outer.launch_kwargs = kwargs
                return _FakeBrowserHandle(outer._page)

        return _Chromium()

    def stop(self):
        pass


class BrowserTest(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.page = _FakePage()
        self.browser = Browser(cwd=self._tmp.name,
                               proxy="http://proxy:8888",
                               browser_factory=lambda: _FakePW(self.page))

    def tearDown(self):
        self._tmp.cleanup()

    def test_navigate_uses_proxy_and_returns_title(self):
        out = self.browser.navigate("https://example.com")
        self.assertEqual(out["title"], "Example Domain")
        self.assertTrue(out["ok"])
        # Proxy is passed explicitly to launch, not left to env.
        self.assertEqual(self.browser._pw.launch_kwargs["proxy"],
                         {"server": "http://proxy:8888"})
        self.assertIn("--no-sandbox",
                      self.browser._pw.launch_kwargs["args"])

    def test_snapshot_truncates_with_flag(self):
        self.browser.navigate("https://example.com")
        out = self.browser.snapshot()
        self.assertTrue(out["truncated"])
        self.assertEqual(len(out["text"]), 8000)

    def test_click_fill_evaluate_passthrough(self):
        self.browser.navigate("https://example.com")
        self.browser.click("button.go")
        self.browser.fill("input.q", "hello")
        out = self.browser.evaluate("1 + 1")
        self.assertEqual(out["result"], "42")
        self.assertIn(("click", "button.go"), self.page.calls)
        self.assertIn(("fill", "input.q", "hello"), self.page.calls)

    def test_screenshot_validates_basename_and_writes(self):
        self.browser.navigate("https://example.com")
        out = self.browser.screenshot("shot.png")
        self.assertEqual(out["filename"], "shot.png")
        self.assertTrue(out["bytes"] > 0)
        self.assertTrue(os.path.exists(out["path"]))
        with self.assertRaises(BrowserError):
            self.browser.screenshot("../evil.png")

    def test_close_is_idempotent(self):
        self.browser.navigate("https://example.com")
        self.assertTrue(self.browser.close()["closed"])
        self.assertTrue(self.browser.close()["closed"])

    def test_click_at_and_bounds(self):
        self.browser.navigate("https://example.com")
        out = self.browser.click_at(100, 200)
        self.assertTrue(out["ok"])
        self.assertEqual(out["x"], 100)
        self.assertIn(("mouse_click", 100, 200), self.page.calls)
        with self.assertRaises(BrowserError):
            self.browser.click_at(2000, 10)
        with self.assertRaises(BrowserError):
            self.browser.click_at("left", 10)

    def test_remote_input_methods(self):
        self.browser.navigate("https://example.com")
        out = self.browser.mouse_move(10, 20)
        self.assertTrue(out["ok"])
        self.assertIn(("mouse_move", 10, 20), self.page.calls)
        out = self.browser.key_press("Enter")
        self.assertTrue(out["ok"])
        self.assertIn(("key_press", "Enter"), self.page.calls)
        out = self.browser.type_text("hi")
        self.assertTrue(out["ok"])
        self.assertIn(("key_type", "hi"), self.page.calls)
        with self.assertRaises(BrowserError):
            self.browser.mouse_move(9999, 10)

    def test_snapshot_refs_and_click_ref(self):
        self.browser.navigate("https://example.com")
        out = self.browser.snapshot_refs()
        self.assertEqual(out["count"], 2)
        self.assertEqual(out["elements"][0]["ref"], "e0")
        self.assertEqual(out["elements"][0]["name"], "GO")
        clicked = self.browser.click_ref("e0")
        self.assertTrue(clicked["ok"])
        self.assertEqual(clicked["ref"], "e0")
        self.assertIn(("locator", "button"), self.page.calls)
        with self.assertRaises(BrowserError):
            self.browser.click_ref("e99")

    def test_screencast_attach_and_pump(self):
        from playwright_remote import SimpleQueue
        sink = SimpleQueue()
        self.browser.set_frame_sink(sink)
        self.browser.navigate("https://example.com")
        self.assertIn(("Page.startScreencast",
                       {"format": "jpeg", "quality": 60,
                        "maxWidth": 1280, "maxHeight": 800,
                        "everyNthFrame": 2}),
                      [(m, p) for m, p in self.page.cdp.sent])
        handler = self.page.cdp.handlers["Page.screencastFrame"]
        handler({"sessionId": "s1", "data": "FRAMEDATA"})
        frame = self.browser.pump_screencast()
        self.assertEqual(frame, "FRAMEDATA")
        self.assertIn(("Page.screencastFrameAck", {"sessionId": "s1"}),
                      [(m, p) for m, p in self.page.cdp.sent])

    def test_navigate_pushes_eager_frame_without_ack(self):
        from playwright_remote import SimpleQueue
        sink = SimpleQueue()
        self.browser.set_frame_sink(sink)
        result = self.browser.navigate("https://example.com")
        self.assertTrue(result["ok"])
        self.assertIn(("Page.captureScreenshot",
                       {"format": "jpeg", "quality": 60}),
                      [(m, p) for m, p in self.page.cdp.sent])
        frame = self.browser.pump_screencast()
        self.assertEqual(frame, "EAGERDATA")
        acks = [(m, p) for m, p in self.page.cdp.sent
                if m == "Page.screencastFrameAck"]
        self.assertEqual(acks, [])

    def test_navigate_survives_failed_eager_capture(self):
        from playwright_remote import SimpleQueue
        sink = SimpleQueue()
        self.browser.set_frame_sink(sink)
        self.page.cdp.fail_capture = True
        try:
            result = self.browser.navigate("https://example.com")
        finally:
            self.page.cdp.fail_capture = False
        self.assertTrue(result["ok"])
        self.assertEqual(self.browser.pump_screencast(), "")

    def test_live_shot_writes_fixed_file_and_status(self):
        self.browser.navigate("https://example.com")
        out = self.browser.live_shot()
        self.assertEqual(out["image"], "browser-live.png")
        self.assertTrue(out["bytes"] > 0)
        self.assertTrue(os.path.exists(out["path"]))
        with open(os.path.join(self._tmp.name,
                               "browser-live.json")) as fh:
            status = json.load(fh)
        self.assertEqual(status["image"], "browser-live.png")
        self.assertEqual(status["url"], "https://example.com")


class FakeMCPBrowser:
    def navigate(self, url):
        return {"url": url, "title": "T", "ok": True}

    def click_at(self, x, y):
        return {"url": "https://x/", "title": "T", "ok": True,
                "x": x, "y": y}

    def live_shot(self, name=None):
        return {"image": name or "browser-live.png",
                "path": "/w/browser-live.png", "bytes": 456,
                "url": "https://x/", "title": "T", "timestamp": 1}

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

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

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

    def snapshot_refs(self):
        return {"elements": [{"ref": "e0", "tag": "button",
                              "role": "", "name": "GO"}], "count": 1}

    def click_ref(self, ref):
        return {"ok": True, "ref": ref}

    def snapshot(self):
        return {"url": "https://x/", "title": "T", "text": "hi",
                "truncated": False}

    def screenshot(self, name=None):
        return {"path": "/w/shot.png", "filename": name or "shot.png",
                "bytes": 123}


class MCPTest(unittest.TestCase):
    def test_tools_list_includes_all_browser_tools(self):
        listed = handle_request({"jsonrpc": "2.0", "id": 1,
                                 "method": "tools/list"})
        names = [t["name"] for t in listed["result"]["tools"]]
        for name in ("browser_navigate", "browser_snapshot",
                     "browser_click", "browser_fill",
                     "browser_evaluate", "browser_screenshot",
                     "browser_close", "browser_click_at",
                     "browser_live_shot"):
            self.assertIn(name, names)

    def test_tools_call_dispatches(self):
        resp = handle_request(
            {"jsonrpc": "2.0", "id": 2, "method": "tools/call",
             "params": {"name": "browser_navigate",
                        "arguments": {"url": "https://example.com"}}},
            browser=FakeMCPBrowser())
        text = resp["result"]["content"][0]["text"]
        self.assertIn("https://example.com", text)
        self.assertFalse(resp["result"]["isError"])

    def test_holder_user_blocks_agent_tools(self):
        import playwright_mcp
        fake = FakeMCPBrowser()
        remote = playwright_mcp._get_bridge(fake)
        remote.set_holder("user")
        try:
            resp = handle_request(
                {"jsonrpc": "2.0", "id": 3, "method": "tools/call",
                 "params": {"name": "browser_navigate",
                            "arguments": {"url": "https://example.com"}}},
                browser=fake)
            self.assertTrue(resp["result"]["isError"])
            self.assertIn("holds browser control",
                          resp["result"]["content"][0]["text"])
        finally:
            remote.set_holder("agent")

    def test_new_tools_listed(self):
        listed = handle_request({"jsonrpc": "2.0", "id": 4,
                                 "method": "tools/list"})
        names = [t["name"] for t in listed["result"]["tools"]]
        for name in ("browser_snapshot_refs", "browser_click_ref",
                     "browser_mouse_move", "browser_key_press",
                     "browser_type_text"):
            self.assertIn(name, names)


class LineReaderTest(unittest.TestCase):
    """Pipelined stdin lines must all arrive (no select starvation).

    Regression test for the live failure where the MCP answered
    initialize and then hung forever: the client wrote
    notifications/initialized + tools/call back-to-back, one
    readline() buffered both, and select() never saw the second
    line again because the fd was drained.
    """

    def test_pipelined_lines_all_delivered(self):
        import time
        from playwright_mcp import _LineReader
        rfd, wfd = os.pipe()
        try:
            reader = _LineReader(rfd)
            os.write(wfd, b"init\n")
            lines, drained = reader.poll()
            self.assertEqual(lines, ["init"])
            self.assertFalse(drained)
            # Second batch arrives as one write, like a real client
            # sending notification + next request back-to-back.
            os.write(wfd, b"notif\nnavigate\n")
            # Simulate the drain-loop delay: both lines sit in the
            # pipe before the next poll.
            time.sleep(0.2)
            got = []
            deadline = time.time() + 2
            while len(got) < 2 and time.time() < deadline:
                lines, drained = reader.poll()
                got.extend(lines)
                if not lines:
                    time.sleep(0.02)
            self.assertEqual(got, ["notif", "navigate"])
        finally:
            os.close(wfd)
            os.close(rfd)

    def test_partial_line_flushed_at_eof(self):
        from playwright_mcp import _LineReader
        rfd, wfd = os.pipe()
        try:
            reader = _LineReader(rfd)
            os.write(wfd, b"partial-no-newline")
            os.close(wfd)
            lines, drained = reader.poll()
            self.assertEqual(lines, [])
            self.assertFalse(drained)
            lines, drained = reader.poll()
            self.assertEqual(lines, ["partial-no-newline"])
            self.assertTrue(drained)
        finally:
            try:
                os.close(rfd)
            except OSError:
                pass


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