#!/usr/bin/env python3
import contextlib
import fcntl
import json
import os
from pathlib import Path
import subprocess
import struct
import sys
import tempfile
import termios
import threading
import time

from ninep import Client


@contextlib.contextmanager
def session(binary, root, name, *options, socket_name=None, file=None, tty=False, network=None,
            launch=None, inherited=None, terminal=None):
    env = {key: value for key, value in os.environ.items()
           if not key.startswith('PARDES_')}
    env.update(HOME=str(root), XDG_RUNTIME_DIR=str(root),
               XDG_CONFIG_HOME=str(root / 'config'), PARDES_NOTIME='1',
               PARDES_DUMP=str(root / (name + '.dump.zon')))
    env.update(inherited or {})
    for language in ['RS', 'C', 'GO', 'TS', 'PY']:
        env['PARDES_LSP_' + language] = ''
    fixture = root / (name + '.txt')
    fixture.write_bytes(b'initial\n')
    address = root / ('pardes-9p-' + (socket_name or name) + '.sock')
    child = None
    master = slave = None
    reader = None
    try:
        args = [binary, '--detach=' + name, *options, str(file or fixture)]
        preexec = None
        if tty:
            master, slave = os.openpty()
            fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', 24, 80, 0, 0))
            env['TERM'] = 'xterm-256color'
            args = [binary, '--tty', '--9p=' + (socket_name or name), *options, str(file or fixture)]
            preexec = lambda: fcntl.ioctl(0, termios.TIOCSCTTY, 0)
        if launch is not None:
            args = [binary, *launch]
        child = subprocess.Popen(args, cwd=root, env=env, stdin=slave,
                                 stdout=slave if tty else subprocess.DEVNULL,
                                 stderr=subprocess.PIPE, start_new_session=True,
                                 preexec_fn=preexec)
        if terminal is not None:
            # Borrowed handles for input-driving probes, valid inside this context.
            terminal.update(pid=child.pid, input_fd=master)
        if launch is not None and socket_name is None:
            address = root / f'pardes-9p-{child.pid}.sock'
        if slave is not None:
            os.close(slave)
            slave = None
            def drain():
                try:
                    while os.read(master, 65536):
                        pass
                except OSError:
                    pass
            reader = threading.Thread(target=drain, daemon=True)
            reader.start()
        deadline = time.monotonic() + 10
        while not address.exists():
            if child.poll() is not None:
                raise AssertionError((args, child.returncode, child.stderr.read().decode()))
            if time.monotonic() > deadline:
                raise TimeoutError('default 9P socket did not appear')
            time.sleep(.005)
        with Client(address) as client:
            if network is not None:
                for dial in client.read('/self/listeners').decode().splitlines():
                    parts = dial.split('!')
                    if parts[0] in ['tcp', 'quic']:
                        assert len(parts) == 3 and 0 < int(parts[2]) < 65536, dial
                        network[parts[0].upper()] = int(parts[2])
                assert set(network) == {'TCP', 'QUIC'}, network
            yield client, address
    finally:
        if terminal is not None:
            terminal.clear()
        if child is not None:
            if child.poll() is None:
                child.terminate()
            try:
                child.wait(timeout=5)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
            child.stderr.close()
        if slave is not None:
            os.close(slave)
        if master is not None:
            os.close(master)
        if reader is not None:
            reader.join(timeout=1)
            assert not reader.is_alive(), 'terminal reader did not stop'


def new_pane(client, contents):
    serial = int(client.read('/self/new/ctl').split()[0])
    client.write(f'/self/pane/{serial}/body', contents)
    return serial


def look(client, path, source=None):
    serial = source if source is not None else new_pane(client, b'')
    client.write(f'/self/pane/{serial}/body', path.encode(), truncate=True)
    client.write(f'/self/pane/{serial}/event', f'ML0 {len(path.encode())}\n'.encode())
    rows = client.read('/self/index').splitlines()
    opened = int(rows[-1].split()[0])
    assert opened != serial, f'Look did not open {path}'
    return opened


def execute(client, serial, command):
    text = command.encode()
    client.write(f'/self/pane/{serial}/body', text, truncate=True)
    client.write(f'/self/pane/{serial}/event', f'MX0 {len(text)}\n'.encode())


