XFESystem

"""Actual full-screen console acceptance; never injects shell or workspace state."""
import hashlib
import json
import time

from PIL import Image

from IndependentBingEvidence import json_blocks


CONSOLE_SNAPSHOT = 'XFE_CONSOLE_SNAPSHOT'
FORBIDDEN_CONSOLE_MARKERS = ('XFE_INDEPENDENT_DESKTOP_FIRST_FRAME', 'XFE_DESKTOP_',
                             'XFE_WINDOW_', 'XFE_BROWSER_', 'XFE_EDITOR_BACKGROUND|', 'XFE_MONACO_')
FAILURE_MARKERS = ('XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL', 'XFE_ZERO_FAILFAST',
                   'XFE_INDEPENDENT_APPLICATION_FAILED', 'XFE_INDEPENDENT_CONSOLE_FAILED')
STAGES = ('HelpPassed', 'RamDirectoryCreateWriteReadPassed', 'HistoryBeyondPreviousPassed',
          'HistoryDraftRestorePassed', 'CaretDeleteInsertEditingPassed', 'HistoryListingPassed',
          'NoDemoOnlyTerminalContent')


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


def terminal_command_output(snapshot, command):
    """Only output after the latest exact echoed command can satisfy a check."""
    assert snapshot.get('Busy') is False and snapshot.get('Input') == '', 'The actual terminal is still busy or has unsubmitted input'
    assert snapshot.get('LastCommand') == command, 'The actual terminal submitted another command'
    prompt = snapshot.get('Prompt')
    assert isinstance(prompt, str) and prompt.endswith('> ') and snapshot.get('Version'), 'The actual terminal has no versioned prompt'
    lines = [line.rstrip() for line in snapshot.get('Text', '').replace('\r', '').split('\n')]
    matches = [index for index, line in enumerate(lines) if line == prompt + command]
    assert matches, 'No exact actual echoed terminal command in the rendered snapshot'
    output = lines[matches[-1] + 1:]
    while output and not output[-1]: output.pop()
    assert not any(line.startswith(('Error:', 'Unknown command:', 'Managed program failed:')) for line in output), 'The actual terminal command failed: ' + repr(output)
    return output


def assert_console_isolation(log):
    assert 'XFE_SYSTEM_EDITION|Console' in log, 'The actual guest did not identify the Console edition'
    assert 'XFE_SYSTEM_EDITION|Desktop' not in log, 'The Console guest entered the Desktop edition'
    assert not any(marker in log for marker in FAILURE_MARKERS), log[-4096:]
    assert not any(marker in log for marker in FORBIDDEN_CONSOLE_MARKERS), 'The Console edition initialized a desktop/browser path'


