XFESystem

"""Deliberate native allocation failure in a disposable QEMU, through real HTTP/JS."""
import re
import time
from TestBrowserPipeline import BrowserFixture


def run_browser_oom_check(qmp, text, wait_log, scale, pointer):
    peer = BrowserFixture('NativeOomNegative')
    peer.resources['/Index'] = ('text/html', '<!doctype html><title>Controlled OOM</title><script>new ArrayBuffer(1073741824);</script>')
    start = len(text())

    def key(code, down):
        qmp('input-send-event', {'events': [{'type': 'key', 'data': {'down': down, 'key': {'type': 'qcode', 'data': code}}}]})
        time.sleep(.03)

    def tap(code):
        key(code, True); key(code, False)

    try:
        target = [round(60 * scale), round(308 * scale)]
        while pointer != target:
            dx = max(-80, min(80, target[0] - pointer[0])); dy = max(-80, min(80, target[1] - pointer[1]))
            qmp('input-send-event', {'events': [{'type': 'rel', 'data': {'axis': 'x', 'value': dx}}, {'type': 'rel', 'data': {'axis': 'y', 'value': dy}}]})
            pointer[0] += dx; pointer[1] += dy; time.sleep(.025)
        for down in (True, False):
            qmp('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'left'}}]}); time.sleep(.1)
        wait_log('XFE_DESKTOP_ACTIVE|Browser', start)
        wait_log('XFE_NETWORK_DHCP_OK|10.0.2.15|DnsServers|1', timeout=20)
        key('ctrl', True); tap('l'); key('ctrl', False)
        for character in peer.address:
            if character == ':': key('shift', True); tap('semicolon'); key('shift', False)
            elif character.isupper(): key('shift', True); tap(character.lower()); key('shift', False)
            else: tap({'.': 'dot', '/': 'slash'}.get(character, character.lower()))
        tap('ret')
        deadline = time.monotonic() + 30
        while time.monotonic() < deadline:
            log = text()[start:]
            if 'XFE_RUNTIME_OOM|' in log and '[FAILFAST] Managed allocation failed; guest halted' in log:
                assert 'CPU EXCEPTION:' not in log and '*** UNHANDLED EXCEPTION ***' not in log, log[-4000:]
                assert any(request['Path'] == '/Index' for request in peer.requests), 'No native HTTP request'
                registers = qmp('human-monitor-command', {'command-line': 'info registers'})
                flags = re.search(r'\bRFL=([0-9A-Fa-f]+)', registers)
                assert flags and int(flags[1], 16) & 0x200 == 0, ('Fatal guest path did not keep interrupts disabled', registers)
                stopped_log = text()
                time.sleep(.2)
                assert text() == stopped_log, 'Other guest threads still produced output after the fatal marker'
                return {'ExpectedAllocationFailureObserved': True, 'ArbitraryPageFault': False,
                        'ApplicationOrGuiRecovered': False, 'HostBrowserUsed': False,
                        'GuestInterruptsDisabled': True, 'GuestRflags': int(flags[1], 16),
                        'SerialQuiescentAfterFatalSeconds': .2, 'Registers': registers,
                        'RequestedArrayBufferBytes': 1073741824, 'HttpRequests': peer.requests.copy(),
                        'Marker': next(line for line in log.splitlines() if line.startswith('XFE_RUNTIME_OOM|'))}
            if 'CPU EXCEPTION:' in log: raise AssertionError(log[-4000:])
            time.sleep(.05)
        raise AssertionError('Native OOM guard was not exercised: ' + text()[start:][-4000:])
    finally:
        peer.close()