path: root/packages/meshbay-cli/tests/test_interactive.py

"""
The interactive mode, driven by Textual's pilot against the in-memory hub and
node: what a person does with the keys, and what lands on disk and on the node.
"""

import asyncio
import time

import pytest
from fake_hub import PASSPHRASE, USERNAME, FakeHub
from fake_node import FakeNode
from meshbay_cli import __main__ as cli
from meshbay_cli.client import pool
from meshbay_cli.commands import context
from meshbay_cli.tui.app import MeshBayApp
from meshbay_cli.tui.panes import LocalPane, RemotePane
from meshbay_cli.tui.screens import MainScreen, SignInScreen
from meshbay_cli.tui.widgets import Choice


@pytest.fixture
def world(monkeypatch, tmp_path):
    hub, node = FakeHub(), FakeNode()
    monkeypatch.setattr(context, "Hub", lambda url: hub)

    async def open_group(account, group):
        return node.session()
    monkeypatch.setattr(pool, "open_group", open_group)
    work = tmp_path / "work"
    work.mkdir()
    monkeypatch.chdir(work)
    return hub, node, work


def _app() -> MeshBayApp:
    args = cli.build_parser().parse_args(["--hub", "https://hub.test"])
    return MeshBayApp(context.Context(args))


async def _until(pilot, condition, timeout: float = 10) -> None:
    deadline = time.monotonic() + timeout
    while not condition():
        assert time.monotonic() < deadline, "timed out"
        await pilot.pause(0.05)


def _names(pane) -> list[str]:
    """Row names as shown, without the mark or a note such as "(read-only)"."""
    return [pane.table.get_row_at(i)[0].plain.split("  (")[0].strip()
            for i in range(pane.table.row_count)]


async def _open(pilot, pane, name: str) -> None:
    await _until(pilot, lambda: name in _names(pane))
    pane.table.move_cursor(row=_names(pane).index(name))
    pane.table.focus()
    await pilot.press("enter")


async def test_signing_in_then_downloading_and_uploading(world):
    hub, node, work = world
    node.add("share", "notes.txt", b"hello")
    node.add("share/2024", "a.jpg", b"A" * 3000)
    node.add("share", "[bold]x[/bold]\x1b.txt", b"tricky")
    (work / "mine.txt").write_bytes(b"from here")

    app = _app()
    async with app.run_test(size=(140, 40)) as pilot:
        # Nothing stored: the sign-in screen, then the panes.
        await _until(pilot, lambda: isinstance(app.screen, SignInScreen))
        app.screen.query_one("#user").value = USERNAME
        app.screen.query_one("#pass").value = PASSPHRASE
        app.screen.query_one("#pass").focus()
        await pilot.press("enter")
        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
        assert len(hub.devices) == 1, "signing in here remembers the machine"

        remote = app.screen.query_one(RemotePane)
        local = app.screen.query_one(LocalPane)
        await _open(pilot, remote, f"Family@{USERNAME}/")
        await _open(pilot, remote, "share/")
        await _until(pilot, lambda: "notes.txt" in _names(remote))

        # A name somebody else chose is shown as text, escapes and all.
        assert "[bold]x[/bold]\\x1b.txt" in _names(remote)

        # Mark two, F5: both land in the local folder.
        for name in ("notes.txt", "2024/"):
            remote.table.move_cursor(row=_names(remote).index(name))
            await pilot.press("space")
        await pilot.press("f5")
        await _until(pilot, lambda: (work / "2024" / "a.jpg").exists()
                     and (work / "notes.txt").exists())
        assert (work / "notes.txt").read_bytes() == b"hello"
        assert (work / "2024" / "a.jpg").read_bytes() == b"A" * 3000

        # Again: already there, asked once for the batch.
        remote.table.move_cursor(row=_names(remote).index("notes.txt"))
        remote.table.focus()
        (work / "notes.txt").write_bytes(b"changed here")
        await pilot.press("f5")
        await _until(pilot, lambda: isinstance(app.screen, Choice))
        await pilot.click("#choice-skip")
        await pilot.pause(0.2)
        assert (work / "notes.txt").read_bytes() == b"changed here"

        # Upload from the left into the open folder.
        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
        local.action_refresh()
        await _until(pilot, lambda: "mine.txt" in _names(local))
        local.table.move_cursor(row=_names(local).index("mine.txt"))
        local.table.focus()
        await pilot.press("f5")
        await _until(pilot, lambda: node.stored("share/mine.txt") == b"from here")
        # The node's index changed: the right pane shows it.
        await _until(pilot, lambda: "mine.txt" in _names(remote))

        # A name already there is asked about, never replaced.
        local.table.focus()
        await pilot.press("f5")
        await _until(pilot, lambda: isinstance(app.screen, Choice))
        await pilot.click("#choice-cancel")
        await pilot.pause(0.2)
        assert node.stored("share/mine.txt") == b"from here"


