"""Browser discovery through the real kernel, without starting a GUI app."""

import unittest
import http.server
import json
import os
import signal
import threading
import shutil
import time
from pathlib import Path

from harness import Albedo, Provider, exclusive, python, text


@exclusive
class BrowserTests(unittest.TestCase):
    def test_headless_shell_preserves_overrides_headed_launches_and_fallback(self):
        code = ""

        def script(request):
            if request["messages"][-1].get("role") == "user":
                return python(code)
            return text("done")

        def prepare(app):
            nonlocal code
            bin_dir = app.workspace / "bin"
            bin_dir.mkdir()
            chrome = bin_dir / "chromium"
            shell = bin_dir / "headless-shell"
            shell_link = app.home / "browsers/chrome-headless-shell"
            shell_link.parent.mkdir()
            shell_link.symlink_to(shell)
            for executable in (chrome, shell):
                executable.write_text(
                    '#!/bin/sh\necho "test browser exited" >&2\nexit 23\n'
                )
                executable.chmod(0o755)
            for variable in (
                "ALBEDO_BROWSER_EXECUTABLE",
                "PRIME_BROWSER_EXECUTABLE",
                "CHROME_PATH",
            ):
                app.daemon.env[variable] = ""

            code = f"""
from pathlib import Path
from albedo_plugins.browser import LaunchError
chrome = {str(chrome)!r}
shell = {str(shell)!r}
os.environ['PATH'] = {str(bin_dir)!r}
async def check(expected, **options):
    try:
        b = await browser.spawn(sandbox=False, timeout=3, **options)
    except LaunchError as error:
        assert error.details['executable'] == expected, error.to_dict()
        assert error.details['reason'] == 'process_exited', error.to_dict()
        assert 'test browser exited' in error.details['stderr_tail'], error.to_dict()
        assert not Path(error.details['profile']).exists()
    else:
        await b.close()
        raise AssertionError('test browser should exit during startup')
await check(shell)
path_shell = Path({str(bin_dir / "chrome-headless-shell")!r})
path_shell.symlink_to(chrome)
await check(chrome)
path_shell.unlink()
await check(chrome, headless=False)
await check(chrome, executable=chrome)
os.environ['ALBEDO_BROWSER_EXECUTABLE'] = chrome
await check(chrome)
del os.environ['ALBEDO_BROWSER_EXECUTABLE']
Path({str(shell_link)!r}).unlink()
await check(chrome)
try:
    await browser.spawn(executable={str(shell_link)!r}, sandbox=False)
except LaunchError as error:
    assert error.details['reason'] == 'executable_not_found', error.to_dict()
else:
    raise AssertionError('missing explicit browser must not fall back')
Path({str(app.workspace / "passed")!r}).write_text('passed')
"""

        provider = Provider(script)
        self.addCleanup(provider.close)
        with Albedo(provider, prepare=prepare) as app:
            session = app.session()
            app.prompt(session, "try the installed browsers").close()
            app.idle(session)
            self.assertTrue((app.workspace / "passed").exists(), app.history(session))


def require_chrome(test):
    if not (
        shutil.which("chromium")
        or shutil.which("google-chrome")
        or shutil.which("chrome-headless-shell")
        or Path("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome").exists()
        or Path("/Applications/Chromium.app/Contents/MacOS/Chromium").exists()
    ):
        test.skipTest("No Chrome or Chromium installed on this system")


class BrowserLifetimeTests(unittest.TestCase):
    def test_a_kernel_killed_outright_takes_its_browser_with_it(self):
        require_chrome(self)
        found = None

        def reply(request):
            if request["messages"][-1].get("role") == "user":
                return python(
                    f"""
import json
from albedo_plugins.browser import launch
b = await browser.spawn()
owner = next(iter(launch._LIVE))
Path({str(found)!r}).write_text(json.dumps(
    {{'kernel': os.getpid(), 'chrome': owner.process.pid, 'profile': str(owner.profile)}}
))
"""
                )
            return text("done")

        provider = Provider(reply)
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            found = app.workspace / "browser.json"
            session = app.session()
            app.prompt(session, "start a browser").close()
            app.idle(session)
            spawned = json.loads(found.read_text())
            # SIGKILL skips the kernel's own cleanup; only the watchdog is left.
            os.kill(spawned["kernel"], signal.SIGKILL)
            deadline = time.monotonic() + 15
            # The watchdog removes the profile once Chrome's group has ended.
            while time.monotonic() < deadline and (
                running(spawned["chrome"]) or Path(spawned["profile"]).exists()
            ):
                time.sleep(0.1)
            self.assertFalse(running(spawned["chrome"]), "Chrome outlived its kernel")
            self.assertFalse(Path(spawned["profile"]).exists())


