LLeMbas/tests/test_canvas_ssh.py

"""Reading and writing a file for somebody who is about to edit it.

The model-facing `read_file`/`write_file` pair is deliberately untouched: what
they return is a contract a model has been shown, and it is the right contract
for a model. It is the wrong one here, and these are the cases that say why.
"""

from __future__ import annotations

import pytest

from lembas.services.agent import ssh as ssh_service
from lembas.services.agent.base import Conflict, ExecError

# Stands up something real -- see the `slow` marker in pyproject.toml.
pytestmark = pytest.mark.slow

asyncssh = pytest.importorskip("asyncssh")


class _Server(asyncssh.SSHServer):
    def begin_auth(self, username: str) -> bool:
        return False


@pytest.fixture
async def machine(tmp_path):
    project = tmp_path / "project"
    project.mkdir()
    server = await asyncssh.create_server(
        _Server,
        "127.0.0.1",
        0,
        server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
        sftp_factory=True,
    )
    port = next(iter(server.sockets)).getsockname()[1]
    line, fingerprint = await ssh_service.capture_host_key("127.0.0.1", port)
    try:
        yield {"port": port, "host_key": line, "dir": str(project), "path": project}
    finally:
        server.close()
        await server.wait_closed()


def _executor(machine) -> ssh_service.SshExecutor:
    return ssh_service.SshExecutor(
        {
            "host": "127.0.0.1",
            "port": machine["port"],
            "username": "tester",
            "auth": "password",
            "credential": "",
            "host_key": machine["host_key"],
        },
        machine["dir"],
    )


# --- Fidelity ------------------------------------------------------------------
async def test_an_escape_sequence_survives_a_round_trip(machine):
    """The whole reason this is not `read_file`. That one ends in
    `clean_output`, which strips ANSI escapes -- right for the output of a
    command, and here it means opening a file and pressing Save rewrites it
    with the escapes gone."""
    original = "red \x1b[31mtext\x1b[0m here\n"
    (machine["path"] / "colours.txt").write_text(original)
    executor = _executor(machine)

    opened = await executor.read_text("colours.txt")
    assert opened.text == original

    await executor.write_text("colours.txt", opened.text, if_unchanged=opened.revision)
    assert (machine["path"] / "colours.txt").read_text() == original


async def test_the_model_facing_read_still_strips_them(machine):
    """Pinned as a pair: the contract a model was shown has not moved."""
    (machine["path"] / "colours.txt").write_text("red \x1b[31mtext\x1b[0m here\n")
    text = await _executor(machine).read_file("colours.txt")
    assert "\x1b[31m" not in text


async def test_undecodable_bytes_are_reported_rather_than_replaced(machine):
    """errors="replace" would hand back U+FFFD for every one of them, and
    saving that back is how a file is quietly destroyed."""
    (machine["path"] / "blob.bin").write_bytes(b"\xff\xfe\x00\x01binary")
    opened = await _executor(machine).read_text("blob.bin")
    assert opened.binary is True
    assert opened.text == ""


async def test_a_nul_byte_early_on_reads_as_binary(machine):
    (machine["path"] / "blob.bin").write_bytes(b"text\x00more text")
    assert (await _executor(machine).read_text("blob.bin")).binary is True


async def test_utf8_beyond_ascii_is_not_binary(machine):
    (machine["path"] / "note.txt").write_text("a mallorn tree — Lothlórien\n")
    opened = await _executor(machine).read_text("note.txt")
    assert opened.binary is False
    assert "Lothlórien" in opened.text


# --- Size ------------------------------------------------------------------------
async def test_a_large_file_opens_truncated(machine):
    (machine["path"] / "big.log").write_text("x" * (ssh_service.MAX_READ_BYTES + 500))
    opened = await _executor(machine).read_text("big.log")
    assert opened.truncated is True
    assert len(opened.text) == ssh_service.MAX_READ_BYTES


async def test_an_oversize_write_is_refused_not_truncated(machine):
    """`write_file` truncates because a model is told how many bytes it wrote.
    Somebody pressing Save would lose the tail with nothing said."""
    executor = _executor(machine)
    (machine["path"] / "big.txt").write_text("small")

    with pytest.raises(ExecError, match="Nothing was written"):
        await executor.write_text("big.txt", "y" * (ssh_service.MAX_WRITE_BYTES + 1))

    assert (machine["path"] / "big.txt").read_text() == "small"


# --- Conflict ---------------------------------------------------------------------
async def test_a_file_that_moved_underneath_refuses_the_save(machine):
    import os

    target = machine["path"] / "note.txt"
    target.write_text("alpha\n")
    executor = _executor(machine)
    opened = await executor.read_text("note.txt")

    # Somebody else's editor, a build, a checkout. The size differs, so this
    # does not depend on the filesystem's mtime resolution.
    target.write_text("something else entirely\n")
    os.utime(target, (0, 0))

    with pytest.raises(Conflict):
        await executor.write_text("note.txt", "beta\n", if_unchanged=opened.revision)

    assert target.read_text() == "something else entirely\n"


async def test_a_save_with_no_token_overwrites(machine):
    """Which is what Overwrite on the conflict card does."""
    target = machine["path"] / "note.txt"
    target.write_text("alpha\n")
    await _executor(machine).write_text("note.txt", "beta\n")
    assert target.read_text() == "beta\n"


async def test_a_new_file_can_be_created(machine):
    """Open a path that is not there, type, Save. The stat finds nothing and
    there is nothing for the token to disagree with."""
    executor = _executor(machine)
    await executor.write_text("fresh.txt", "hello\n", if_unchanged="0:0")
    assert (machine["path"] / "fresh.txt").read_text() == "hello\n"


async def test_the_revision_moves_after_a_write(machine):
    """Or the second save from the same tab would always conflict."""
    target = machine["path"] / "note.txt"
    target.write_text("alpha\n")
    executor = _executor(machine)

    opened = await executor.read_text("note.txt")
    written = await executor.write_text(
        "note.txt", "much longer contents\n", if_unchanged=opened.revision
    )
    assert written.revision != opened.revision

    await executor.write_text("note.txt", "again\n", if_unchanged=written.revision)
    assert target.read_text() == "again\n"


async def test_reading_something_that_is_not_there_says_so(machine):
    with pytest.raises(ExecError, match="no file"):
        await _executor(machine).read_text("nowhere.txt")