def run_console(qmp, proof, text, input_kind='PS2'):
    started = time.monotonic(); beginning = len(text()); selected = []
    result = {'Kind': 'XfeIndependentConsoleWorkflow', 'Passed': False,
              'Input': f'Actual QMP {input_kind} keyboard packets only', 'HostCommandExecution': False,
              'HostWorkspaceReads': False, 'Commands': []}

    def healthy():
        log = text(); assert_console_isolation(log); return log

    def screenshot(name):
        path = proof / (name + '.ppm')
        qmp.call('screendump', {'filename': str(path)})
        with Image.open(path) as rendered:
            rgb = rendered.convert('RGB')
            assert rgb.size == (1280, 800), 'The actual console has unexpected framebuffer dimensions'
            rgb.save(proof / (name + '.png'))
        return {'Path': name + '.png', 'Sha256': digest(proof / (name + '.png'))}

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

    def type_text(value):
        special = {':': 'shift-semicolon', '.': 'dot', '/': 'slash', '-': 'minus', '_': 'shift-minus', ' ': 'spc'}
        for char in value:
            key(special.get(char, 'shift-' + char.lower() if char.isupper() else char))

    def wait_command(after, command, required_lines=()):
        deadline = time.monotonic() + 30
        last_error = 'No fresh complete rendered console snapshot'
        while time.monotonic() < deadline:
            packets = json_blocks(healthy(), CONSOLE_SNAPSHOT, after)
            if packets:
                packet = packets[-1]
                try:
                    lines = terminal_command_output(packet['Value'], command)
                    assert all(required in lines for required in required_lines), 'Missing actual post-command output: ' + repr(required_lines)
                    selected.append(packet)
                    result['Commands'].append({'Command': command, 'Passed': True, 'SerialOffset': packet['SerialOffset'],
                        'Prompt': packet['Value']['Prompt'], 'Version': packet['Value']['Version'], 'Busy': False,
                        'PostCommandOutput': lines})
                    return packet['Value'], lines
                except AssertionError as error:
                    last_error = str(error)
            time.sleep(.1)
        raise AssertionError('Actual console command timeout: ' + command + ': ' + last_error)

    def execute(command, required_lines=()):
        after = len(text()); type_text(command); key('ret')
        return wait_command(after, command, required_lines)

    try:
        healthy()
        result['BootScreenshot'] = screenshot('Console-Boot')
        snapshot, lines = execute('help')
        assert any('md Path | mf Path | cat Path | write Path Text' in line for line in lines), 'The console help lacks real file commands'
        assert any('Up/Down: history' in line for line in lines), 'The console help lacks history/caret editing'
        assert any('Storage: RAM session; shutdown discards files.' in line for line in lines), 'The console did not disclose RAM-only storage'
        assert 'Run Demo' not in snapshot['Text'], 'The removed demo-only terminal content remains'
        result['HelpPassed'] = result['NoDemoOnlyTerminalContent'] = True
        execute('md CliProof'); execute('mf CliProof/Command.txt')
        write_command = 'write CliProof/Command.txt NativeConsoleReady'
        execute(write_command)
        execute('cat CliProof/Command.txt', ('NativeConsoleReady',))
        result['RamDirectoryCreateWriteReadPassed'] = True

        after = len(text()); key('up'); key('up'); key('ret')
        wait_command(after, write_command)
        result['HistoryBeyondPreviousPassed'] = True
        after = len(text()); type_text('echo draft'); key('up'); key('down'); key('ret')
        wait_command(after, 'echo draft', ('draft',))
        result['HistoryDraftRestorePassed'] = True
        after = len(text()); type_text('echo aXc'); key('home')
        for _ in range(6): key('right')
        key('delete'); type_text('b'); key('right'); key('left'); key('insert'); type_text('d'); key('insert'); key('end'); key('ret')
        wait_command(after, 'echo abd', ('abd',))
        result['CaretDeleteInsertEditingPassed'] = True
        _, lines = execute('history')
        assert any(write_command in line for line in lines) and any('echo abd' in line for line in lines), 'Actual history omitted executed commands'
        result['HistoryListingPassed'] = True
        # The interpreter executes the same embedded .NET PE as the desktop.
        # Its own output plus the native completion marker must both appear.
        after = len(text())
        _, lines = execute('run ManagedHello.dll', ('Managed ConsoleIL exit: 0',))
        assert 'XFE_INDEPENDENT_CONSOLE_IL_OK' in healthy()[after:], 'The actual managed sample did not execute'
        assert any('XFE_MANAGED_APP_OK' in line for line in lines), 'The embedded application produced no own success output'
        assert any('XFE_MANAGED_XFECONSOLE_OK' in line for line in lines), 'The embedded XFEConsole sample produced no own success output'
        result['ManagedSamplePassed'] = True
        result['Screenshot'] = screenshot('Console-Commands')
        assert result['Screenshot']['Sha256'] != result['BootScreenshot']['Sha256'], 'Console keyboard input produced no changed guest framebuffer'
        # Check once more after an input-free delay; no desktop fallback or
        # asynchronous browser initialization may arrive after CLI success.
        time.sleep(2); healthy()
        result['NoDesktopOrBrowserInitialization'] = True
        result['Passed'] = True
    except Exception as error:
        result['Failure'] = str(error)
        raise
    finally:
        raw = proof / 'ConsoleCommandEvidence.log'
        raw.write_text('\n'.join(item['Raw'] for item in selected), encoding='utf-8')
        result['RawGuestEvidence'] = {'Path': raw.name, 'Sha256': digest(raw), 'SelectedRenderedSnapshots': len(selected), 'StartSerialOffset': beginning}
        result['ElapsedHostSeconds'] = time.monotonic() - started
        (proof / 'ConsoleWorkflow.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    return result