path: root/test/column_tags.py

#!/usr/bin/env python3
"""Exercise editable global, column and pane tags through real host input.

python3 -B test/column_tags.py zig-out/column-review/bin/pardes-gui \\
    /tmp/pardes-column-tags --project /path/to/agave

9P observes state and prepares owned scratch panes. SGR mouse events and
keyboard bytes enter through the session's real PTY input parser. SDL mode
captures GPU PNGs; --tty records the core grid and styles. This is not an
independent terminal-emulator screenshot or an SDL pixel-coordinate test.
"""
import argparse
import codecs
import contextlib
import json
import os
from pathlib import Path
import re

from fs import session
from ninep import Client
from ui_review import capture, command, panes, require, wait_for


class Input:
    def __init__(self, fd):
        self.fd = fd

    def send(self, data):
        os.write(self.fd, data.encode() if isinstance(data, str) else data)

    def mouse(self, button, col, row, release=False, drag=False):
        code = button + (32 if drag else 0)
        self.send(f'\x1b[<{code};{col + 1};{row + 1}{"m" if release else "M"}')

    def click(self, col, row, button=0):
        self.mouse(button, col, row)
        self.mouse(button, col, row, release=True)

    def drag(self, first, last, row):
        self.mouse(0, first, row)
        self.mouse(0, last, row, drag=True)
        self.mouse(0, last, row, release=True)

    def escape(self):
        # SDL's test feed reserves a partial ESC]777;finger prefix. CSI-u
        # identifies Escape unambiguously without waiting for another byte.
        self.send(b'\x1b[27u')


def row_text(client, row):
    frame = client.screen()
    return ''.join(cell[0] for cell in frame['cells'][row * frame['cols']:(row + 1) * frame['cols']]).rstrip()


def locate(client, word, row=None, minimum_col=0):
    """Find an ASCII needle using cell positions, even after wide glyphs."""
    frame = client.screen()
    rows = range(frame['rows']) if row is None else [row]
    for y in rows:
        cells = frame['cells'][y * frame['cols']:(y + 1) * frame['cols']]
        for x in range(minimum_col, frame['cols']):
            if ''.join(cell[0] for cell in cells[x:x + len(word)]) == word:
                return x, y
    return None


def await_text(client, row, text):
    require(wait_for(lambda: text in row_text(client, row)), f'row {row} never showed {text!r}')


def tag(client, serial):
    return client.read(f'/pane/{serial}/tag').decode().rstrip('\n')


def new_serial(client, previous):
    return next((serial for serial in panes(client) if serial not in previous), None)


def exec_tag_without_cleanup(client, serial, text):
    """Restore can replace the whole core, so never touch the old pane after it."""
    client.write(f'/pane/{serial}/tag', (' ' + text).encode())
    whole = client.read(f'/pane/{serial}/tag')
    start = whole.rfind(text.encode())
    client.write(f'/pane/{serial}/event', f'Mx{start} {start + len(text.encode())}\n'.encode())


def reconnect(address):
    try:
        return Client(address)
    except (OSError, EOFError):
        return None


