#!/usr/bin/env python3
"""Actual public QEMU CLI, guest software, twelve ports and removal lifecycle."""
import argparse
import io
import json
import hashlib
import lzma
import os
import pty
import selectors
import termios
from pathlib import Path
import subprocess
import sys
import tarfile
import tempfile
import time

project = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(project / 'tools'))
from image_formats import digest, gpt, LINUX_FILESYSTEM
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--cli', type=Path, required=True)
parser.add_argument('--qemu-runner', type=Path)
args = parser.parse_args()
work = Path(tempfile.mkdtemp(prefix='emu-test.', dir=project / 'out'))
builder = Path((project / 'out/workstation-images-current.txt').read_text().strip())
assert json.loads((builder / 'acceptance.json').read_text())['status'] == 'passed'
software = builder / 'software.img'
shared = builder / 'shared.img'
log = (work / 'commands.log').open('w')
cli = str(args.cli.resolve())
session = work / 'normal'
start_options = ['--qemu-runner', str(args.qemu_runner.resolve())] if args.qemu_runner else []
inputs = [software, shared, project / 'out/fds-system-cli.img', project / 'out/fds-initramfs.img', project / 'out/kernel/boot/kernel_2712.img', Path(cli)]
(work / 'inputs.sha256').write_text(''.join(f'{digest(path)}  {path}\n' for path in inputs))

def invoke(*command, ok=True):
    result = subprocess.run([cli, '--session', str(session), *map(str, command)], capture_output=True, text=True, timeout=240)
    log.write(repr(command) + '\n' + result.stdout + result.stderr)
    log.flush()
    if (result.returncode == 0) != ok and command and command[0] == 'guest':
        # Preserve live service evidence before the finally block stops the VM.
        # Do not retry or relax the failed command's acceptance condition.
        diagnostics = subprocess.run([
            cli, '--session', str(session), 'guest', '--', 'sh', '-c',
            'cat /run/log/cartridged/current; s6-svstat /run/service/cartridged; '
            'dmesg | tail -40',
        ], capture_output=True, text=True, timeout=60)
        log.write('Failure diagnostics:\n' + diagnostics.stdout + diagnostics.stderr)
        log.flush()
    assert (result.returncode == 0) == ok, (command, result.returncode, result.stdout, result.stderr)
    return result.stdout

def guest(*command, ok=True):
    return invoke('guest', '--', *command, ok=ok)

def query(*command):
    return json.loads(guest('fds', '--json', *command))

def wait_bay(number, expected, timeout=60):
    deadline = time.monotonic() + timeout
    while True:
        state = query('bay', number)['bays'][0]
        if state['state'] == expected:
            return state
        assert state['state'] != 'error' or expected == 'error', state
        assert time.monotonic() < deadline, state

def mounts():
    return guest('cat', '/proc/self/mountinfo')

def clean(number):
    text = mounts()
    assert f'/run/fds/software/{number:02}/' not in text
    assert '/run/fds/apps/demo.workstation' not in text

def interactive_program_checks():
    master, slave = pty.openpty()
    original = termios.tcgetattr(slave)
    console = subprocess.Popen([cli, '--session', str(session), 'console'], stdin=slave, stdout=slave, stderr=slave)
    selector = selectors.DefaultSelector(); selector.register(master, selectors.EVENT_READ)
    pending = bytearray()
    def until(marker):
        output = pending; deadline = time.monotonic() + 30
        while marker not in output:
            assert time.monotonic() < deadline, output
            for _, _ in selector.select(max(0, deadline-time.monotonic())):
                chunk = os.read(master, 8192); assert chunk
                output.extend(chunk)
        end = output.index(marker) + len(marker)
        result = bytes(output[:end]); del output[:end]
        log.write(repr(result) + '\n'); log.flush()
        return result
    try:
        until(b'FDS> ')
        os.write(master, b"stty -g >/tmp/program-tty-before; b01:demo.report:shell -c 'printf \"READY:%s\\n\" tty; read -r line; printf \"REPLY:%s\\n\" \"$line\"'\n")
        until(b'READY:tty\r\n')
        os.write(master, b'literal interactive input\n')
        until(b'REPLY:literal interactive input\r\n')
        until(b'FDS> ')
        os.write(master, b"b01:demo.report:shell -c 'printf \"SIGNAL:%s\\n\" ready; exec tail -f /dev/null'; printf 'RETURN:%s\\n' \"$?\"\n")
        until(b'SIGNAL:ready\r\n')
        os.write(master, b'\x03')
        until(b'RETURN:130\r\n')
        until(b'FDS> ')
        os.write(master, b"b01:demo.report:shell -c 'printf \"STOP:%s\\n\" ready; read -r resumed; printf \"RESUMED:%s\\n\" \"$resumed\"; exec tail -f /dev/null'\n")
        until(b'STOP:ready\r\n')
        os.write(master, b'\x1a')
        until(b'Stopped')
        until(b'FDS> ')
        os.write(master, b'fg\n')
        until(b"exec tail -f /dev/null'\r\n")
        # The fg command echo precedes terminal handoff and SIGCONT. Require
        # actual input/output from the resumed foreground program before Ctrl-C.
        os.write(master, b'continue\n')
        until(b'RESUMED:continue\r\n')
        os.write(master, b'\x03')
        until(b'FDS> ')
        os.write(master, b"test \"$(stty -g)\" = \"$(cat /tmp/program-tty-before)\" && printf 'RESTORE:%s\\n' passed\n")
        until(b'RESTORE:passed\r\n')
        os.write(master, b'\x1d')
        assert console.wait(timeout=10) == 0
        assert termios.tcgetattr(slave) == original
    finally:
        if console.poll() is None: console.kill(); console.wait()
        selector.close(); os.close(master); os.close(slave)

