"""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)
"""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)