def running(pid):
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    return True


class BrowserGestureTests(unittest.TestCase):
    def test_gestures_and_statement_bodies_in_a_real_browser(self):
        require_chrome(self)

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_GET(self):
                body = b"<!doctype html><title>CSP test</title>"
                self.send_response(200)
                self.send_header("Content-Type", "text/html")
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Content-Security-Policy", "script-src 'self'")
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, format, *args):
                pass

        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        self.addCleanup(server.server_close)
        self.addCleanup(thread.join)
        self.addCleanup(server.shutdown)
        code = """
from albedo_plugins.browser import SelectionError, JavaScriptError
b = await browser.spawn()
try:
    page = await b.new_page()
    await page.goto('data:text/html,<button style="position:absolute;left:10px;top:10px;width:90px;height:40px" onclick="window.clicks=(window.clicks||0)+1">Drag handle</button><a href="#">Other link</a><button>Other button</button>')
    assert await page.evaluate('1 + 2') == 3
    assert await page.evaluate('() => 4') == 4
    assert await page.evaluate('window.runs = 0; return 5') == 5
    try:
        await page.evaluate("window.runs++; throw new SyntaxError('runtime')")
    except JavaScriptError:
        pass
    else:
        raise AssertionError('runtime error was swallowed')
    assert await page.evaluate('window.runs') == 1
    try:
        await page.evaluate("(() => { window.runs++; throw new SyntaxError('runtime'); })()")
    except JavaScriptError:
        pass
    else:
        raise AssertionError('expression runtime error was swallowed')
    assert await page.evaluate('window.runs') == 2
    await page.click((30, 30))
    await page.click_text('Drag handle')
    assert await page.evaluate('window.clicks') == 2
    for label in ('missing', 'Other'):
        try:
            await page.click_text(label)
        except SelectionError as error:
            assert ('found 0' if label == 'missing' else 'found 2') in str(error)
        else:
            raise AssertionError('ambiguous or missing text was clicked')
    await page.evaluate('''
        window.moves = []; window.releases = 0;
        document.addEventListener('pointermove', e => {if(e.buttons === 1) window.moves.push([e.clientX,e.clientY]);});
        document.addEventListener('pointerup', () => window.releases++);
        return true;
    ''')
    snapshot = await page.observe()
    node = browser.one(snapshot, role='button', name='Drag handle')
    await page.drag(node, (200, 150), steps=5)
    assert await page.evaluate('window.moves.length') == 5
    assert await page.evaluate('window.moves.at(-1)') == [200, 150]
    assert await page.evaluate('window.releases') == 1
    await page.drag((200, 150), node, steps=3)
    assert await page.evaluate('window.moves.length') == 8
    assert await page.evaluate('window.releases') == 2
    print('document drag: 8 held-button moves, 2 releases; clicks: 2')
    await page.goto(CSP_URL)
    try:
        await page.evaluate("new Function('return 1')()")
    except JavaScriptError as error:
        assert 'unsafe-eval' in str(error.details), error
    else:
        raise AssertionError('CSP must block Function compilation')
    assert await page.evaluate('1 + 2') == 3
    assert await page.evaluate('localStorage.clear(); return 5') == 5
    print('CSP expression and statement body passed')
finally:
    await b.close()
"""

        def reply(request):
            if request["messages"][-1].get("role") == "user":
                return python(
                    code.replace(
                        "CSP_URL", repr(f"http://127.0.0.1:{server.server_port}/")
                    )
                )
            return text("done")

        provider = Provider(reply)
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            session = app.session()
            app.prompt(session, "exercise browser gestures").close()
            app.idle(session)
            history = str(app.history(session))
            self.assertIn("CSP expression and statement body passed", history)
            self.assertIn(
                "document drag: 8 held-button moves, 2 releases; clicks: 2", history
            )