# Compatibility fixture only: the public creator never emits archive payloads.
legacy_metadata=work/'legacy-metadata'; (legacy_metadata/'FDS').mkdir(parents=True)
legacy_payload=work/'legacy-payload'; (legacy_payload/'bundles').mkdir(parents=True)
legacy_program=b'#!/bin/sh\nprintf "Legacy reader: %s\\n" "$1"\n'
legacy_tar=io.BytesIO()
with tarfile.open(fileobj=legacy_tar,mode='w',format=tarfile.USTAR_FORMAT) as archive:
    directory=tarfile.TarInfo('bin');directory.type=tarfile.DIRTYPE;directory.mode=0o755;archive.addfile(directory)
    program=tarfile.TarInfo('bin/legacy');program.mode=0o755;program.size=len(legacy_program);archive.addfile(program,io.BytesIO(legacy_program))
legacy_bytes=lzma.compress(legacy_tar.getvalue(),format=lzma.FORMAT_XZ)
(legacy_payload/'bundles/legacy.reader.tar.xz').write_bytes(legacy_bytes)
(legacy_metadata/'FDS/CARTRIDGE.TOML').write_text('format=1\n[cartridge]\nid="legacy.fixture"\nname="Legacy reader fixture"\nclass="program"\nversion="1"\n[media]\nwritable=false\n')
(legacy_metadata/'FDS/SOFTWARE.TOML').write_text(f'format=1\n[[software]]\nid="legacy.reader"\nname="Legacy reader"\nversion="1"\narchitecture="any"\npartition=2\narchive_bytes={len(legacy_bytes)}\nunpacked_bytes={len(legacy_program)}\nentries=2\nsha256="{hashlib.sha256(legacy_bytes).hexdigest()}"\n[software.commands]\nlegacy="bin/legacy"\n')
legacy_parts=[]
for name,tree in [('FDS_METADATA',legacy_metadata),('FDS_PAYLOAD02',legacy_payload)]:
    filesystem=work/(name+'.erofs')
    subprocess.run([str(project/'tools/in-image-tools'),'mkfs.erofs','--quiet','-T','0',str(filesystem),str(tree)],check=True,stdout=log,stderr=log)
    legacy_parts.append((name,LINUX_FILESYSTEM,filesystem))
legacy=work/'legacy.img';gpt(legacy,legacy_parts)
subprocess.run([str(project/'out/workstation/fds-cartridge'),'--image-tool-runner',str(project/'tools/in-image-tools'),'inspect',str(legacy)],check=True,stdout=log,stderr=log)

