LLeMbas/tests/test_agent_browse.py

"""Walking a directory on the far side, to choose where a chat works.

Everything here goes through a real SFTP server on 127.0.0.1. A stub would
prove nothing worth proving: the whole point of `scan_dir` is that it reads
`SFTPName.attrs` to tell a directory from a file, and a stub that returns
whatever shape the test wants cannot be wrong about that.
"""

from __future__ import annotations

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from lembas.db.models import SshProfile, User
from lembas.services.agent import ssh as ssh_service

# 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 tree(tmp_path):
    """A real sshd with SFTP, over a directory laid out to be walked."""
    root = tmp_path / "project"
    (root / "src" / "deep").mkdir(parents=True)
    (root / "docs").mkdir()
    (root / "README.md").write_text("hello")
    (root / "src" / "main.py").write_text("print()")
    (root / ".hidden").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, "fingerprint": fingerprint, "root": str(root)}
    finally:
        server.close()
        await server.wait_closed()


def _spec(tree: dict) -> dict:
    return {
        "host": "127.0.0.1",
        "port": tree["port"],
        "username": "tester",
        "auth": "key",
        "password": "",
        "private_key": "",
        "key_passphrase": "",
        "host_key": tree["host_key"],
        "connect_timeout": 15,
    }


# --- scan_dir ----------------------------------------------------------------
async def test_a_listing_says_which_rows_are_directories(tree):
    """The whole reason this exists beside `list_dir`, which returns names."""
    executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
    entries = await executor.scan_dir("")

    kinds = {entry.name: entry.is_dir for entry in entries}
    assert kinds["src"] is True
    assert kinds["docs"] is True
    assert kinds["README.md"] is False


async def test_directories_sort_before_files(tree):
    """The order somebody navigating expects: the things you can walk into."""
    executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
    names = [entry.name for entry in await executor.scan_dir("")]

    assert names.index("src") < names.index("README.md")
    assert names.index("docs") < names.index("README.md")


async def test_files_carry_a_size(tree):
    executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
    entries = {entry.name: entry for entry in await executor.scan_dir("")}

    assert entries["README.md"].size == len("hello")


async def test_dotfiles_are_listed(tree):
    """Shown, not filtered. A picker that hides `.config` is one somebody has
    to work around, and the account on the far side is the boundary here --
    not a taste for tidy listings."""
    executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
    names = [entry.name for entry in await executor.scan_dir("")]

    assert ".hidden" in names
    assert "." not in names and ".." not in names


async def test_a_relative_path_is_measured_from_the_project_directory(tree):
    executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
    names = [entry.name for entry in await executor.scan_dir("src")]

    assert names == ["deep", "main.py"]


async def test_an_absolute_path_walks_out_of_the_project_directory(tree):
    """Deliberate, and matching `_resolve`'s own docstring.

    Containment is the far-side account's job, not this layer's: `shell_run`
    could leave the project directory in one line, so a picker that refused to
    would be a comfort rather than a control. The browser opens *at* the
    project directory; it does not fence it.
    """
    executor = ssh_service.SshExecutor(_spec(tree), tree["root"])
    names = [entry.name for entry in await executor.scan_dir("/")]

    assert names  # the real filesystem root, listed without complaint


async def test_a_missing_directory_says_so_in_words(tree):
    executor = ssh_service.SshExecutor(_spec(tree), tree["root"])

    with pytest.raises(ssh_service.ExecError) as caught:
        await executor.scan_dir("nowhere")

    assert "no directory" in caught.value.message.lower()


# --- The route ---------------------------------------------------------------
@pytest.fixture
def served_tree(tmp_path):
    """The same tree, but with the server on a thread and a loop of its own.

    Its own loop matters: the tests below drive the app through the synchronous
    TestClient, so a server sharing the test's loop could never accept the
    connection the route is trying to make -- `client.get(...)` is still
    blocking it. The async fixture above is fine for calling `scan_dir`
    directly, and useless here.
    """
    import asyncio
    import threading

    root = tmp_path / "served"
    (root / "src").mkdir(parents=True)
    (root / "README.md").write_text("hello")

    loop = asyncio.new_event_loop()
    thread = threading.Thread(target=loop.run_forever, daemon=True)
    thread.start()

    async def start():
        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)
        return server, port, line, fingerprint

    server, port, line, fingerprint = asyncio.run_coroutine_threadsafe(start(), loop).result(10)
    try:
        yield {"port": port, "host_key": line, "fingerprint": fingerprint, "root": str(root)}
    finally:

        async def stop():
            server.close()
            await server.wait_closed()

        asyncio.run_coroutine_threadsafe(stop(), loop).result(10)
        loop.call_soon_threadsafe(loop.stop)
        thread.join(timeout=5)


def _profile(db, tree: dict, *, host_key: str | None = None) -> SshProfile:
    user = db.scalars(select(User)).first()
    profile = SshProfile(
        owner_id=user.id,
        name="Test box",
        host="127.0.0.1",
        port=tree["port"],
        username="tester",
        host_key=tree["host_key"] if host_key is None else host_key,
        host_fingerprint=tree["fingerprint"],
        default_dir=tree["root"],
    )
    db.add(profile)
    db.commit()
    return profile


def test_browsing_lists_the_project_directory(client: TestClient, db, registered, served_tree):
    profile = _profile(db, served_tree)

    body = client.get(f"/api/agents/{profile.id}/browse").text

    assert "src" in body
    assert "README.md" in body
    assert served_tree["root"] in body


