#!/usr/bin/env python3
"""Verify the real unprivileged FDS prompt and readiness independent of eudev."""
import io
import json
from pathlib import Path
import subprocess
import sys
import tarfile
import tempfile

project = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(project/'tools'))
from image_formats import digest
from vm_test import VM

work = Path(tempfile.mkdtemp(prefix='m5-vm.', dir=project/'out'))
rootfs = project/'out/rootfs-cli.tar'
normal = (project/'out/fds-system-cli.img').resolve()
# Preserve the compiled graph and actual scripts. Only gate the udev executable
# in a disposable image; no clock delay or hardware access is involved.
stage2 = 'etc/s6-linux-init/current/scripts/rc.init'
replacements = {
    stage2: f'#!/bin/bash\nset -euo pipefail\nmkfifo -m 0666 /run/fds-test-udev-gate\nexec /{stage2}.real "$@"\n',
    'usr/bin/udevd': '#!/bin/bash\nset -euo pipefail\nprintf "FDS_TEST_UDEV_WAIT\\n" >/dev/console\nIFS= read -r release </run/fds-test-udev-gate\nexec /usr/bin/udevd.real "$@"\n',
}
with tarfile.open(rootfs) as source, tarfile.open(work/'gated.tar', 'w', format=tarfile.PAX_FORMAT) as output:
    found = set()
    for member in source:
        name = member.name
        if name in replacements:
            member.name += '.real'
            found.add(name)
        output.addfile(member, source.extractfile(member) if member.isfile() else None)
    assert found == replacements.keys()
    for name, text in replacements.items():
        data = text.encode()
        member = tarfile.TarInfo(name)
        member.mode = 0o755
        member.size = len(data)
        output.addfile(member, io.BytesIO(data))
(work/'gated').mkdir()
with (work/'gated-image.log').open('wb') as log:
    subprocess.run([str(project/'image/build-system-cartridge'), '--rootfs', str(work/'gated.tar'),
                    '--output-directory', str(work/'gated')], check=True, stdout=log, stderr=subprocess.STDOUT)
images = [('normal', normal), ('blocked-eudev', work/'gated/system.img')]
for name, image in images:
    before = digest(image)
    with VM(work, name, image) as vm:
        vm.expect(rb'FDS> ')
        vm.send('printf "\\nFDS_UID:%s\\n" "$(id -u)"; printf "FDS_HOME:%s:%s\\n" "$HOME" "$PWD"; touch "$HOME/prompt-write-check" && printf "FDS_HOME_WRITABLE\\n"; if touch /etc/unexpected-write 2>/dev/null; then printf "FDS_ROOT_BAD\\n"; else printf "FDS_ROOT_READ_ONLY\\n"; fi')
        vm.expect(rb'^FDS_UID:1000\r?$')
        vm.expect(rb'^FDS_HOME:/home/fds:/home/fds\r?$')
        vm.expect(rb'^FDS_HOME_WRITABLE\r?$')
        vm.expect(rb'^FDS_ROOT_READ_ONLY\r?$')
        if name == 'blocked-eudev':
            assert b'FDS_TEST_UDEV_WAIT' in vm.data, 'eudev was not held at the readiness gate'
            vm.send('printf "release\\n" >/run/fds-test-udev-gate; printf "FDS_GATE_RELEASED\\n"')
            vm.expect(rb'^FDS_GATE_RELEASED\r?$')
        vm.send('printf "\\nFDS_TRACE_BEGIN\\n"; fds --json boot-profile; printf "\\nFDS_TRACE_END\\n"')
        captured = vm.expect(rb'^FDS_TRACE_BEGIN\r?\n(\{.*?\})\r?\n\r?\nFDS_TRACE_END\r?$')
        report = json.loads(captured.group(1))
        assert report['missing_events'] == ['desktop-ready'], report
        assert 'virt' in report['platform'].lower(), report
        assert report['durations_ns']['kernel-to-console'] == report['events_ns']['console-ready']
        assert report['durations_ns']['s6-to-console'] > 0
        assert b'FDS_STAGE0_HANDOFF' not in vm.data, 'Production console leaked debug markers'
        (work/(name+'-boot-profile.json')).write_text(json.dumps(report, indent=2)+'\n')
        if name == 'normal':
            vm.send('exit')
            vm.expect(rb'READY\.\r?\nFDS> ')
            vm.send('printf "\\nFDS_TRACE_BEGIN\\n"; fds --json boot-profile; printf "\\nFDS_TRACE_END\\n"')
            restarted = json.loads(vm.expect(rb'^FDS_TRACE_BEGIN\r?\n(\{.*?\})\r?\n\r?\nFDS_TRACE_END\r?$').group(1))
            assert restarted['events_ns']['console-ready'] == report['events_ns']['console-ready'], 'Console restart replaced the first boot observation'
        print(f'PASS: {name}: FDS user prompt, ephemeral home, read-only SYSTEM and measured trace', flush=True)
    assert digest(image) == before, 'Read-only VM test altered SYSTEM'

# Run the real ARM regression tool on measured reports with controlled deltas.
report = json.loads((work/'normal-boot-profile.json').read_text())
baseline = work/'baseline.json'
baseline.write_text(json.dumps(report))
current = work/'current.json'
report['durations_ns']['kernel-to-console'] += 100_000_001
current.write_text(json.dumps(report))
command = [str(project/'tools/in-void'), 'qemu-aarch64', str(project/'out/fds-boottrace'), 'compare', str(baseline), str(current)]
assert subprocess.run(command, capture_output=True).returncode != 0
subprocess.run([*command, '--explain', 'Deliberate threshold test; not a measured regression'], check=True)
report['platform'] = 'different-test-platform'
current.write_text(json.dumps(report))
assert subprocess.run(command, capture_output=True).returncode != 0
target = project/'out/m5-vm-latest'
temporary = target.with_suffix('.next')
temporary.symlink_to(work.name)
temporary.replace(target)
print(f'PASS: M5 console and boottrace verification: {work}')
print('NOTE: test VMs exit via QMP after console checks; orderly shutdown is covered by M4/M10')
print('SKIP: physical Pi boot latency, firmware timing and display responsiveness')