# A workstation-created disposable DATA fixture, never a host block device.
data_root = work / 'data-root'
(data_root / 'FDS').mkdir(parents=True)
(data_root / 'FDS/CARTRIDGE.TOML').write_text('format=1\n[cartridge]\nid="demo.data"\nname="Emulator DATA"\nclass="data"\nversion="1"\n[media]\nwritable=true\n')
filesystem = work / 'data.ext4'
with filesystem.open('xb') as stream: stream.truncate(32 * 1024 * 1024)
subprocess.run([str(project / 'tools/in-image-tools'), 'mke2fs', '-q', '-t', 'ext4', '-b', '4096', '-L', 'FDS_DATA', '-E', 'lazy_itable_init=0,lazy_journal_init=0,root_owner=1000:1000', '-d', str(data_root), str(filesystem)], check=True)
data = work / 'data.img'
gpt(data, [('FDS_DATA', LINUX_FILESYSTEM, filesystem)])
original_data = digest(data)
original_software = digest(software)
try:
    report = json.loads(invoke('start', *start_options))
    assert report['qemu']['running']
    initial_state = (session / 'session.json').read_bytes()
    invoke('start', *start_options, ok=False)
    assert (session / 'session.json').read_bytes() == initial_state
    assert guest('id', '-u').strip() == '1000'
    master, slave = pty.openpty()
    original_termios = termios.tcgetattr(slave)
    console = subprocess.Popen([cli, '--session', str(session), 'console'], stdin=slave, stdout=slave, stderr=slave)
    selector = selectors.DefaultSelector(); selector.register(master, selectors.EVENT_READ)
    def console_until(marker):
        output = bytearray(); deadline = time.monotonic() + 20
        while marker not in output:
            assert time.monotonic() < deadline, output
            for _, _ in selector.select(max(0, deadline-time.monotonic())):
                chunk = os.read(master, 8192); assert chunk
                output.extend(chunk)
        return output
    try:
        console_until(b'FDS> ')
        os.write(master, b'id -u\n')
        assert b'1000' in console_until(b'\r\n1000\r\n')
        os.write(master, b'\x1d')
        assert console.wait(timeout=10) == 0
        assert termios.tcgetattr(slave) == original_termios
        assert json.loads(invoke('status'))['qemu']['running']
    finally:
        if console.poll() is None: console.kill(); console.wait()
        selector.close(); os.close(master); os.close(slave)
    assert all(b['state'] == 'empty' for b in query('bays')['bays'])
    invoke('insert', 13, software, ok=False)
    invoke('insert', 1, '/dev/null', ok=False)
    assert guest('printf', '%s', 'literal $(id); and spaces') == 'literal $(id); and spaces'
    assert guest('printf', '%s', 'Kernel panic / FDS_STAGE0_ERROR: is just guest data') == 'Kernel panic / FDS_STAGE0_ERROR: is just guest data'
    guest('false', ok=False)
    for number in range(1, 13):
        invoke('insert', number, shared if number == 6 else software)
        state = wait_bay(number, 'mounted_read_only')
        assert state['devices'][0]['serial'] == f'FDS-{number:02}'
        assert state['devices'][0]['topology'].endswith('/' + str(number))
        catalogue = state['software']['software']
        assert [s['partition'] for s in catalogue] == ([2, 2] if number == 6 else [2, 3])
        invoke('insert', number, software, ok=False)
        if number in [1, 6, 12]:
            guest('fds', 'run', number, '--', 'demo.hello:hello')
            guest('fds', 'run', number, '--', 'demo.report:report')
            guest('touch', f'/run/fds/software/{number:02}/payload02/programs/demo.hello/unexpected', ok=False)
            guest('fds', 'run', number, '--', '../escape', ok=False)
            guest('fds', 'run', number, '--', 'demo.hello:missing', ok=False)
            assert guest('hello', 'literal $(id); and spaces').strip() == 'Hello from an FDS AArch64 software cartridge: literal $(id); and spaces'
            assert guest(f'b{number:02}:demo.report:shell', '-c', 'id -u').strip() == '1000'
            assert guest('sh', '-c', f'printf "pipe input" | b{number:02}:demo.report:shell -c "cat"') == 'pipe input'
            assert guest('sh', '-c', f'b{number:02}:demo.report:shell -c "exit 37"; printf "%s" "$?"') == '37'
            assert guest('sh', '-c', f'cd /tmp; b{number:02}:demo.report:shell -c "pwd"').strip() == '/tmp'
            assert '/cache-' not in mounts()
            assert state['commands']
        if number == 1:
            interactive_program_checks()

        if number == 6:
            guest('fds', 'run', number, '--', 'demo.report:report', 'hold')
            guest('sh', '-c', 'report hold >/tmp/direct-hold.log 2>&1 & echo $! >/tmp/direct-launcher')
            deadline = time.monotonic() + 20
            while query('bay', number)['bays'][0]['consumers'] < 2:
                assert time.monotonic() < deadline
            assert query('bay', number)['bays'][0]['consumers'] >= 2
            invoke('unplug', number)
        else:
            invoke('eject', number)
        state = wait_bay(number, 'empty')
        assert state['consumers'] == 0
        clean(number)
        if number == 6:
            deadline = time.monotonic() + 20
            while guest('sh', '-c', 'test ! -e /proc/$(cat /tmp/direct-launcher); printf "%s" "$?"').strip() != '0':
                assert time.monotonic() < deadline
        assert not guest('sh', '-c', 'command -v hello || true').strip()
    # Two different cartridges export the same names. Lowest bay wins, and
    # fully qualified commands remain available before and after an eject.
    invoke('insert', 8, software); wait_bay(8, 'mounted_read_only')
    invoke('insert', 2, builder / 'collision.img'); wait_bay(2, 'mounted_read_only')
    assert '/02/' in guest('where', 'FDS_APP')
    assert '/08/' in guest('b08:demo.report:where', 'FDS_APP')
    invoke('eject', 2); wait_bay(2, 'empty')
    assert '/08/' in guest('where', 'FDS_APP')
    invoke('eject', 8); wait_bay(8, 'empty')
    remaining = json.loads(invoke('status'))
    assert remaining['state']['cartridges'] == {}
    assert len(remaining['block_nodes']) == 2, remaining['block_nodes']
    output = guest('cat', '/run/log/cartridged/current')
    assert 'Hello from an FDS AArch64 software cartridge' in output
    assert 'FDS cartridge system report' in output
    assert 'uid=1000(fds)' in output
    (work / 'program-output.log').write_text(output)
    invoke('insert', 3, builder / 'program-corrupt.img')
    failed = wait_bay(3, 'error')
    assert 'digest' in failed['detail'].lower() or 'hash' in failed['detail'].lower(), failed
    assert '/run/fds/software/03/' not in mounts()
    invoke('unplug', 3)
    wait_bay(3, 'empty')
    invoke('insert', 4, builder / 'catalogue-mapping.img')
    wait_bay(4, 'error')
    invoke('eject', 4, ok=False)
    invoke('unplug', 4)
    wait_bay(4, 'empty')
    clean(4)
    inserted = json.loads(invoke('insert', 2, data))
    overlay = Path(inserted['inserted']['overlay'])
    assert overlay.is_file() and overlay.suffix == '.qcow2'
    wait_bay(2, 'mounted_read_write')
    guest('sh', '-c', 'printf "overlay survived safe eject\\n" >/data/workstation.txt')
    assert guest('cat', '/data/workstation.txt').strip() == 'overlay survived safe eject'
    invoke('eject', 2)
    wait_bay(2, 'empty')
    assert digest(data) == original_data
    assert overlay.is_file()
    assert len(json.loads(invoke('status'))['block_nodes']) == 2
    invoke('insert', 7, legacy); wait_bay(7, 'mounted_read_only')
    assert guest('legacy', 'compatibility').strip() == 'Legacy reader: compatibility'
    assert '/run/fds/software/07/cache-legacy.reader' in mounts()
    guest('fds', 'run', 7, '--', 'legacy.reader:legacy', 'background')
    invoke('eject', 7); wait_bay(7, 'empty'); clean(7)
    assert not guest('sh', '-c', 'command -v legacy || true').strip()
    # Keep a directly mounted software tree active across native shutdown.
    invoke('insert', 12, software)
    wait_bay(12, 'mounted_read_only')
    guest('fds', 'run', 12, '--', 'demo.report:report', 'hold')
    invoke('stop')
    assert 'FDS_SHUTDOWN_FINAL' in (session / 'console.log').read_text()
    retained_log = (session / 'console.log').read_bytes()
    retained_data = {p: digest(p) for p in session.glob('data-*.qcow2')}
    assert retained_data
    previous_state = json.loads((session / 'session.json').read_text())
    restarted = json.loads(invoke('start', *start_options))
    assert restarted['qemu']['running'] and restarted['state']['cartridges'] == {}
    assert restarted['state']['name'] != previous_state['name']
    assert any(json.loads(p.read_text()) == previous_state for p in session.glob('previous-*.json'))
    assert (session / 'console.log').read_bytes().startswith(retained_log)
    assert all(digest(p) == value for p, value in retained_data.items())
    assert guest('id', '-u').strip() == '1000'
    assert all(b['state'] == 'empty' for b in query('bays')['bays'])
    invoke('insert', 1, software); wait_bay(1, 'mounted_read_only')
    assert 'Hello from' in guest('hello', 'restarted')
    invoke('stop', '--force')
    assert json.loads(invoke('start', *start_options))['qemu']['running']
    assert guest('id', '-u').strip() == '1000'
    invoke('stop')
