XFESystem

"""Host-only parser regressions, including each byte boundary of actual guest logs.

This does not boot QEMU or qualify a framebuffer/performance acceptance result.
"""
import argparse
import hashlib
import json
import pathlib
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'Tools'))
from TestDesktopUi import parse_render_stats


def expect_malformed(record):
    try:
        parse_render_stats(record + '\n')
    except ValueError as error:
        assert 'Malformed complete XFE_RENDER_STATS record' in str(error), str(error)
        return str(error)
    raise AssertionError('A complete malformed record was silently accepted: ' + repr(record))


def check_record(record):
    assert record.endswith('\n')
    parsed = parse_render_stats(record)
    assert len(parsed) == 1 and parsed[0]['Frequency'] > 0
    # Every prefix before the LF is still an unacknowledged serial tail. Keep
    # the previous sample intact until that exact terminator has arrived.
    for offset in range(len(record)):
        assert parse_render_stats(record[:offset]) == [], (offset, record[:offset])
        assert parse_render_stats(record + record[:offset]) == parsed, offset
    assert parse_render_stats(record + record) == parsed + parsed
    return len(record) * 2 + 1


def run(log_paths):
    sample = ('XFE_RENDER_STATS|518311632290|Frames|2205|Partial|2143|Pixels|1053509288|Glyphs|67|Fallback|0|'
              'LastRenderTicks|56330|Frequency|4447722000|MotionReuses|3|MotionPixels|3512816|ScanoutPixelsSaved|2267872')
    checks = check_record(sample + '\n') + check_record(sample + '\r\n')
    # Complete protocol corruption must fail even when good samples precede it.
    malformed = [sample.replace('|Frequency|4447722000', '|Frequency|0'),
                 sample.replace('|Frequency|4447722000', ''),
                 sample + '|Frames|1', sample + '|Timestamp|1', sample + '|Unpaired',
                 sample.replace('|Frames|2205', '|Frames|'),
                 sample.replace('|Frames|2205', '|Frames|-1'),
                 sample.replace('|Frames|2205', '|Frames|+1'),
                 sample.replace('|Frames|2205', '|Frames|1.5'),
                 sample.replace('|Frames|2205', '|Frames|1e3'),
                 sample.replace('|Frames|2205', '|Frames|２２０５'),
                 sample.replace('|Frames|2205', '|Frames|1\v2'),
                 sample.replace('|Frames|2205', '|Frames|9223372036854775808'),
                 sample.replace('|Frames|2205', '|Frames|1\r2'),
                 sample.replace('|Frames|2205', '|1Invalid|2205'),
                 'XFE_RENDER_STATS', 'XFE_RENDER_STATS|123']
    failures = [expect_malformed(record) for record in malformed]
    for record in malformed:
        expect_malformed(sample + '\n' + record)
    assert parse_render_stats('Unrelated diagnostic\n' + sample + '\nUnfinished diagnostic') == parse_render_stats(sample + '\n')
    assert parse_render_stats(sample + '|FutureCounter|12\n')[0]['FutureCounter'] == 12

    logs = []
    for path in log_paths:
        raw = path.read_bytes()
        text = raw.decode('utf-8', errors='replace')
        parsed = parse_render_stats(text)
        assert parsed, 'No actual complete render-stat records: ' + str(path)
        records = [line for line in text.splitlines(keepends=True) if line.startswith('XFE_RENDER_STATS|') and line.endswith('\n')]
        assert len(records) == len(parsed)
        boundary_checks = sum(check_record(record) for record in records)
        checks += boundary_checks
        logs.append({'Path': str(path), 'Sha256': hashlib.sha256(raw).hexdigest(), 'CompleteRecords': len(parsed),
                     'EveryOffsetChecks': boundary_checks, 'LastCompleteRecord': parsed[-1],
                     'UnterminatedRenderStatsTailIgnored': text.rsplit('\n', 1)[-1].startswith('XFE_RENDER_STATS|')})
    return {'Kind': 'HostNativeRenderStatsParserRegression', 'QemuLaunched': False, 'Passed': True,
            'ParserSha256': hashlib.sha256((ROOT / 'Tools/TestDesktopUi.py').read_bytes()).hexdigest(),
            'EveryOffsetChecks': checks, 'CompleteMalformedCasesRejected': len(malformed), 'MalformedDiagnostics': failures,
            'RealGuestLogs': logs}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--log', action='append', required=True, type=pathlib.Path)
    parser.add_argument('--output', required=True, type=pathlib.Path)
    options = parser.parse_args()
    result = run([path.resolve() for path in options.log])
    options.output.parent.mkdir(parents=True, exist_ok=True)
    options.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps({key: result[key] for key in ('Kind', 'Passed', 'EveryOffsetChecks', 'CompleteMalformedCasesRejected')}))
    return 0


if __name__ == '__main__':
    raise SystemExit(main())