def discovery(binary):
    with tempfile.TemporaryDirectory(prefix='pardes-discovery-') as directory:
        with session(binary, Path(directory), 'discovery') as (client, _):
            before = client.read('/self/index')
            entries = client.list('/self/new')
            assert {'README', 'ctl', 'body', 'addr', 'data', 'event'} <= set(entries)
            guide = client.read('/self/README')
            assert guide == client.read('/self/new/README')
            assert b'Each open creates a separate pane' in guide
            # Walk and stat each entry, as a filesystem browser does, without open.
            from ninep import string
            for name in entries:
                client.fid += 1
                fid = client.fid
                client.rpc(110, struct.pack('<IIH', 1, fid, 3) +
                           string('self') + string('new') + string(name))
                try:
                    client.rpc(124, struct.pack('<I', fid))
                finally:
                    client.close(fid)
            assert client.read('/self/index') == before, 'browsing created a pane'
            first = int(client.read('/self/new/ctl').split()[0])
            second = int(client.read('/self/new/ctl').split()[0])
            assert first != second
            client.write(f'/self/pane/{first}/body', b'first pane', truncate=True)
            assert client.read(f'/self/pane/{first}/body') == b'first pane'
            assert client.read(f'/self/pane/{second}/body') == b''
            before_ids = set(client.list('/self/pane'))
            client.write('/self/new/body', b'created with text', truncate=True)
            made = set(client.list('/self/pane')) - before_ids
            assert len(made) == 1
            assert client.read('/self/pane/' + made.pop() + '/body') == b'created with text'
    print('9P discovery: listing/stat/README are inert; opens create independent panes')


