"""Compare verified images using real guest e1000 HTTP, DOM, CSS and presented frames.

The loopback HTTP server serves a declared test fixture. It never fetches Bing,
proxies the Internet or substitutes a screenshot for guest rendering.
"""
import argparse
import hashlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
from pathlib import Path
import re
import shutil
import socket
import statistics
import subprocess
import threading
import time

from PIL import Image
from TestBoot import Qmp
from IndependentBingEvidence import network_arguments, json_blocks, PRESENTED, window_geometry

ROOT = Path(__file__).resolve().parents[2]


def digest(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def fixture():
    sheets = {}
    for sheet in range(2):
        sheets[f'/sheet{sheet}.css'] = ''.join(
            f'.parent p.item{i}{{color:#{i % 0xffffff:06x}}}'
            for i in range(sheet * 800, (sheet + 1) * 800)).encode()
    html = ('<!doctype html><html><head><title>XFE native loading fixture</title>'
            '<style>body{margin:0;background:#ffffff}p{margin:0;font-size:14px;line-height:20px}</style>'
            '<link rel=stylesheet href=/sheet0.css><link rel=stylesheet href=/sheet1.css>'
            '</head><body><main class=parent>' + ''.join(
                f'<p class=item{i * 15}>Native row {i}</p>' for i in range(80)) +
            '</main></body></html>').encode()
    return html, sheets


def run(receipt_path, label, output, require_reuse):
    receipt = json.loads(receipt_path.read_text(encoding='utf-8-sig'))
    assert receipt['Passed'] and receipt['ContainsCosmos'] is False
    image = Path(receipt['ImagePath'])
    assert digest(image) == receipt['ImageSha256']
    result = dict(Kind='XfeNativeBrowserLoadProfile', Passed=False, Label=label,
                  ImageSha256=digest(image), VerifiedReceipt=str(receipt_path),
                  VerifiedReceiptSha256=digest(receipt_path), VirtualCpus=8, ExecutingCpus=1,
                  HostProxy=False, HostDownloadedPage=False, HostScreenshotSubstitution=False,
                  HarnessSha256=digest(__file__))
    output.mkdir(parents=True)
    html, sheets = fixture()
    served = []

    class Handler(BaseHTTPRequestHandler):
        protocol_version = 'HTTP/1.1'

        def do_GET(self):
            path = self.path.split('?', 1)[0]
            body = html if path == '/load' else sheets.get(path)
            if body is None:
                self.send_error(404)
                return
            served.append(dict(Path=self.path, Client=list(self.client_address),
                               BodySha256=hashlib.sha256(body).hexdigest(), Bytes=len(body)))
            self.send_response(200)
            self.send_header('Content-Type', 'text/html; charset=utf-8' if path == '/load' else 'text/css')
            self.send_header('Content-Length', str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            self.wfile.flush()

        def log_message(self, *_):
            pass

    server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    qemu = qmp = None
    serial_path = output / 'Serial.log'

    def text():
        return serial_path.read_text(encoding='utf-8', errors='replace') if serial_path.exists() else ''

    def wait(predicate, seconds=180):
        end = time.monotonic() + seconds
        while time.monotonic() < end:
            assert qemu.poll() is None, 'Guest exited'
            log = text()
            assert not any(marker in log for marker in ('XFE_NATIVE_FAIL', 'XFE_RUNTIME_FATAL',
                'XFE_ZERO_FAILFAST', 'XFE_INDEPENDENT_APPLICATION_FAILED', 'XFE_EDITOR_BACKGROUND|SessionError|'))
            if predicate(log):
                return log
            time.sleep(.025)
        raise AssertionError('Guest loading profile timed out: ' + text()[-4096:])

    def key(chord):
        qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
        time.sleep(.12)

    def type_text(value):
        codes = {':': 'shift-semicolon', '.': 'dot', '/': 'slash', '-': 'minus',
                 '?': 'shift-slash', '=': 'equal'}
        for char in value:
            key(codes.get(char, char))

    try:
        shutil.copy2(ROOT / '.tools/firmware/edk2-i386-vars.fd', output / 'FirmwareVars.fd')
        with socket.socket() as reservation:
            reservation.bind(('127.0.0.1', 0))
            port = reservation.getsockname()[1]
        command = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35', '-accel', 'tcg',
            '-cpu', 'max', '-smp', '8', '-m', '1024M', '-vga', 'std', '-global', 'VGA.vgamem_mb=64',
            '-display', 'none', *network_arguments(True), '-rtc', 'base=utc',
            '-drive', f'if=pflash,format=raw,readonly=on,file={ROOT / ".tools/firmware/edk2-x86_64-code.fd"}',
            '-drive', f'if=pflash,format=raw,file={output / "FirmwareVars.fd"}',
            '-drive', f'file={image},format=raw,if=ide,snapshot=on', '-serial', f'file:{serial_path}',
            '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown']
        result['QemuArguments'] = command
        with (output / 'Qemu.log').open('w') as log:
            qemu = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
                                    creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
            end = time.monotonic() + 30
            while qmp is None:
                try:
                    qmp = Qmp(port)
                except OSError:
                    assert time.monotonic() < end
                    time.sleep(.1)
            wait(lambda s: 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME' in s)
            if 'XFE_DESKTOP_ACTIVE|Terminal' in text():
                key('alt-f4')
            key('ctrl-esc'); key('down'); key('down'); key('ret')
            wait(lambda s: 'XFE_DESKTOP_ACTIVE|Browser' in s)
            browser_id, _ = window_geometry(text(), 'Browser')
            # Product defaults remain intact. Time only the subsequent declared
            # local-network fixture, once the genuine default navigation retires.
            wait(lambda s: f'XFE_EDITOR_BACKGROUND|Ready|Window|{browser_id}|' in s)
            rows = []
            for iteration in range(1, 4):
                url = f'http://10.0.2.2:{server.server_port}/load?run={iteration}'
                key('ctrl-l'); type_text(url)
                start = len(text()); served_start = len(served)
                begun = time.monotonic_ns()
                qmp.call('human-monitor-command', {'command-line': 'sendkey ret'})

                def presented(log):
                    return any(m[3] == url for m in PRESENTED.finditer(log, start))

                log = wait(presented)
                elapsed = (time.monotonic_ns() - begun) / 1_000_000
                time.sleep(max(0, .12 - elapsed / 1000))  # Allow the test key to release outside the timing.
                proofs = [entry['Value'] for entry in json_blocks(log, 'XFE_REMOTE_PAGE_PROOF', start) if entry['Value']['Address'] == url]
                assert proofs and proofs[-1]['BodySha256'] == hashlib.sha256(html).hexdigest()
                proof = proofs[-1]
                assert len(proof['LoadedStyles']) == 2
                assert 'Nativerow0' in ''.join(run['Text'] for run in proof['TextRuns']).replace(' ', '')
                assert proof['Diagnostics'] == ''
                requests = served[served_start:]
                assert len(requests) == 3 and {r['Path'].split('?', 1)[0] for r in requests} == {'/load', '/sheet0.css', '/sheet1.css'}
                connections = len({tuple(r['Client']) for r in requests})
                if require_reuse:
                    assert connections == 1, 'Guest did not reuse the actual HTTP wire'
                qmp.call('screendump', {'filename': str(output / f'Frame{iteration}.ppm')})
                with Image.open(output / f'Frame{iteration}.ppm') as frame:
                    rgb = frame.convert('RGB')
                    rgb.save(output / f'Frame{iteration}.png')
                    _, (x, y, width, height) = window_geometry(log, 'Browser')
                    assert proof['ViewportWidth'] == width - 4 and proof['ViewportHeight'] == height - 96
                    web = rgb.crop((x + 2, y + 67, x + width - 2, y + height - 29))
                    web.save(output / f'WebViewport{iteration}.png')
                    pixels = hashlib.sha256(web.tobytes()).hexdigest()
                rows.append(dict(Iteration=iteration, Address=url, EnterToPresentedHostMilliseconds=elapsed,
                    Requests=requests, TcpConnections=connections, DomReadback=proof,
                    WebViewportPixelsSha256=pixels,
                    ScreenshotSha256=digest(output / f'Frame{iteration}.png')))
            result.update(Passed=True, Cases=rows,
                MedianEnterToPresentedHostMilliseconds=statistics.median(r['EnterToPresentedHostMilliseconds'] for r in rows),
                DistinctFixtureTcpConnections=len({tuple(r['Client']) for r in served}),
                FixtureHtmlSha256=hashlib.sha256(html).hexdigest(),
                FixtureStylesSha256={k: hashlib.sha256(v).hexdigest() for k, v in sheets.items()})
    except Exception as error:
        result['Error'] = str(error)
    finally:
        if qemu is not None:
            if qemu.poll() is None and qmp is not None:
                try:
                    qmp.call('quit')
                except (OSError, RuntimeError):
                    pass
            try:
                qemu.wait(timeout=10)
            except subprocess.TimeoutExpired:
                qemu.terminate(); qemu.wait(timeout=10); result['ForcedCleanup'] = True
            result['QemuExitCode'] = qemu.returncode
        if qmp is not None:
            qmp.close()
        server.shutdown(); server.server_close()
        result['ImageUnchanged'] = digest(image) == receipt['ImageSha256']
        result['Passed'] &= result.get('QemuExitCode') == 0 and result['ImageUnchanged']
        (output / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(json.dumps({k: v for k, v in result.items() if k not in ('Cases', 'QemuArguments')}, ensure_ascii=True))
    return result


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--baseline-proof', type=Path, required=True)
    parser.add_argument('--optimized-proof', type=Path, required=True)
    args = parser.parse_args()
    output = ROOT / 'build/BrowserLoadSpeed' / f'NativeComparison-{time.time_ns()}'
    before = run(args.baseline_proof.resolve(), 'dev.3 baseline', output / 'Before', False)
    after = run(args.optimized_proof.resolve(), 'optimized', output / 'After', True)
    passed = before['Passed'] and after['Passed']
    comparison = dict(Kind='XfeNativeBrowserLoadComparison', Passed=passed, FullReleaseQualified=False,
        Cases='Same fixture bytes, QEMU TCG max, 8 virtual CPUs / 1 executing BSP; sequential runs',
        Before=str(output / 'Before/Result.json'), After=str(output / 'After/Result.json'))
    if passed:
        assert before['FixtureHtmlSha256'] == after['FixtureHtmlSha256']
        assert before['FixtureStylesSha256'] == after['FixtureStylesSha256']
        # Layout/text/color must agree, in addition to received fixture hashes.
        assert [r['DomReadback']['TextRuns'] for r in before['Cases']] == [r['DomReadback']['TextRuns'] for r in after['Cases']]
        assert [r['WebViewportPixelsSha256'] for r in before['Cases']] == [r['WebViewportPixelsSha256'] for r in after['Cases']]
        comparison.update(BeforeMilliseconds=before['MedianEnterToPresentedHostMilliseconds'],
                          AfterMilliseconds=after['MedianEnterToPresentedHostMilliseconds'],
                          Speedup=before['MedianEnterToPresentedHostMilliseconds'] / after['MedianEnterToPresentedHostMilliseconds'],
                          RenderedTextGeometryAndColorsEqual=True, AllWebViewportPixelsEqual=True)
    (output / 'Comparison.json').write_text(json.dumps(comparison, indent=2), encoding='utf-8')
    print(json.dumps(comparison))
    raise SystemExit(0 if passed else 1)