def run(args):
    root = args.output.resolve()
    root.mkdir(parents=True, mode=0o700, exist_ok=True)
    root.chmod(0o700)
    config = root / 'config/pardes'
    config.mkdir(parents=True, exist_ok=True)
    (config / 'init').write_text('Shell /bin/sh\nTheme orchard\n')
    source = args.project.resolve() / 'gossip/src/crds_value.rs'
    original = source.read_bytes()
    name = 'column-tags-' + str(os.getpid())
    launch = ['-c', 'stty cols 150 rows 48; exec "$@"', 'column-tags',
              str(args.binary.resolve()), '--9p=' + name, str(source)]
    env = {} if args.tty else {
        'PARDES_TEST': '1', 'PARDES_TEST_CAPTURE_DIR': str(root),
        'PARDES_TEST_COLS': '150', 'PARDES_TEST_ROWS': '48', 'SDL_VIDEODRIVER': 'x11',
    }
    handles = {}
    checks = []
    with session('/bin/sh', root, name, socket_name=name, tty=True,
                 launch=launch, inherited=env, terminal=handles) as (client, address), contextlib.ExitStack() as connections:
        print('socket', address, flush=True)
        keys = Input(handles['input_fd'])
        await_text(client, 1, 'New Tty Find Grep Joincol')
        require(locate(client, 'New Tty', row=1) == (0, 1), 'left column tag does not start flush-left')
        require(tag(client, 1).startswith(str(source) + ' Save Tty Del'),
                'generated gap still separates the filename from pane commands')
        checks.append('pane commands follow the filename with one space')

        # Global header: actual left drag replaces Joincol, End appends Unicode,
        # and backspace removes a wide character as one grapheme.
        keys.drag(7, 14, 0)
        keys.send('Help')
        require(wait_for(lambda: row_text(client, 0).startswith('Newcol Help ')), 'global drag replacement failed')
        keys.send(b'\x1b[F')
        keys.send(' Ω界')
        await_text(client, 0, ' Ω界')
        keys.send(b'\x7f')
        require(wait_for(lambda: row_text(client, 0).endswith(' Ω')), 'wide-character backspace split a grapheme')
        keys.send('界')
        await_text(client, 0, ' Ω界')
        global_text = row_text(client, 0)
        keys.escape()
        checks.append('global drag replacement, typing, Unicode and grapheme backspace')

        before = panes(client)
        keys.click(*require(locate(client, 'Newcol', row=0), 'Newcol missing'), button=1)
        # Newcol makes an empty column and gives its tag the keyboard; the
        # pane made next fills it.
        require(wait_for(lambda: client.read('/focus').strip() == b''), 'Newcol did not focus its empty column')
        right = int(client.read('/pane/new'))
        right_name = root / 'right.rs'
        client.write(f'/pane/{right}/name', f'{right_name}\n'.encode())
        right_body = b'// Column-local scratch\nfn main() {\n    let ready = true;\n}\n'
        client.write(f'/pane/{right}/body', right_body)
        client.write(f'/pane/{right}/dirty', b'0\n')
        require(wait_for(lambda: locate(client, 'New Tty', row=1, minimum_col=75)), 'right column tag missing')
        require(locate(client, 'New Tty', row=1, minimum_col=75) == (75, 1),
                'right column tag retains an inset')
        checks.append('column tags start at the first cell of each column')

        # Column header uses half-open drag ranges; replacement must not leak
        # into its neighbour, the global header, or the active pane's body.
        col_start, _ = locate(client, 'New Tty', row=1, minimum_col=75)
        keys.click(col_start, 1)
        keys.send(b'\x1b[F')
        keys.send(' Scope λ界')
        await_text(client, 1, 'Scope λ界')
        scope_x, _ = require(locate(client, 'Scope', row=1, minimum_col=75), 'column suffix missing')
        keys.drag(scope_x, scope_x + len('Scope'), 1)
        keys.send('Work')
        await_text(client, 1, 'Work λ界')
        require('Scope' not in row_text(client, 1), 'column replacement appended to selection')
        keys.escape()
        require(row_text(client, 0) == global_text, 'column edit changed global tag')
        checks.append('column drag replacement and Unicode are scoped to that column')

        # Click the left source body, then execute New on the inactive RIGHT
        # column. A distinct marker must render on the right half afterwards.
        keys.click(12, 4)
        before = panes(client)
        keys.click(col_start, 1, button=1)
        extra = require(wait_for(lambda: new_serial(client, before)), 'right-column New did not create a pane')
        client.write(f'/pane/{extra}/body', b'RIGHT-COLUMN-NEW\n')
        require(wait_for(lambda: locate(client, 'RIGHT-COLUMN-NEW', minimum_col=75)), 'right-column command targeted another column')
        client.write(f'/pane/{extra}/exec', b'Del\n')
        checks.append('middle-click New targets the inactive right column')

        # An owned short-path file makes its complete tag visible. Pane-tag
        # selections retain the existing inclusive endpoint convention.
        save_x, pane_y = require(wait_for(lambda: locate(client, 'Save Tty', minimum_col=75)), 'right pane actions missing')
        keys.click(save_x, pane_y)
        keys.send(b'\x1b[F')
        keys.send(' TailMark')
        require(wait_for(lambda: tag(client, right).endswith(' TailMark')), 'pane tail typing failed')
        mark_x, mark_y = require(locate(client, 'TailMark', minimum_col=75), 'pane tail marker not visible')
        keys.drag(mark_x, mark_x + len('TailMark') - 1, mark_y)
        keys.send('Local')
        require(wait_for(lambda: tag(client, right).endswith(' Local')), 'pane selection replacement failed')
        keys.escape()
        checks.append('pane command tail supports typing and selection replacement')

        filename_x, filename_y = require(locate(client, 'right.rs', minimum_col=75), 'short filename not visible')
        keys.click(filename_x, filename_y)
        keys.send('X')
        require(wait_for(lambda: 'Xright.rs' in tag(client, right)), 'filename draft did not become editable')
        keys.escape()
        require(wait_for(lambda: tag(client, right).startswith(str(right_name) + ' ')), 'Escape did not cancel filename draft')
        require(not (root / 'Xright.rs').exists(), 'editing a filename wrote a file')
        require(client.read(f'/pane/{right}/body') == right_body, 'tag editing changed file body')
        checks.append('filename draft cancels without disk writes or body changes')

        # Keep a real embedded shell in the right column for useful captures.
        keys.click(12, 4)
        before = panes(client)
        keys.click(*require(locate(client, 'Tty', row=1, minimum_col=75), 'Tty missing'), button=1)
        terminal = require(wait_for(lambda: new_serial(client, before)), 'column Tty did not create a pane')
        require('pty' in client.list(f'/pane/{terminal}'), 'Tty created a non-terminal pane')
        client.write(f'/pane/{terminal}/pty/data', b"printf 'Column-local terminal ready\\n'; pwd\r")
        require(wait_for(lambda: b'\nColumn-local terminal ready\n' in client.read(f'/pane/{terminal}/body').replace(b'\r', b'')), 'terminal command output missing')
        checks.append('right-column embedded terminal executes harmless commands')

        # Persistence is checked through real Dump/Restore dispatch and the
        # resulting live header rows, not just serializer string matching.
        command(client, right, 'Dump')
        dump_path = root / (name + '.dump.zon')
        require(wait_for(dump_path.exists), 'Dump did not create an isolated session file')
        dumped = dump_path.read_bytes()
        saved_global = re.search(rb'\.topbar_custom\s*=\s*"((?:[^"\\]|\\.)*)"', dumped)
        require(saved_global and codecs.escape_decode(saved_global[1])[0].decode() == global_text,
                'global custom tag absent from dump')
        require(b'New Tty Find Grep Joincol Work' in dumped, 'column custom tag absent from dump')
        keys.click(1, 0)
        keys.send(b'\x1b[F')
        keys.send(' TEMP')
        await_text(client, 0, ' TEMP')
        keys.escape()
        try:
            exec_tag_without_cleanup(client, right, 'Restore ' + str(dump_path))
        except (ConnectionResetError, BrokenPipeError, EOFError):
            # Restore replaces the core and listener; even the write's final
            # clunk may lose its connection after successful dispatch.
            pass
        client.socket.close()
        client = connections.enter_context(require(wait_for(lambda: reconnect(address)), '9P did not return after Restore'))
        require(wait_for(lambda: row_text(client, 0) == global_text), 'Restore lost the custom global tag')
        await_text(client, 1, 'Work λ界')
        checks.append('Dump/Restore preserves custom global and column text')

        source_serial = require(next((serial for serial, text in panes(client).items() if str(source) in text), None), 'source pane lost after restore')
        right = require(next((serial for serial, text in panes(client).items() if str(right_name) in text), None), 'scratch pane lost after restore')
        (root / 'verified-headers.json').write_text(json.dumps({
            'global': global_text, 'columns_row': row_text(client, 1),
            'pane_tag': tag(client, right), 'dump': str(dump_path),
        }, indent=2, ensure_ascii=False))

        # Keep the tested Unicode and persistence evidence above. The published
        # workspace uses useful command text, edited through the same real input
        # route rather than modifying screenshots or injecting rendered cells.
        clean_global = 'Newcol Help Changelog Tutor Dump NextColor Debug Kill'
        keys.click(1, 0)
        keys.send(b'\x1b[H\x1b[1;2F')
        keys.send(clean_global)
        require(wait_for(lambda: row_text(client, 0) == clean_global), 'clean global tag replacement failed')
        keys.escape()
        col_start, _ = require(locate(client, 'New Tty', row=1, minimum_col=75), 'right column tag missing after restore')
        keys.click(col_start, 1)
        keys.send(b'\x1b[H\x1b[1;2F')
        keys.send('New Tty Find Grep Joincol Work')
        require(wait_for(lambda: row_text(client, 1).endswith('New Tty Find Grep Joincol Work')),
                'clean column tag replacement failed')
        keys.escape()
        clean_pane = str(right_name) + ' Save Tty Del Collapse'
        current_pane = tag(client, right)
        require(current_pane.startswith(clean_pane), 'unexpected scratch tag before presentation cleanup')
        suffix = current_pane[len(clean_pane):]
        require(suffix.isascii(), 'presentation cleanup expects its own ASCII test suffix')
        keys.click(*require(locate(client, 'Save Tty', minimum_col=75), 'scratch pane commands missing'))
        keys.send(b'\x1b[F')
        keys.send(b'\x7f' * len(suffix))
        require(wait_for(lambda: tag(client, right) == clean_pane), 'clean pane command tail replacement failed')
        keys.escape()
        checks.append('presentation tags cleaned through actual input after preserving Unicode evidence')
        for theme in ['orchard', 'atelier']:
            command(client, source_serial, 'Theme ' + theme)
            capture(client, root, 'columns-' + theme, not args.tty)
        # The source path exceeds the left column's width. Moving its tag
        # caret to End must reveal the actions without changing the filename.
        source_tag = tag(client, source_serial)
        keys.click(12, 2)
        keys.send(b'\x1b[F')
        require(wait_for(lambda: (point := locate(client, 'Save Tty Del', row=2))
                         and point[0] < 75), 'long-path tag did not reveal its command tail at End')
        require(tag(client, source_serial) == source_tag, 'caret movement changed the source tag')
        capture(client, root, 'columns-tag-reveal', not args.tty)
        keys.escape()
        checks.append('long-path pane tag reveals its command tail when the caret moves to End')
        require(client.read(f'/pane/{source_serial}/body') == original == source.read_bytes(), 'Agave source changed')
        checks.append('Agave source bytes unchanged')
        for serial in panes(client):
            if 'pty' in client.list(f'/pane/{serial}'):
                client.write(f'/pane/{serial}/pty/data', b'exit\r')
    report = {'checks': checks, 'cols': 150, 'rows': 48, 'host': 'tty' if args.tty else 'sdl',
              'input': 'keyboard and SGR mouse through the host PTY parser',
              'source': str(source), 'artifacts': str(root)}
    (root / 'report.json').write_text(json.dumps(report, indent=2))
    print(json.dumps(report, indent=2))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('binary', type=Path)
    parser.add_argument('output', type=Path)
    parser.add_argument('--project', required=True, type=Path)
    parser.add_argument('--tty', action='store_true')
    run(parser.parse_args())