def test(binary, quic=False):
    started = time.monotonic()
    for options, message in [
        (['--fs'], b'no such option'),
        (['--fs9'], b'no such option'),
        (['--9p'], b'--9p needs a socket name'),
        (['--9p='], b'invalid 9P socket name'),
        (['--9p=a/b'], b'invalid 9P socket name'),
        (['--attach=work', '--9p=work'], b'--attach has none'),
        (['--9p-tcp'], b'--9p-tcp needs'),
        (['--9p-tcp='], b'--9p-tcp needs'),
        (['--9p-tcp=tcp!localhost!5640'], b'--9p-tcp needs'),
        (['--9p-tcp=quic!127.0.0.1!5640'], b'--9p-tcp needs'),
        (['--9p-tcp=tcp!127.0.0.1!5640', '--9p-tcp=tcp!127.0.0.1!5641'], b'specified twice'),
        (['--attach=work', '--9p-tcp=tcp!127.0.0.1!5640'], b'--attach has none'),
        (['--9p-quic'], b'--9p-quic needs'),
        (['--9p-quic='], b'--9p-quic needs'),
        (['--9p-quic=tcp!127.0.0.1!5640'], b'--9p-quic needs'),
        (['--mount=os=work'], b'invalid mount name'),
        (['--mount=peer='], b'invalid mount name or dial'),
        (['--mount=peer=one', '--mount=peer=two'], b'duplicate mount'),
        (['--mount=peer=tcp!127.0.0.1!0'], b'invalid mount dial'),
        (['--attach=work', '--mount=peer=one'], b'--attach has none'),
    ]:
        result = subprocess.run([binary, *options], capture_output=True, timeout=5)
        assert result.returncode == 1 and message in result.stderr, (options, result)
    with tempfile.TemporaryDirectory(prefix='pardes-fs-') as directory:
        root = Path(directory)
        with session(binary, root, 'renamed', '--9p=explicit', socket_name='explicit') as (named, named_address):
            assert named_address.name == 'pardes-9p-explicit.sock'
            assert not (root / 'pardes-9p-renamed.sock').exists()
            assert named.read('/self/pane/1/body') == b'initial\n'
        with session(binary, root, 'first') as (client, address):
            assert set(client.list('/')) == {'os', 'self'}
            assert {'pane', 'index', 'new'} <= set(client.list('/self'))
            assert client.read('/self/pane/1/body') == b'initial\n'
            assert client.read('/os' + str(root / 'first.txt')) == b'initial\n'
            assert 'first.txt' in client.list('/os' + str(root))
            assert client.read('/self/pane/1/../1/body') == b'initial\n'
            assert client.read('/os' + str(root) + '/../' + root.name + '/first.txt') == b'initial\n'

            listing = root / 'listing'
            listing.mkdir()
            (listing / 'broken').symlink_to(listing / 'missing')
            for number in range(24):
                (listing / f'entry-{number:02}').touch()
            with Client(address, msize=256) as small:
                assert set(small.list('/os' + str(listing))) == {
                    f'entry-{number:02}' for number in range(24)}

            frozen = client.open('/self/screen')
            before = bytearray(client.read_fid(frozen, count=31))
            client.write('/self/pane/1/body', b'screen changed\n')
            while chunk := client.read_fid(frozen, len(before)):
                before.extend(chunk)
            client.close(frozen)
            old_screen = json.loads(before)
            old_text = ''.join(cell[0] for cell in old_screen['cells'])
            assert 'initial' in old_text and 'screen changed' not in old_text
            assert 'screen changed' in ''.join(cell[0] for cell in client.screen()['cells'])

            payload = ('const value = "λ";\n' * 3000).encode()
            serial = new_pane(client, payload)
            body = f'/self/pane/{serial}/body'
            assert client.read(body) == payload
            client.write(body, b'replaced\n', truncate=True)
            assert client.read(body) == b'replaced\n'

            event = client.open(f'/self/pane/{serial}/event')
            client.write(body, b'event\n')
            record = client.read_fid(event)
            assert record.startswith(b'EI') and b'event\n' in record, record
            client.close(event)

            os_path = '/os' + str(root / 'first.txt')
            client.write(os_path, b'written through 9P\n', truncate=True)
            assert (root / 'first.txt').read_bytes() == b'written through 9P\n'
            assert client.read(os_path) == b'written through 9P\n'

            for index in range(12):
                with Client(address) as other:
                    assert other.read(body) == b'replaced\nevent\n', index

            missing = False
            try:
                client.read('/self/pane/4294967295/body')
            except OSError:
                missing = True
            assert missing
            assert client.read(body) == b'replaced\nevent\n'

            source = look(client, '/virtual/src/pardes.zig')
            assert b'pub const Pardes' in client.read(f'/self/pane/{source}/body')

            (root / 'index').write_bytes(b'OS takes precedence\n')
            opened = look(client, 'index')
            assert client.read(f'/self/pane/{opened}/body') == b'OS takes precedence\n'
            opened = look(client, '/n/self/index')
            index = client.read(f'/self/pane/{opened}/body')
            assert index.split()[0] == b'1' and b'first.txt' in index, index
            (root / 'explicit.txt').write_bytes(b'explicit OS mount\n')
            opened = look(client, '/n/os' + str(root / 'explicit.txt'))
            assert client.read(f'/self/pane/{opened}/body') == b'explicit OS mount\n'

            own = root / 'pardes-9p-second.sock'
            with session(binary, root, 'second', '--mount=peer=' + str(address),
                         '--mount=own=' + str(own)) as (remote, _):
                opened = look(remote, '/n/own/self/pane/1/body')
                assert remote.read(f'/self/pane/{opened}/body') == b'initial\n'
                opened = look(remote, '/n/peer/self/pane/' + str(serial) + '/body')
                assert remote.read(f'/self/pane/{opened}/body') == b'replaced\nevent\n'
                remote.write(f'/self/pane/{opened}/body', b'saved to peer\n', truncate=True)
                remote.write(f'/self/pane/{opened}/ctl', b'put\n')
                assert client.read(body) == b'saved to peer\n'

                large = b'mounted file contents\n' * 60000
                large_path = root / 'large.txt'
                large_path.write_bytes(large)
                opened = look(remote, '/n/peer/os' + str(large_path))
                assert remote.read(f'/self/pane/{opened}/body') == large
                edited = large[:-3] + b'edited\n'
                remote.write(f'/self/pane/{opened}/body', edited, truncate=True)
                remote.write(f'/self/pane/{opened}/ctl', b'put\n')
                assert large_path.read_bytes() == edited
                assert client.read('/os' + str(large_path)) == edited

            with session(binary, root, 'runtime') as (remote, own_address):
                control = new_pane(remote, b'')
                execute(remote, control, 'Mount peer ' + str(address))
                opened = look(remote, '/n/peer/self/pane/1/body', source=control)
                assert remote.read(f'/self/pane/{opened}/body') == client.read('/self/pane/1/body')
                execute(remote, control, 'Mount peer ' + str(own_address))
                execute(remote, control, 'Unmount peer')
                remote.write(f'/self/pane/{opened}/body', b'runtime mounted Save\n', truncate=True)
                remote.write(f'/self/pane/{opened}/ctl', b'put\n')
                assert client.read('/self/pane/1/body') == b'runtime mounted Save\n'
                assert remote.read('/self/pane/1/body') == b'initial\n'
                roots = look(remote, '/n', source=control)
                assert b'/n/peer/\n' in remote.read(f'/self/pane/{roots}/body')
                remote.write(f'/self/pane/{opened}/ctl', b'delete\n')
                execute(remote, control, 'Unmount peer')
                remote.write(f'/self/pane/{roots}/ctl', b'get\n')
                assert remote.read(f'/self/pane/{roots}/body') == b'/n/os/\n/n/self/\n'
                execute(remote, control, 'Mount peer ' + str(own_address))
                opened = look(remote, '/n/peer/self/pane/1/body', source=control)
                assert remote.read(f'/self/pane/{opened}/body') == b'initial\n'

            with session(binary, root, 'mounted-startup', '--mount=peer=' + str(address),
                         file='/n/peer/self/pane/1/body:1') as (remote, _):
                assert remote.read('/self/pane/1/body') == b'runtime mounted Save\n'

            client.write(f'/self/pane/{serial}/ctl', b'delete\n')
            assert str(serial) not in client.list('/self/pane')

        for tty in [False, True]:
            name = 'restore tty space' if tty else 'restore detached space'
            with session(binary, root, name, tty=tty) as (old, address):
                old.write('/self/pane/1/body', b'dumped state\n', truncate=True)
                control = new_pane(old, b'')
                execute(old, control, 'Restore ' + str(root / 'missing dump.zon'))
                assert old.read('/self/pane/1/body') == b'dumped state\n'
                execute(old, control, 'Mount own ' + str(address))
                execute(old, control, 'Dump')
                saved = root / (name + '.dump.zon')
                deadline = time.monotonic() + 5
                while not saved.exists():
                    assert time.monotonic() < deadline, 'Dump did not write its state'
                    time.sleep(.005)
                held_body = old.open('/self/pane/1/body')
                held_screen = old.open('/self/screen')
                old_ids = {int(row.split()[0]) for row in old.read('/self/index').splitlines()}
                assert old.read_fid(held_body) == b'dumped state\n'
                assert old.read_fid(held_screen, count=1) == b'{'
                old.write('/self/pane/1/body', b'changed after dump\n', truncate=True)
                try:
                    execute(old, control, 'Restore ' + str(saved))
                except (EOFError, ConnectionResetError, BrokenPipeError):
                    pass
                try:
                    old.read_fid(held_body)
                except (EOFError, ConnectionResetError, BrokenPipeError):
                    pass
                else:
                    raise AssertionError('Restore left an old body fid usable')
                with Client(address) as restored:
                    restored_ids = [int(row.split()[0]) for row in restored.read('/self/index').splitlines()]
                    assert len(restored_ids) == 2 and old_ids.isdisjoint(restored_ids), restored_ids
                    file_id, control_id = restored_ids
                    contents = restored.read(f'/self/pane/{file_id}/body')
                    assert contents == b'dumped state\n', (contents, saved.read_text(), restored.read('/self/index'))
                    assert restored.read(f'/self/pane/{control_id}/body') == b'Dump'
                    assert restored.screen()['cols'] == 80
                    opened = look(restored, '/n/own/os' + str(root / (name + '.txt')), source=control_id)
                    assert restored.read(f'/self/pane/{opened}/body') == b'initial\n'

        with session(binary, root, 'forwarding') as (client, address):
            child_dir = root / 'child-working-directory'
            child_dir.mkdir()
            spaced = child_dir / 'space name.txt'
            spaced.write_bytes(b'first line\nsecond line\n')
            env = os.environ.copy()
            env.update(PARDES_9P=str(address), PARDES_PANE='1', PARDES_FORWARD_LOOK='1')
            for word, expected, selected, reuse in [('space name.txt:2:4', spaced.read_bytes(), [14, 14], False),
                                                    ('/n/self/pane/1/body', b'initial\n', None, False),
                                                    ('/virtual/pane/1/body:1:2-4', b'initial\n', [1, 4], True)]:
                before = {int(row.split()[0]) for row in client.read('/self/index').splitlines()}
                result = subprocess.run([binary, '--tty', word], cwd=child_dir, env=env,
                                        capture_output=True, timeout=3)
                assert result.returncode == 0, (word, result.stderr)
                after = {int(row.split()[0]) for row in client.read('/self/index').splitlines()}
                if reuse:
                    assert after == before
                else:
                    opened, = after - before
                assert client.read(f'/self/pane/{opened}/body') == expected
                assert client.read('/self/pane/1/body') == b'initial\n'
                if selected is not None:
                    addr = client.open(f'/self/pane/{opened}/addr')
                    client.write(f'/self/pane/{opened}/ctl', b'addr=dot\n')
                    actual = list(map(int, client.read_fid(addr).split()))
                    client.close(addr)
                    assert actual == selected, (word, actual, selected)
            addr = client.open('/self/pane/1/addr', 2)
            ctl = client.open('/self/pane/1/ctl', 1)
            for fid, text in [(addr, b'#0,#3'), (ctl, b'limit=addr\n'), (addr, b'#0')]:
                client.rpc(118, struct.pack('<IQI', fid, 0, len(text)) + text)
            try:
                for _ in range(8):
                    result = subprocess.run([binary], cwd=child_dir, env=env,
                                            capture_output=True, timeout=3)
                    assert result.returncode == 1 and b'already inside pardes' in result.stderr
                    text = b'/tial/'
                    try:
                        client.rpc(118, struct.pack('<IQI', addr, 0, len(text)) + text)
                    except OSError:
                        pass
                    else:
                        raise AssertionError('no-argument child cleared the parent address limit')
            finally:
                client.close(addr)
                client.close(ctl)
            client.read('/self/pane/1/ctl')

            control = new_pane(client, b'')
            execute(client, control, 'Dump')
            saved = root / 'forwarding.dump.zon'
            deadline = time.monotonic() + 3
            while not saved.exists():
                assert time.monotonic() < deadline, 'forwarding fixture Dump did not finish'
                time.sleep(.005)
            before = client.read('/self/index')
            inherited = {key: env[key] for key in ['PARDES_9P', 'PARDES_PANE', 'PARDES_FORWARD_LOOK']}
            cases = [
                ('mount', ['--mount=peer=' + str(address), '/n/peer/self/pane/1/body'], None),
                ('name', ['--9p=forwarded-name'], 'forwarded-name'),
                ('tcp', ['--9p-tcp=tcp!127.0.0.1!0'], None),
                ('load', ['-l', str(saved)], None),
                ('shells', ['-n', '3'], None),
                ('toggle', ['--tty-toggle=q'], None),
                ('toggle-arg', ['--tty-toggle', 'q'], None),
            ]
            if quic:
                cases.append(('quic', ['--9p-quic=quic!127.0.0.1!0'], None))
            for name, options, socket_name in cases:
                with session(binary, root, 'forward-' + name, tty=True, socket_name=socket_name,
                             launch=[*([] if name == 'shells' else ['--tty']), *options],
                             inherited=inherited) as (local, local_address):
                    assert local_address != address
                    index = local.read('/self/index').splitlines()
                    if name == 'mount':
                        assert len(index) == 1 and b'/n/peer/' in index[0], index
                        assert local.read('/self/pane/1/body') == b'initial\n'
                    elif name == 'load':
                        assert local.read('/self/index') == before
                        assert local.read('/self/pane/1/body') == b'initial\n'
                    else:
                        assert len(index) == (3 if name == 'shells' else 1), index
                    if name in ['tcp', 'quic']:
                        assert (name + '!127.0.0.1!').encode() in local.read('/self/listeners')
                    assert client.read('/self/index') == before
                    assert client.read('/self/pane/1/body') == b'initial\n'

            for name, identity, word in [
                ('stale-file', dict(inherited, PARDES_9P=str(root / 'absent.sock')), str(spaced)),
                ('stale-missing', dict(inherited, PARDES_9P=str(root / 'absent.sock')), 'missing-child-file.txt'),
                ('stale-noarg', dict(inherited, PARDES_9P=str(root / 'absent.sock')), None),
                ('stale-pane', dict(inherited, PARDES_PANE='4294967295'), str(spaced)),
                ('disabled', dict(inherited, PARDES_FORWARD_LOOK='0'), str(spaced)),
            ]:
                with session(binary, root, name, tty=True, inherited=identity,
                             launch=['--tty', *([word] if word else [])]) as (local, local_address):
                    assert local_address != address
                    if word == str(spaced):
                        assert local.read('/self/pane/1/body') == spaced.read_bytes()
                    elif word:
                        assert word.encode() in local.read('/self/index') or word in ''.join(
                            cell[0] for cell in local.screen()['cells'])
                    assert client.read('/self/index') == before

        for tty in [False, True]:
            name = 'lsp-tty' if tty else 'lsp-detached'
            with session(binary, root, name, tty=tty) as (client, _):
                control = new_pane(client, b'')
                execute(client, control, 'Lspinfo')
                deadline = time.monotonic() + 3
                while True:
                    matches = [int(row.split()[0]) for row in client.read('/self/index').splitlines()
                               if b'+Lsp' in row]
                    if matches:
                        break
                    assert time.monotonic() < deadline, (name, 'Lspinfo did not complete')
                    time.sleep(.005)
                assert len(matches) == 1, matches
                result = client.read(f'/self/pane/{matches[0]}/body')
                assert result.startswith(b'backend: zls-inproc+lsp-client\n'), result
                assert b'protocol servers (lsp-client):' in result, result
                assert client.read('/self/pane/1/body') == b'initial\n'
                assert '+Lsp' in ''.join(cell[0] for cell in client.screen()['cells'])

        with session(binary, root, 'highlight') as (client, _):
            client.write('/self/pane/1/ctl', b'name keywords.zig\n')
            source = b'// padding\n' * 9999 + b'fn check() void {\n    if (true) return;\n}\n'
            client.write('/self/pane/1/body', source, truncate=True)
            screen = client.screen()
            assert len(screen['cells']) == screen['cols'] * screen['rows']
            found = set()
            for row in range(screen['rows']):
                cells = screen['cells'][row * screen['cols']:(row + 1) * screen['cols']]
                text = ''.join(cell[0] for cell in cells)
                for word in ['fn', 'if']:
                    start = text.find(word + ' ')
                    if start < 0:
                        continue
                    styles = [screen['styles'][cell[1]] for cell in cells[start:start + len(word)]]
                    assert all(style['bold'] and style['fg'] == styles[0]['fg'] for style in styles), (word, styles)
                    found.add(word)
            assert found == {'fn', 'if'}, screen

        with session(binary, root, 'long-names') as (client, _):
            directory = root / ('directory-' + 'd' * 54)
            directory.mkdir()
            for length in [29, 63, 127, 255]:
                name = 'f' * (length - 4) + '.txt'
                path = directory / name
                path.write_bytes(b'long filename\n')
                wire_path = '/os' + str(path)
                assert name in client.list('/os' + str(directory))
                assert client.read(wire_path) == b'long filename\n'
                client.write(wire_path, b'written through long filename\n', truncate=True)
                assert path.read_bytes() == b'written through long filename\n'
                opened = look(client, '/n/os' + str(path))
                assert client.read(f'/self/pane/{opened}/body') == path.read_bytes()
            opened = look(client, '/n/os' + str(directory))
            listing = client.read(f'/self/pane/{opened}/body')
            for path in directory.iterdir():
                assert ('/n/os' + str(path)).encode() in listing
    if quic:
        quic_test(binary)
    print(f'9P integration: passed ({time.monotonic() - started:.3f}s)')


