import os
from pathlib import Path
import shutil
import socket
import sys
import threading
import time
import unittest
from urllib.parse import urljoin, urlparse


ROOT = Path(__file__).resolve().parents[1]
APP_DIR = ROOT / 'app'
TESTS_DIR = ROOT / 'tests'
sys.path.insert(0, str(APP_DIR))
sys.path.insert(0, str(TESTS_DIR))

try:
    from playwright.sync_api import sync_playwright
except ImportError:
    sync_playwright = None

import uvicorn

from admin_api import EDGE_MARKER_HEADER, OPERATOR_HEADER, SECURITY_HEADERS
import test_admin_api as admin_test_fixture
from worker_api import create_worker_app


PUBLIC_PREFIX = '/browser-fixture-prefix'


def _browser_executable():
    configured = os.environ.get('TRUF_BROWSER_EXECUTABLE')
    candidates = [
        configured,
        shutil.which('google-chrome'),
        shutil.which('chromium'),
        shutil.which('chromium-browser'),
        shutil.which('chrome'),
        r'C:\Program Files\Google\Chrome\Application\chrome.exe',
        r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe',
    ]
    return next((value for value in candidates if value and os.path.isfile(value)), None)


BROWSER_EXECUTABLE = _browser_executable()


class _PrefixAdapter:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope['type'] != 'http' or not (
            scope['path'] == PUBLIC_PREFIX
            or scope['path'].startswith(PUBLIC_PREFIX + '/')
        ):
            await self.app(scope, receive, send)
            return
        suffix = scope['path'][len(PUBLIC_PREFIX):]
        mapped = dict(scope)
        mapped['path'] = '/admin-internal' + suffix
        mapped['raw_path'] = mapped['path'].encode('ascii')
        await self.app(mapped, receive, send)


class AdminBrowserTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        if sync_playwright is None:
            raise RuntimeError(
                'Playwright is required; install tests/requirements-browser.txt'
            )
        cls.fixture = admin_test_fixture.AdminAPITests(methodName='runTest')
        cls.fixture.setUp()
        app = create_worker_app(
            cls.fixture.worker, reaper_interval_seconds=3600,
            admin_service=cls.fixture.admin,
        )
        cls.app = _PrefixAdapter(app)
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
            probe.bind(('127.0.0.1', 0))
            cls.port = probe.getsockname()[1]
        config = uvicorn.Config(
            cls.app, host='127.0.0.1', port=cls.port,
            log_level='critical', lifespan='off', access_log=False,
        )
        cls.server = uvicorn.Server(config)
        cls.thread = threading.Thread(target=cls.server.run, daemon=True)
        cls.thread.start()
        deadline = time.monotonic() + 10
        while not cls.server.started and cls.thread.is_alive():
            if time.monotonic() >= deadline:
                raise RuntimeError('browser fixture server did not start')
            time.sleep(0.02)
        if not cls.server.started:
            raise RuntimeError('browser fixture server failed')
        cls.base_url = f'http://127.0.0.1:{cls.port}{PUBLIC_PREFIX}'

    @classmethod
    def tearDownClass(cls):
        cls.server.should_exit = True
        cls.thread.join(timeout=10)
        cls.fixture.tearDown()

    def _assert_storage_empty(self, page):
        state = page.evaluate('''async () => ({
            local: Object.keys(localStorage),
            session: Object.keys(sessionStorage),
            caches: 'caches' in self ? await caches.keys() : [],
            serviceWorkers: 'serviceWorker' in navigator
                ? (await navigator.serviceWorker.getRegistrations()).length : 0,
            indexedDb: indexedDB.databases
                ? (await indexedDB.databases()).map(item => item.name || '') : [],
        })''')
        self.assertEqual(state, {
            'local': [], 'session': [], 'caches': [],
            'serviceWorkers': 0, 'indexedDb': [],
        })

    def _assert_page_contract(self, page, response, *, script_count=0):
        self.assertIsNotNone(response)
        self.assertEqual(response.status, 200)
        headers = response.all_headers()
        for name, expected in SECURITY_HEADERS.items():
            self.assertEqual(headers.get(name.lower()), expected)
        self.assertEqual(page.locator('script').count(), script_count)
        if script_count:
            self.assertEqual(
                page.locator('script:not([src])').count(), 0,
            )
        event_attributes = page.locator('*').evaluate_all('''elements =>
            elements.flatMap(element => Array.from(element.attributes))
                .filter(attribute => attribute.name.toLowerCase().startsWith('on'))
                .map(attribute => attribute.name)
        ''')
        self.assertEqual(event_attributes, [])
        scroll_width = page.evaluate('document.documentElement.scrollWidth')
        client_width = page.evaluate('document.documentElement.clientWidth')
        overflowing = page.locator('*').evaluate_all('''elements => elements
            .filter(element => element.getBoundingClientRect().right >
                document.documentElement.clientWidth + 1)
            .slice(0, 10)
            .map(element => ({
                tag: element.tagName,
                className: element.className,
                right: element.getBoundingClientRect().right,
                scrollWidth: element.scrollWidth,
                display: getComputedStyle(element).display,
                width: getComputedStyle(element).width,
                parent: element.parentElement && {
                    tag: element.parentElement.tagName,
                    className: element.parentElement.className,
                    width: getComputedStyle(element.parentElement).width,
                    overflow: getComputedStyle(element.parentElement).overflowX,
                },
            }))''')
        self.assertLessEqual(scroll_width, client_width, overflowing)
        for target in page.locator('a[href], link[href], form[action]').evaluate_all(
            '''elements => elements.map(element =>
                new URL(element.getAttribute('href') || element.getAttribute('action'),
                        document.baseURI).href)'''
        ):
            parsed = urlparse(target)
            self.assertEqual(parsed.netloc, f'127.0.0.1:{self.port}')
            self.assertTrue(parsed.path.startswith(PUBLIC_PREFIX))

    def test_navigation_csp_mobile_and_secret_non_retention(self):
        console_errors = []
        with sync_playwright() as playwright:
            launch_options = {'headless': True, 'args': ['--disable-gpu']}
            if BROWSER_EXECUTABLE is not None:
                launch_options['executable_path'] = BROWSER_EXECUTABLE
            browser = playwright.chromium.launch(**launch_options)
            context = browser.new_context(
                viewport={'width': 1440, 'height': 900},
                extra_http_headers={
                    EDGE_MARKER_HEADER: admin_test_fixture.MARKER,
                    OPERATOR_HEADER: admin_test_fixture.OPERATOR,
                },
            )
            context.grant_permissions(
                ['clipboard-read', 'clipboard-write'],
                origin=f'http://127.0.0.1:{self.port}',
            )
            page = context.new_page()
            page.on(
                'console',
                lambda message: console_errors.append(message.text)
                if message.type == 'error' else None,
            )

            response = page.goto(self.base_url + '/secrets', wait_until='load')
            self._assert_page_contract(page, response)
            textarea = page.locator('textarea[name="document_text"]')
            initial_secret = textarea.input_value()
            self.assertIn('editor-secret', initial_secret)
            self.assertEqual(page.content().count('editor-secret'), 1)
            self.assertNotIn(
                'editor-secret',
                page.locator('body').evaluate('''body => {
                    const clone = body.cloneNode(true);
                    clone.querySelectorAll('textarea').forEach(item => item.remove());
                    return clone.textContent;
                }'''),
            )
            self.assertEqual(textarea.get_attribute('autocomplete'), 'off')
            self.assertEqual(textarea.locator('xpath=ancestor::form').get_attribute('autocomplete'), 'off')
            self._assert_storage_empty(page)

            unsaved = 'browser-unsaved-secret-must-not-persist'
            textarea.fill(initial_secret + '\n' + unsaved)
            page.goto(self.base_url + '/overview', wait_until='load')
            page.go_back(wait_until='load')
            page.reload(wait_until='load')
            self.assertNotIn(unsaved, textarea.input_value())
            self.assertNotIn(unsaved, page.content())
            self._assert_storage_empty(page)

            response = page.goto(self.base_url + '/', wait_until='load')
            self._assert_page_contract(page, response)
            response = page.goto(
                self.base_url + '/assignments/5', wait_until='load',
            )
            self._assert_page_contract(page, response, script_count=1)
            copy_button = page.locator('[data-copy-target]').first
            expected_json = page.locator('.canonical-json').first.input_value()
            copy_button.click()
            deadline = time.monotonic() + 5
            while (
                copy_button.text_content() != 'Copied canonical JSON'
                and time.monotonic() < deadline
            ):
                page.wait_for_timeout(20)
            self.assertEqual(copy_button.text_content(), 'Copied canonical JSON')
            self.assertEqual(
                page.evaluate('navigator.clipboard.readText()'), expected_json,
            )
            self._assert_storage_empty(page)

            response = page.goto(self.base_url + '/', wait_until='load')
            self._assert_page_contract(page, response)
            navigation = page.locator('nav a').evaluate_all(
                'links => links.map(link => new URL(link.href).href)'
            )
            self.assertGreaterEqual(len(navigation), 8)
            for target in navigation:
                self.assertTrue(urlparse(target).path.startswith(PUBLIC_PREFIX))
                response = page.goto(target, wait_until='load')
                self._assert_page_contract(page, response)

            self.assertEqual(console_errors, [])
            page.goto(self.base_url + '/overview', wait_until='load')
            page.evaluate('''() => {
                const probe = document.createElement('script');
                probe.textContent = 'window.__trufInlineCspProbe = true';
                document.body.appendChild(probe);
            }''')
            page.wait_for_timeout(100)
            self.assertFalse(page.evaluate('Boolean(window.__trufInlineCspProbe)'))

            page.set_viewport_size({'width': 390, 'height': 844})
            for path in ('/supervisor', '/files', '/audit', '/secrets'):
                response = page.goto(self.base_url + path, wait_until='load')
                self._assert_page_contract(page, response)
            response = page.goto(
                self.base_url + '/assignments/5', wait_until='load',
            )
            self._assert_page_contract(page, response, script_count=1)

            self._assert_storage_empty(page)
            context.close()
            browser.close()


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