finally:
    subprocess.run([cli, '--session', str(session), 'stop', '--force'], capture_output=True, timeout=40)

# Isolated admin console fixture for service interruption and recovery. The
# preceding normal-image tests used UID 1000; production login is unchanged.
archive = work / 'admin.tar'
with tarfile.open(project / 'out/rootfs-cli.tar') as source, tarfile.open(archive, 'w', format=tarfile.PAX_FORMAT) as target:
    name = 'usr/libexec/fds/console-session'
    assert source.getmember(name)
    for member in source:
        if member.name == name: continue
        target.addfile(member, source.extractfile(member) if member.isfile() else None)
    member = tarfile.TarInfo(name)
    payload = b"#!/bin/bash\n# Isolated workstation lifecycle test only.\nexec env HOME=/root PS1='FDS> ' bash --noprofile --norc\n"
    member.size = len(payload); member.mode = 0o755
    target.addfile(member, io.BytesIO(payload))
admin = work / 'admin-image'; admin.mkdir()
with (work / 'admin-build.log').open('w') as build_log:
    subprocess.run([str(project / 'image/build-system-cartridge'), '--rootfs', str(archive), '--output-directory', str(admin)], check=True, stdout=build_log, stderr=subprocess.STDOUT)
session = work / 'admin'
try:
    invoke('start', '--system', admin / 'system.img', *start_options)
    assert guest('id', '-u').strip() == '0'
    invoke('insert', 1, software)
    wait_bay(1, 'mounted_read_only')
    guest('sh', '-c', 'cd /home/fds; /run/fds/bin/report hold >/tmp/restart-direct.log 2>&1 &')
    deadline = time.monotonic() + 20
    while query('bay', 1)['bays'][0]['consumers'] == 0:
        assert time.monotonic() < deadline
    assert '/run/fds/software/01/payload03' in mounts()
    assert '/cache-' not in mounts()
    guest('s6-rc', '-l', '/run/s6-rc', '-d', 'change', 'cartridged')
    guest('s6-rc', '-l', '/run/s6-rc', '-u', 'change', 'cartridged')
    wait_bay(1, 'mounted_read_only')
    assert query('bay', 1)['bays'][0]['consumers'] == 0
    assert '/run/fds/software/01/cache-' not in mounts()
    guest('fds', 'run', 1, '--', 'demo.hello:hello')
    assert 'Hello from' in guest('sh', '-c', 'cd /home/fds; /run/fds/bin/hello')
    guest('mkdir', '-m', '700', '/tmp/private-cwd')
    assert 'Permission denied' in guest('sh', '-c', 'cd /tmp/private-cwd; /run/fds/bin/hello', ok=False)
    # Extra payload aliases must prevent SAFE; program trees are read-only.
    guest('mkdir', '/run/extra-payload')
    guest('mount', '--bind', '/run/fds/software/01/payload02', '/run/extra-payload')
    invoke('eject', 1, ok=False)
    guest('umount', '/run/extra-payload')
    invoke('eject', 1)
    wait_bay(1, 'empty')
    clean(1)
    guest('fds-burn', 'create', 'program', '/tmp', '/tmp/program.img', ok=False)
    invoke('stop')
