XFESystem

"""Actual 4K Monaco edit/save smoke; not physical scanout or Windows parity."""
import argparse
import hashlib
import json
from pathlib import Path
import re
import shutil
import socket
import subprocess
import sys
import time
import traceback
from PIL import Image

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT/'Tools'))
from TestPointerCadence import QmpClient, require_backing
sys.path.insert(0, str(ROOT/'Tools/CSharpBoot'))
from ReadImage import inspect_image, verify_single_edition
from MultiprocessorEvidence import validate_multiprocessors


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


def run(args):
    out=args.output.resolve();out.mkdir(parents=True,exist_ok=False)
    result={'Kind':'XfeNative4KEditorSmoke','Passed':False,'Physical240HzVerified':False,
            'WindowsFluidityParityPassed':False,'PhysicalHardwareAcceptancePassed':False,
            'ImageSha256':digest(args.image),'KernelSha256':digest(args.kernel),
            'HarnessSha256':{p:digest(ROOT/p) for p in ('Tools/IndependentKernel/TestHighResolutionDesktop.py',
                'Tools/IndependentKernel/MultiprocessorEvidence.py',
                'Tools/TestPointerCadence.py','Tools/CSharpBoot/ReadImage.py')}}
    process=qmp=None
    try:
        build=json.loads(Path(str(args.kernel)+'.json').read_text(encoding='utf-8-sig'))
        assert build['Passed'] and build['KernelSha256']==result['KernelSha256']
        assert all(digest(ROOT/p)==h for p,h in build['SourceSha256'].items())
        inspected,files=inspect_image(args.image)
        manifest=json.loads(Path(str(args.image)+'.json').read_text(encoding='utf-8-sig'))
        assert files['System/Kernel.elf']==args.kernel.read_bytes() and manifest['Resolution']=='0x0'
        result['SingleEditionImageProof']=verify_single_edition(inspected,files,manifest)
        shutil.copyfile(ROOT/'.tools/firmware/edk2-i386-vars.fd',out/'Vars.fd')
        with socket.socket() as reserve:reserve.bind(('127.0.0.1',0));port=reserve.getsockname()[1]
        command=[str(ROOT/'.tools/qemu/qemu-system-x86_64.exe'),'-machine','q35','-accel','whpx','-cpu','max',
            '-smp',str(args.virtual_cpus),'-m','1024M','-vga','std','-global','VGA.vgamem_mb=64','-global','VGA.edid=on',
            '-global','VGA.xres=3840','-global','VGA.yres=2160','-global','VGA.xmax=3840','-global','VGA.ymax=2160',
            '-drive',f'if=pflash,format=raw,readonly=on,file={ROOT/".tools/firmware/edk2-x86_64-code.fd"}',
            '-drive',f'if=pflash,format=raw,file={out/"Vars.fd"}',
            '-drive',f'if=ide,format=raw,snapshot=on,file={args.image.resolve()}',
            '-display','none','-net','none','-serial',f'file:{out/"Serial.log"}',
            '-qmp',f'tcp:127.0.0.1:{port},server=on,wait=off','-no-reboot','-no-shutdown']
        result['QemuArguments']=command
        with (out/'Qemu.log').open('wb') as log:
            process=subprocess.Popen(command,cwd=ROOT,stdout=log,stderr=subprocess.STDOUT,
                                     creationflags=getattr(subprocess,'CREATE_NO_WINDOW',0))
        deadline=time.monotonic()+180
        while time.monotonic()<deadline:
            assert process.poll() is None,'Private QEMU exited during startup'
            try:qmp=QmpClient(port);break
            except (OSError,TimeoutError):time.sleep(.1)
        assert qmp is not None
        require_backing(qmp.call('query-block'),args.image)
        def healthy():
            assert process.poll() is None,'Private QEMU exited'
            value=(out/'Serial.log').read_text(errors='replace') if (out/'Serial.log').exists() else ''
            assert not any(x in value for x in ('XFE_DIAGNOSTIC_EVIDENCE_LOST|','XFE_NATIVE_FAIL','XFE_NATIVE_EXCEPTION','XFE_RUNTIME_FATAL',
                'XFE_ZERO_FAILFAST','XFE_INDEPENDENT_APPLICATION_FAILED','XFE_RUNTIME_OOM|'))
            return value
        def wait(predicate,timeout=180):
            end=time.monotonic()+timeout
            while time.monotonic()<end:
                value=healthy()
                if predicate(value):return value
                time.sleep(.05)
            raise TimeoutError('4K actual editor acknowledgement missing')
        first=wait(lambda value:'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME|Width|3840|Height|2160|ScalePercent|150|' in value)
        if 'Source/XFESystem.NativeKernel/Multiprocessor.cs' in build['SourceSha256']:
            result['MultiprocessorEvidence']=validate_multiprocessors(first,args.virtual_cpus)
            result['ExecutingCpus']=result['MultiprocessorEvidence']['ExecutingCpus']
            result['ManagedSchedulerCpus']=1
        assert re.search(r'XFE_WINDOW_STATE\|1\|Files\|150\|66\|800\|500\|',wait(lambda value:'XFE_WINDOW_STATE|1|Files|' in value))
        pointer=[1920,1080]
        def click(x,y):
            target=[round(x*1.5),round(y*1.5)]
            while pointer!=target:
                dx=max(-64,min(64,target[0]-pointer[0]));dy=max(-64,min(64,target[1]-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(.01)
            for down in (True,False):
                qmp.call('input-send-event',{'events':[{'type':'btn','data':{'down':down,'button':'left'}}]});time.sleep(.08)
        def key(chord):
            qmp.call('human-monitor-command',{'command-line':'sendkey '+chord});time.sleep(.12)
        def model(value):
            values=re.findall(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END',value,re.S)
            return values[-1] if values else None
        def screenshot(name):
            ppm=out/(name+'.ppm');png=out/(name+'.png');qmp.call('screendump',{'filename':str(ppm)})
            with Image.open(ppm) as frame:
                assert frame.size==(3840,2160);frame.convert('RGB').save(png)
            return {'Path':png.name,'Sha256':digest(png)}
        # The first actual sorted Files entry is the unchanged seeded Hello.cs.
        click(220,166)
        expected='using System;\n\nConsole.ForegroundColor = ConsoleColor.Cyan;\nConsole.WriteLine("Hello XFESystem!");\nConsole.ResetColor();\n'
        painted=re.compile(r'^XFE_MONACO_PRESENTATION\|Window\|2\|RasterPaints\|[1-9][0-9]*\|CachedPaints\|[0-9]+\|WorkerThread\|[0-9]+\r?$',re.M)
        # A file-backed serial stream can be observed halfway through a record.
        # Require the complete positive acknowledgement, not its prefix.
        value=wait(lambda value:model(value)==expected and 'XFE_EDITOR_BACKGROUND|Ready|' in value and painted.search(value))
        before=screenshot('4K-Monaco-Before')
        def comment_pixels(capture):
            # The seeded file has no comments. Require actual new comment ink
            # in the code area; a changed header, caret or pointer cannot pass.
            with Image.open(out/capture['Path']) as frame:
                code=frame.crop(tuple(round(v*1.5) for v in (172+44,88+67+28,172+680,88+67+170))).convert('RGB')
                return sum(all(abs(pixel[i]-color)<=12 for i,color in enumerate((0x57,0xa6,0x4a))) for pixel in code.get_flattened_data())
        initial_comment_pixels=comment_pixels(before)
        key('ctrl-end');key('slash');key('slash');key('q')
        edited=expected+'//q'
        wait(lambda value:model(value)==edited)
        key('ctrl-s')
        wait(lambda value:'XFE_DESKTOP_SAVED_TEXT_BEGIN\n'+edited+'\nXFE_DESKTOP_SAVED_TEXT_END' in value)
        end=time.monotonic()+60
        while time.monotonic()<end:
            after=screenshot('4K-Monaco-Edited')
            edited_comment_pixels=comment_pixels(after)
            if edited_comment_pixels-initial_comment_pixels>=45:break
            healthy();time.sleep(.15)
        assert edited_comment_pixels-initial_comment_pixels>=45,'Actual new comment glyph pixels are absent at 4K'
        assert before['Sha256']!=after['Sha256'],'Actual edited framebuffer did not change'
        qmp.call('quit');process.wait(timeout=15)
        assert process.returncode==0 and digest(args.image)==result['ImageSha256']
        assert all(digest(ROOT/p)==h for p,h in build['SourceSha256'].items())
        assert all(digest(ROOT/p)==h for p,h in result['HarnessSha256'].items())
        result.update(Passed=True,ActualWidth=3840,ActualHeight=2160,ScalePercent=150,
                      InitialModel=expected,EditedModel=edited,ActualEditAndSave=True,Screenshots=[before,after],
                      InitialCommentPixels=initial_comment_pixels,EditedCommentPixels=edited_comment_pixels,
                      QemuExitCode=process.returncode,ImageUnchanged=True,SerialSha256=digest(out/'Serial.log'))
    except Exception as error:
        result['Failure']=str(error) or type(error).__name__
        result['Traceback']=traceback.format_exc()
    finally:
        if qmp:
            try:qmp.close()
            except OSError:pass
        if process and process.poll() is None:
            process.terminate()
            try:process.wait(timeout=10)
            except subprocess.TimeoutExpired:process.kill();process.wait(timeout=10)
        (out/'Result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
    print(json.dumps({'Passed':result['Passed'],'Failure':result.get('Failure'),'Receipt':str(out/'Result.json')}))
    return 0 if result['Passed'] else 1


if __name__=='__main__':
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('kernel',type=Path);parser.add_argument('image',type=Path);parser.add_argument('output',type=Path)
    parser.add_argument('--virtual-cpus',type=int,choices=(1,2,8,32),default=8)
    raise SystemExit(run(parser.parse_args()))