XFESystem

"""Actual independent VGA scanouts, UI input, topology changes and USB cold-boot persistence."""
import argparse
import json
from pathlib import Path
import re
import shutil
import socket
import subprocess
import time

from PIL import Image
from TestBoot import ROOT, Qmp, digest, read_json
from IndependentDocumentEvidence import window_geometry


def verify_displays(image):
    image = Path(image).resolve()
    manifest = read_json(str(image) + '.json')
    assert manifest['Edition'] == 'Desktop' and manifest['UsbWorkspace']
    proof = ROOT / 'build/IndependentKernel' / ('MultiDisplay-' + str(time.time_ns()))
    proof.mkdir()
    disk = proof / 'PrivateUsb.img'
    shutil.copyfile(image, disk)
    sources = dict(manifest['IndependentKernelBuild']['SourceSha256'])
    sources['Tools/IndependentKernel/TestMultiDisplay.py'] = digest(__file__)
    result = {'Kind': 'XfeIndependentMultiDisplayProof', 'Version': 1, 'Passed': False,
              'ImagePath': str(image), 'ImageSha256': digest(image), 'ManifestSha256': digest(str(image) + '.json'),
              'KernelSha256': manifest['KernelSha256'], 'SourceHashes': sources,
              'Input': 'Actual QMP keyboard and relative mouse packets only', 'HostWorkspaceInjection': False,
              'OutputBackend': 'Two independent PCI VGA devices and real GOP/Limine framebuffers',
              'PhysicalHardwareAcceptancePassed': False, 'HotplugAcceptancePassed': False,
              'Boots': [], 'EvidenceFiles': []}

    def ref(path):
        return {'Path': path.relative_to(proof).as_posix(), 'Sha256': digest(path)}

    try:
        for boot_number in (1, 2, 3):
            folder = proof / ('Boot' + str(boot_number)); folder.mkdir()
            shutil.copyfile(ROOT / '.tools/firmware/edk2-i386-vars.fd', folder / 'Vars.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', 'none',
                       '-device', 'VGA,id=Display1,vgamem_mb=64,edid=on,xres=1280,yres=800,xmax=1280,ymax=800',
                       '-display', 'none', '-net', 'none', '-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={folder / "Vars.fd"}',
                       '-device', 'qemu-xhci,id=Usb', '-drive', f'if=none,id=BootUsb,format=raw,file={disk}',
                       '-device', 'usb-storage,drive=BootUsb,bus=Usb.0,bootindex=1',
                       '-serial', f'file:{folder / "Serial.log"}', '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off',
                       '-no-reboot', '-no-shutdown']
            if boot_number != 3:
                command += ['-device', 'secondary-vga,id=Display2,vgamem_mb=64,edid=on,xres=1280,yres=800,xmax=1280,ymax=800']
            boot = {'Number': boot_number, 'QemuArguments': command, 'InputEvents': [], 'Captures': []}
            result['Boots'].append(boot)
            qmp = qemu = None

            def text():
                assert qemu.poll() is None, 'Multi-display guest exited unexpectedly'
                value = (folder / 'Serial.log').read_text(encoding='utf-8', errors='replace')
                assert not any(marker in value for marker in ('XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL',
                    'XFE_ZERO_FAILFAST', 'XFE_INDEPENDENT_APPLICATION_FAILED', 'Rendering failed:')), value[-3000:]
                return value

            def wait(marker, after=0, seconds=60):
                end = time.monotonic() + seconds
                while time.monotonic() < end:
                    if marker in text()[after:]:
                        return
                    time.sleep(.1)
                raise AssertionError('Display acknowledgement timeout: ' + marker + '\n' + text()[-3000:])

            def state():
                records = re.findall(r'^XFE_DESKTOP_DISPLAYS\|([^\r\n]+)\r?\n', text(), re.M)
                assert records, 'No actual presented display state'
                fields = records[-1].split('|')
                return dict(zip(fields[::2], fields[1::2]))

            def wait_state(**required):
                end = time.monotonic() + 35
                while time.monotonic() < end:
                    # FIRST_FRAME precedes the asynchronous diagnostics worker.
                    # Missing/partial records are pending until the same deadline;
                    # they must never be substituted for a presented state.
                    records = re.findall(r'^XFE_DESKTOP_DISPLAYS\|([^\r\n]+)\r?\n', text(), re.M)
                    if records:
                        fields = records[-1].split('|')
                        actual = dict(zip(fields[::2], fields[1::2])) if len(fields) % 2 == 0 else {}
                        if all(actual.get(key) == str(value) for key, value in required.items()): return
                    time.sleep(.1)
                raise AssertionError('Unexpected actual display state: ' + repr(state()) + ' expected ' + repr(required))

            def key(chord):
                boot['InputEvents'].append({'Key': chord, 'SerialOffset': len(text())})
                qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord + ' 30'})
                time.sleep(.17)

            pointer = [640, 400]

            def move(x, y):
                boot['InputEvents'].append({'PointerTarget': [x, y]})
                while pointer != [x, y]:
                    dx, dy = max(-80, min(80, x-pointer[0])), max(-80, min(80, y-pointer[1]))
                    qmp.call('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(.035)

            def button(down):
                qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'left'}}]})
                time.sleep(.14)

            def click(x, y):
                move(x, y); button(True); button(False)

            def captures(name, expected='different'):
                time.sleep(.4)
                before = state()
                images = []
                for index in range(1, 2 if boot_number == 3 else 3):
                    ppm = folder / (name + '-' + str(index) + '.ppm')
                    qmp.call('screendump', {'filename': str(ppm), 'device': 'Display' + str(index), 'head': 0})
                    with Image.open(ppm) as source:
                        frame = source.convert('RGB'); assert frame.size == (1280, 800), frame.size
                        path = ppm.with_suffix('.png'); frame.save(path); images.append(frame.copy())
                    record = ref(path); result['EvidenceFiles'].append(record)
                    boot['Captures'].append({'Name': name, 'Output': index, **record, 'State': before})
                assert state() == before, 'Display state changed while capturing independent outputs'
                if expected == 'same':
                    assert images[0].tobytes() == images[1].tobytes(), 'Duplicated scanouts have different actual pixels'
                if expected == 'different':
                    assert images[0].tobytes() != images[1].tobytes(), 'Independent scanouts unexpectedly contain identical pixels'
                if expected == 'first-black':
                    assert images[0].getextrema() == ((0, 0), (0, 0), (0, 0)), 'Inactive display is not black'
                    assert images[1].getextrema() != ((0, 0), (0, 0), (0, 0)), 'Selected display is blank'
                return images

            try:
                with (folder / 'Qemu.log').open('w', encoding='utf-8') as log:
                    qemu = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
                        creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
                end = time.monotonic() + 20
                while qmp is None:
                    assert qemu.poll() is None, (folder / 'Qemu.log').read_text()
                    try:
                        qmp = Qmp(port)
                    except OSError:
                        assert time.monotonic() < end
                        time.sleep(.1)
                wait('XFE_INDEPENDENT_DESKTOP_FIRST_FRAME', seconds=120)
                wait('XFE_WORKSPACE|Mode|USB|Persistent|1|')
                assert 'XFE_DISPLAY_OUTPUTS|Count|' + ('1' if boot_number == 3 else '2') + '|' in text()
                if boot_number == 1:
                    wait_state(Mode='Duplicate', Outputs=2, Pending=0)
                    captures('Initial-Duplicate', 'same')
                    key('meta_l-p'); wait('XFE_DESKTOP_ACTIVE|Settings')
                    captures('Settings', 'same')
                    key('i'); wait_state(Identify=1); captures('Identify')
                    wait_state(Identify=0); captures('Identify-Removed', 'same')
                    _, (x, y, w, h) = window_geometry(text(), 'Settings')
                    controls_y = y + 100 + min(220, max(110, h-385)) + 12
                    click(x + w - 150, controls_y + 16)
                    click(x + w - 150, controls_y + 34 + 34 + 16)
                    click(x + 90, y + h - 57)
                    wait_state(Mode='Extend', Width=2560, Pending=1)
                    captures('Extend-Preview')
                    key('k'); wait_state(Pending=0)

                    # Drag the actual monitor tile across its neighbor. Preview
                    # changes must reach separate physical scanouts, then Escape reverts.
                    area_w, area_h = w - 48, min(220, max(110, h-385))
                    scale = min((area_w-36)/2560, (area_h-30)/800)
                    tile_x = int(x+24+(area_w-2560*scale)/2+1920*scale)
                    tile_y = int(y+100+area_h/2)
                    move(tile_x, tile_y); button(True); move(int(tile_x-2560*scale), tile_y); button(False)
                    key('ret'); wait_state(Pending=1)
                    captures('Arranged-Left')
                    key('esc'); wait_state(Pending=0)
                    captures('Arrangement-Reverted')

                    # A fresh file window moves across the real monitor edge.
                    key('alt-f4'); key('meta_l-e'); time.sleep(.4)
                    window_id, (fx, fy, fw, fh) = window_geometry(text(), 'Files')
                    move(fx+70, fy+16); button(True); move(2140, 100); button(False)
                    key('meta_l-up'); time.sleep(.4)
                    _, geometry = window_geometry(text(), 'Files')
                    assert list(geometry) == [1280, 0, 1280, 800], geometry
                    captures('Window-On-Second')
                    boot['WindowMovedAndMaximizedOnSecond'] = True

                    key('meta_l-p'); key('2'); key('p'); key('ret')
                    wait_state(Primary=2, Pending=1); key('k'); wait_state(Pending=0)
                    captures('Second-Main')
                    key('m'); key('ret'); wait_state(Mode='SelectedOnly', Primary=2, Pending=1)
                    captures('Only-Second', 'first-black')
                    wait_state(Mode='Extend', Pending=0)  # real 15-second timer, no host rollback
                    captures('Timeout-Restored')
                    boot['AutomaticRevertPassed'] = True
                    key('m'); key('m'); key('ret'); wait_state(Mode='Duplicate', Pending=1)
                    key('k'); wait_state(Pending=0); captures('Duplicate-Again', 'same')
                    key('m'); key('ret'); wait_state(Mode='Extend', Pending=1)
                    key('k'); wait_state(Pending=0, Primary=2)
                    captures('Saved-Extended')
                elif boot_number == 2:
                    wait_state(Mode='Extend', Outputs=2, Width=2560, Primary=2, Pending=0)
                    captures('ColdBoot-Restored')
                    boot['SavedLayoutRestored'] = True
                else:
                    wait_state(Mode='Duplicate', Outputs=1, Width=1280, Primary=1, Pending=0)
                    captures('Missing-Display-Safe-Fallback', 'single')
                    boot['MissingDisplaySafeFallback'] = True
                boot['Passed'] = True
                qmp.call('quit'); qemu.wait(timeout=10)
                boot['QemuExitCode'] = qemu.returncode; assert qemu.returncode == 0
            finally:
                if qmp is not None:
                    qmp.connection.close()
                if qemu is not None and qemu.poll() is None:
                    qemu.terminate(); qemu.wait(timeout=10)
                if (folder / 'Serial.log').exists():
                    result['EvidenceFiles'].append(ref(folder / 'Serial.log'))
        original, actual = image.read_bytes(), disk.read_bytes()
        offset, size = manifest['WorkspaceFirstLba']*512, manifest['WorkspaceBytes']
        assert original[:offset+512] == actual[:offset+512] and original[offset+size:] == actual[offset+size:]
        assert original[offset+512:offset+size] != actual[offset+512:offset+size]
        assert digest(image) == result['ImageSha256']
        assert all(digest(ROOT / path) == value for path, value in sources.items())
        result.update(Passed=True, SourcesUnchanged=True, SourceImageUnchanged=True, OnlyWorkspaceSectorsChanged=True,
            DuplicatePixelEqualityPassed=True, IndependentExtendPassed=True, DistinctPhysicalIdentifyPassed=True,
            SelectedOnlyPassed=True, AutomaticRevertPassed=True, WindowMovedAndMaximizedOnSecond=True,
            UsbColdBootLayoutRestorePassed=True, MissingDisplaySafeFallbackPassed=True)
    except Exception as error:
        result['Failure'] = str(error)
    output = proof / 'Result.json'
    output.write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(json.dumps({'Passed': result['Passed'], 'Result': str(output), 'Failure': result.get('Failure')}, indent=2), flush=True)
    return result, output


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--image', type=Path, required=True)
    args = parser.parse_args()
    passed, _ = verify_displays(args.image)
    raise SystemExit(0 if passed['Passed'] else 1)