finally:
    subprocess.run([cli, '--session', str(session), 'stop', '--force'], capture_output=True, timeout=40)
assert digest(software) == original_software and digest(data) == original_data
record = dict(status='passed', ordinary_guest_uid=1000, public_emulator_cli=True,
              all_twelve_usb_bays=True, interactive_console_detach_and_terminal_restore=True, both_payload_partitions_executed=True,
              shared_partition_executed=True, readonly_program_trees_without_extraction=True, legacy_archive_host_and_guest_reader=True,
              direct_path_arguments_pipes_exit_status_cwd=True, interactive_program_io_signals_and_restore=True, command_collision_fallback=True,
              safe_eject=True, forced_removal_stops_consumer=True,
              corrupt_program_and_catalogue_rejected=True, data_overlay_preserves_source=True,
              restart_cleans_mounts_commands_and_consumers=True, extra_mount_blocks_safe=True,
              native_shutdown_with_active_software=True, same_directory_restart_after_stop=True,
              same_directory_restart_after_force=True, restart_retains_logs_overlays_and_mapping=True,
              live_session_restart_rejected=True, physical_pi='not tested')
(work / 'acceptance.json').write_text(json.dumps(record, indent=2) + '\n')
(project / 'out/workstation-emulator-current.txt').write_text(str(work) + '\n')
print('PASS: public emulator, guest software and lifecycle acceptance:', work)
print('SKIP: physical Pi, USB power-cycle behavior and timing require hardware')