def test_browsing_offers_a_way_back_up(client: TestClient, db, registered, served_tree):
    profile = _profile(db, served_tree)

    body = client.get(f"/api/agents/{profile.id}/browse", params={"path": served_tree["root"]}).text

    assert "Up a level" in body


def test_a_host_key_that_was_never_confirmed_is_refused_in_words(
    client: TestClient, db, registered, served_tree
):
    """Rather than the known_hosts prose `connect_kwargs` would raise, which is
    accurate and means nothing to somebody looking at a directory picker."""
    profile = _profile(db, served_tree, host_key="")

    body = client.get(f"/api/agents/{profile.id}/browse").text

    assert "has not been confirmed" in body


def test_somebody_elses_connection_is_not_browsable(
    client: TestClient, db, registered, served_tree
):
    """These are credentials to somebody's machine, so ownership is the whole
    authorisation -- `sharing.py` grants reading, and a host you can read is a
    host you can log in to.

    The permission is granted to everybody first, so what is being tested is
    ownership and not the `agent.ssh` gate in front of it. Without that the
    second account is refused before the question is even asked, and the test
    would pass whether or not the ownership check existed.
    """
    from lembas.services import settings_store

    profile = _profile(db, served_tree)
    settings_store.update(db, {"default_permissions": {"agent.ssh": True}})

    client.post("/auth/logout", follow_redirects=False)
    client.post(
        "/auth/register",
        data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
        follow_redirects=False,
    )

    # 404 and not 403: whether that connection exists is not this endpoint's to
    # reveal to somebody who does not own it.
    assert client.get(f"/api/agents/{profile.id}/browse").status_code == 404


# --- Reading the project directory again -------------------------------------
def _agent_chat(db, profile):
    from lembas.db.models import KIND_AGENT, Chat, Connection, Model

    connection = Connection(name="c", base_url="http://127.0.0.1:1", api_key_encrypted="")
    db.add(connection)
    db.commit()
    db.add(Model(connection_id=connection.id, model_id="m", capabilities_json={"tools": True}))
    db.commit()

    chat = Chat(
        user_id=profile.owner_id,
        model_id="m",
        connection_id=connection.id,
        kind=KIND_AGENT,
        ssh_profile_id=profile.id,
        project_dir=profile.default_dir,
    )
    db.add(chat)
    db.commit()
    return chat


def test_reindexing_walks_the_tree_again(client: TestClient, db, registered, served_tree):
    """The listing is built only when a reply starts and then held for five
    minutes, so anything done in the terminal panel is invisible to it until
    then. This is the way to say "look again"."""
    from lembas.services.agent import index as index_service

    profile = _profile(db, served_tree)
    chat = _agent_chat(db, profile)

    response = client.post(f"/api/chats/{chat.id}/index")
    assert response.status_code == 200, response.text
    body = response.json()

    assert body["ok"] is True
    assert body["files"] >= 2  # README.md and src/
    assert index_service.cached(profile.id, served_tree["root"]) is not None


def test_reindexing_a_plain_chat_says_there_is_nothing_to_read(
    client: TestClient, db, registered, served_tree
):
    from lembas.db.models import Chat, Connection, Model

    connection = Connection(name="c", base_url="http://127.0.0.1:1", api_key_encrypted="")
    db.add(connection)
    db.commit()
    db.add(Model(connection_id=connection.id, model_id="m"))
    db.commit()
    chat = Chat(user_id=_profile(db, served_tree).owner_id, model_id="m",
                connection_id=connection.id)
    db.add(chat)
    db.commit()

    assert client.post(f"/api/chats/{chat.id}/index").status_code == 409


def test_reindexing_somebody_elses_chat_is_not_possible(
    client: TestClient, db, registered, served_tree
):
    profile = _profile(db, served_tree)
    chat = _agent_chat(db, profile)

    client.post("/auth/logout", follow_redirects=False)
    client.post(
        "/auth/register",
        data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
        follow_redirects=False,
    )

    assert client.post(f"/api/chats/{chat.id}/index").status_code == 404


# --- Picking a file rather than a directory ---------------------------------------
def test_files_are_inert_when_a_directory_is_wanted(
    client: TestClient, db, registered, served_tree
):
    """The default, and the older of the two. Hiding files would make a folder
    of nothing but files look empty, which is worse than showing what is there
    and not letting it be chosen."""
    profile = _profile(db, served_tree)

    body = client.get(
        f"/api/agents/{profile.id}/browse", params={"path": served_tree["root"]}
    ).text

    assert "README.md" in body
    assert "data-file-open" not in body


def test_files_become_choices_when_a_file_is_wanted(
    client: TestClient, db, registered, served_tree
):
    """Canvas asks for `pick=file`. One listing serves both, because a second
    copy is a second place for the path arithmetic to be got subtly
    differently -- and getting it differently means a file that opens to the
    wrong path, or to nothing."""
    profile = _profile(db, served_tree)

    body = client.get(
        f"/api/agents/{profile.id}/browse",
        params={"path": served_tree["root"], "pick": "file"},
    ).text

    assert "data-file-open" in body
    assert f'data-file-open="{served_tree["root"]}/README.md"' in body
    # Directories stay a step rather than becoming a choice.
    assert "data-dir-open" in body


def test_an_unknown_pick_falls_back_to_directories(
    client: TestClient, db, registered, served_tree
):
    """It arrives off a query string, so it is read as one of two things rather
    than trusted -- the same shape every other value read off a request here
    takes."""
    profile = _profile(db, served_tree)

    body = client.get(
        f"/api/agents/{profile.id}/browse",
        params={"path": served_tree["root"], "pick": "whatever"},
    ).text

    assert "data-file-open" not in body