async def test_a_remembered_session_opens_on_the_panes(world, capsys):
    hub, node, _ = world
    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
                                              "--password", PASSPHRASE, "login"]) == 0
    app = _app()
    async with app.run_test(size=(120, 30)) as pilot:
        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
        remote = app.screen.query_one(RemotePane)
        await _until(pilot, lambda: f"Family@{USERNAME}/" in _names(remote))


async def test_a_read_only_folder_takes_no_upload(world):
    hub, node, work = world
    node.roots[0]["writable"] = False
    node.add("share", "notes.txt", b"hello")
    (work / "mine.txt").write_bytes(b"x")
    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
                                              "--password", PASSPHRASE, "login"]) == 0
    app = _app()
    async with app.run_test(size=(120, 30)) as pilot:
        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
        remote = app.screen.query_one(RemotePane)
        local = app.screen.query_one(LocalPane)
        await _open(pilot, remote, f"Family@{USERNAME}/")
        await _open(pilot, remote, "share/")
        await _until(pilot, lambda: "notes.txt" in _names(remote))
        await _until(pilot, lambda: "mine.txt" in _names(local))
        local.table.move_cursor(row=_names(local).index("mine.txt"))
        local.table.focus()
        await pilot.press("f5")
        await pilot.pause(0.3)
        assert node.stored("share/mine.txt") is None
        assert any("does not accept uploads" in n.message for n in app._notifications)


async def test_delete_removes_what_is_yours_and_asks_first(world):
    from fake_hub import USER_ID
    hub, node, work = world
    mine = node.add("share", "mine.txt", b"1", uploader_id=USER_ID)
    theirs = node.add("share", "theirs.txt", b"2", uploader_id="someone-else")
    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
                                              "--password", PASSPHRASE, "login"]) == 0
    app = _app()
    async with app.run_test(size=(120, 30)) as pilot:
        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
        remote = app.screen.query_one(RemotePane)
        await _open(pilot, remote, f"Family@{USERNAME}/")
        await _open(pilot, remote, "share/")
        await _until(pilot, lambda: "theirs.txt" in _names(remote))

        # Somebody else's file, marked with one's own: nothing is deleted.
        for name in ("mine.txt", "theirs.txt"):
            remote.table.move_cursor(row=_names(remote).index(name))
            await pilot.press("space")
        await pilot.press("delete")
        await _until(pilot, lambda: isinstance(app.screen, Choice))
        assert "theirs.txt" in str(app.screen.message)
        await pilot.click("#choice-ok")
        assert mine["id"] in node.files and theirs["id"] in node.files

        # One's own: asked, cancelled, asked again, deleted, and gone from the pane.
        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
        remote.marks.clear()
        remote.redraw()
        remote.table.move_cursor(row=_names(remote).index("mine.txt"))
        await pilot.press("delete")
        await _until(pilot, lambda: isinstance(app.screen, Choice))
        await pilot.click("#choice-cancel")
        assert mine["id"] in node.files
        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
        await pilot.press("delete")
        await _until(pilot, lambda: isinstance(app.screen, Choice))
        await pilot.click("#choice-delete")
        await _until(pilot, lambda: mine["id"] not in node.files)
        await _until(pilot, lambda: "mine.txt" not in _names(remote))
        assert theirs["id"] in node.files

        # On the left, Delete removes nothing of this machine's.
        app.screen.query_one(LocalPane).table.focus()
        await pilot.press("delete")
        await pilot.pause(0.2)
        assert any("select them on the right" in n.message for n in app._notifications)


async def _right_click(pilot, widget, offset=(0, 0)) -> None:
    """Through App.on_event, as the terminal driver sends it; the pilot's own
    clicks skip it."""
    from textual import events
    from textual.pilot import _get_mouse_message_arguments
    args = _get_mouse_message_arguments(widget, offset, button=3)
    for cls in (events.MouseDown, events.MouseUp):
        pilot.app.post_message(cls(**args))
        await pilot.pause()


def _log_text(screen) -> str:
    from textual.widgets import RichLog
    return "\n".join("".join(seg.text for seg in strip)
                     for strip in screen.query_one(RichLog).lines)