def quic_test(binary):
    for tty in [False, True]:
        for host in ['127.0.0.1', '::1']:
            with tempfile.TemporaryDirectory(prefix='pardes-quic-fs-') as directory:
                root = Path(directory)
                ports = {}
                with session(binary, root, 'server', '--9p-tcp=tcp!' + host + '!0',
                             '--9p-quic=quic!' + host + '!0', tty=tty, network=ports) as (unix, address):
                    assert set(unix.read('/self/listeners').decode().splitlines()) == {
                        'unix!' + str(address), 'tcp!' + host + '!' + str(ports['TCP']),
                        'quic!' + host + '!' + str(ports['QUIC'])}
                    with Client((host, ports['TCP'])) as tcp:
                        unix_body = unix.open('/self/pane/1/body')
                        tcp_body = tcp.open('/self/pane/1/body')
                        with session(binary, root, 'client') as (remote, _):
                            control = new_pane(remote, b'')
                            dial = 'quic!' + host + '!' + str(ports['QUIC'])
                            execute(remote, control, 'Mount peer ' + dial)
                            opened = look(remote, '/n/peer/self/pane/1/body', source=control)
                            body = f'/self/pane/{opened}/body'
                            assert remote.read(body) == b'initial\n'
                            payload = 'saved café λ\n'.encode() + b'x' * (3 * 8192 + 27) + b'\n'
                            remote.write(body, payload, truncate=True)
                            remote.write(f'/self/pane/{opened}/ctl', b'put\n')
                            assert unix.read_fid(unix_body) == payload[:unix.msize - 11]
                            assert tcp.read('/self/pane/1/body') == payload
                            tcp.write('/self/pane/1/body', b'TCP to QUIC Get\n', truncate=True)
                            remote.write(f'/self/pane/{opened}/ctl', b'get\n')
                            assert remote.read(body) == b'TCP to QUIC Get\n'
                            assert unix.read_fid(unix_body) == b'TCP to QUIC Get\n'
                            assert tcp.read_fid(tcp_body) == b'TCP to QUIC Get\n'
                            screen_pane = look(remote, '/n/peer/self/screen', source=control)
                            screen = json.loads(remote.read(f'/self/pane/{screen_pane}/body'))
                            assert len(screen['cells']) == screen['cols'] * screen['rows']
                            assert 'TCP to QUIC Get' in ''.join(cell[0] for cell in screen['cells'])
                            large = ('normal λ line\n' * 20000).encode()
                            fixture = root / 'large.txt'
                            fixture.write_bytes(large)
                            large_pane = look(remote, '/n/peer/os' + str(fixture), source=control)
                            assert remote.read(f'/self/pane/{large_pane}/body') == large
                            edited = b'QUIC file saved\n'
                            remote.write(f'/self/pane/{large_pane}/body', edited, truncate=True)
                            remote.write(f'/self/pane/{large_pane}/ctl', b'put\n')
                            assert fixture.read_bytes() == edited
                            for pane in [opened, screen_pane, large_pane]:
                                remote.write(f'/self/pane/{pane}/ctl', b'delete\n')
                            execute(remote, control, 'Unmount peer')
                            roots = look(remote, '/n', source=control)
                            assert remote.read(f'/self/pane/{roots}/body') == b'/n/os/\n/n/self/\n'
                            assert unix.read_fid(unix_body) == b'TCP to QUIC Get\n'
                            assert tcp.read_fid(tcp_body) == b'TCP to QUIC Get\n'
                        control = new_pane(unix, b'')
                        execute(unix, control, 'Mount own ' + dial)
                        own = look(unix, '/n/own/self/pane/1/body', source=control)
                        assert unix.read(f'/self/pane/{own}/body') == b'TCP to QUIC Get\n'
                        unix.write(f'/self/pane/{own}/body', b'same-session QUIC Save\n', truncate=True)
                        unix.write(f'/self/pane/{own}/ctl', b'put\n')
                        assert unix.read_fid(unix_body) == b'same-session QUIC Save\n'
                        assert tcp.read_fid(tcp_body) == b'same-session QUIC Save\n'
                        unix.write(f'/self/pane/{own}/ctl', b'delete\n')
                        execute(unix, control, 'Unmount own')
                        unix.close(unix_body)
                        tcp.close(tcp_body)


if __name__ == '__main__':
    if len(sys.argv) == 3 and sys.argv[2] == '--discovery':
        discovery(str(Path(sys.argv[1]).resolve()))
        raise SystemExit(0)
    if len(sys.argv) not in [2, 3] or (len(sys.argv) == 3 and sys.argv[2] != '--quic'):
        raise SystemExit('usage: fs.py <pardes> [--quic | --discovery]')
    test(str(Path(sys.argv[1]).resolve()), quic=len(sys.argv) == 3)