async def test_the_interactive_chat_joins_talks_leaves_and_comes_back(world, monkeypatch):
    from meshbay_cli.tui.chat import ChatGroupsScreen, ChatRoomScreen
    from textual.widgets import DataTable, Input
    hub, node, _ = world
    hub.groups.append({"id": "g9", "name": "Quiet", "owner_username": "bob-test",
                       "node_online": True, "is_admin": False})
    quiet = FakeNode()
    quiet.ack["enabled_apps"] = ["files"]
    sessions = {}

    async def open_group(account, group):
        target = quiet if group["id"] == "g9" else node
        sessions[group["id"]] = target.session()
        return sessions[group["id"]]
    monkeypatch.setattr(pool, "open_group", open_group)
    node.say("earlier, from bob")
    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
                                              "--password", PASSPHRASE, "login"]) == 0

    app = MeshBayApp(context.Context(cli.build_parser().parse_args(
        ["--hub", "https://hub.test"])), ChatGroupsScreen)
    async with app.run_test(size=(120, 30)) as pilot:
        await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
        table = app.screen.query_one(DataTable)
        await _until(pilot, lambda: table.row_count == 2)

        # A group whose chat is off is not entered.
        table.move_cursor(row=1)                      # Quiet@bob-test
        await pilot.press("enter")
        await _until(pilot, lambda: any("chat turned off" in n.message
                                        for n in app._notifications))
        assert isinstance(app.screen, ChatGroupsScreen)

        table.move_cursor(row=0)                      # Family@alice-test
        await pilot.press("enter")
        await _until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
        room = app.screen
        await _until(pilot, lambda: "<bob-test> earlier, from bob" in _log_text(room))

        # Posting: sealed, signed, stored on the node, shown here at once.
        room.query_one(Input).value = "hello from the room"
        await pilot.press("enter")
        await _until(pilot, lambda: "<alice-test> hello from the room" in _log_text(room))
        assert len(node.chat) == 2

        # What the node pushes appears as it arrives.
        bob = node.say("live, from bob")
        sessions["g1"].ch._emit({**node.chat[-1], "type": "chat_msg"})
        await _until(pilot, lambda: "<bob-test> live, from bob" in _log_text(room))
        assert bob.pk_ed

        # /part goes back; entering again shows the conversation once, and a
        # pushed message once — the room left behind no handler.
        room.query_one(Input).value = "/part"
        await pilot.press("enter")
        await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
        await pilot.press("enter")
        await _until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
        room = app.screen
        await _until(pilot, lambda: "live, from bob" in _log_text(room))
        node.say("once")
        sessions["g1"].ch._emit({**node.chat[-1], "type": "chat_msg"})
        await _until(pilot, lambda: "<bob-test> once" in _log_text(room))
        await pilot.pause(0.2)
        assert _log_text(room).count("<bob-test> once") == 1

        # Esc goes back too; q quits from the list.
        await pilot.press("escape")
        await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
        await pilot.press("q")
    assert app.return_code in (0, None)


async def test_a_right_click_copies_from_the_chat_and_pastes_into_the_line(world, monkeypatch):
    """The selection, else the line under the pointer; pasted as one line."""
    from meshbay_cli.tui import clipboard
    from meshbay_cli.tui.chat import ChatGroupsScreen, ChatInput, ChatLog, ChatRoomScreen
    hub, node, _ = world
    written, held = [], {"text": None}

    async def write_text(text):
        written.append(text)
        return True

    async def read_text():
        return held["text"]
    monkeypatch.setattr(clipboard, "write_text", write_text)
    monkeypatch.setattr(clipboard, "read_text", read_text)
    node.say("earlier, from bob")
    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
                                              "--password", PASSPHRASE, "login"]) == 0

    app = MeshBayApp(context.Context(cli.build_parser().parse_args(
        ["--hub", "https://hub.test"])), ChatGroupsScreen)
    async with app.run_test(size=(120, 30)) as pilot:
        await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
        await pilot.pause(0.3)
        await pilot.press("enter")
        await _until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
        room = app.screen
        log = room.query_one(ChatLog)
        await _until(pilot, lambda: "<bob-test> earlier, from bob" in _log_text(room))
        y = next(i for i, line in enumerate(log.lines) if "earlier, from bob" in line.text)
        line = log.lines[y].text

        # Nothing selected: the line under the pointer.
        await _right_click(pilot, log, (4, y))
        await _until(pilot, lambda: written)
        assert written[-1] == line.rstrip() and app.clipboard == written[-1]
        assert "<bob-test> earlier, from bob" in written[-1]

        # Selected with the mouse: that text, and only it.
        x = line.index("bob-test")
        await pilot.mouse_down(log, offset=(1 + x, y))
        await pilot.hover(log, offset=(x + len("bob-test"), y))
        await pilot.mouse_up(log, offset=(x + len("bob-test"), y))
        await pilot.pause(0.1)
        assert log.text_selection is not None
        await _right_click(pilot, log, (1, y + 3))
        await _until(pilot, lambda: len(written) == 2)
        assert written[-1] == "bob-test"

        # Pasted into the line written in, at the cursor, lines side by side.
        field = room.query_one(ChatInput)
        field.value = "say: "
        field.cursor_position = len(field.value)
        held["text"] = "first line\nsecond line\n"
        await _right_click(pilot, field, (2, 0))
        await _until(pilot, lambda: field.value != "say: ")
        assert field.value == "say: first line second line"

        # No tool to read the system clipboard: what was copied here.
        held["text"] = None
        field.value = ""
        await _right_click(pilot, field, (2, 0))
        await _until(pilot, lambda: field.value)
        assert field.value == "bob-test"
        assert app.screen is room                     # a right-click left nothing behind
        await pilot.press("escape")