LLeMbas/tests/test_agent_index.py

"""Listing a project directory: the ladder, the cache, and the budget.

The ladder is exercised against a fake executor rather than a real host,
because what is being tested is *which rung is chosen* and what happens when
one falls through -- and a real box either has git or does not, which would
make the interesting cases unreachable.

`scan_dir` itself is tested against a real SFTP server in test_agent_browse.py.
"""

from __future__ import annotations

import time

import pytest

from lembas.services.agent import index as index_service
from lembas.services.agent.base import ExecError, ExecResult, RemoteEntry


class _Fake:
    """An executor that answers whatever the test says, and records the asking."""

    def __init__(self, *, answers: dict[str, ExecResult] | None = None, tree=None):
        self.answers = answers or {}
        self.tree = tree or {}
        self.commands: list[str] = []
        self.scans: list[str] = []

    async def run(self, request) -> ExecResult:
        self.commands.append(request.command)
        for fragment, result in self.answers.items():
            if fragment in request.command:
                return result
        return ExecResult(exit_status=1, output="")

    async def scan_dir(self, path: str) -> list[RemoteEntry]:
        self.scans.append(path)
        if path not in self.tree:
            raise ExecError(f"There is no directory at {path}.")
        return self.tree[path]


def _ok(output: str) -> ExecResult:
    return ExecResult(exit_status=0, output=output)


@pytest.fixture(autouse=True)
def _clean_cache():
    index_service.clear()
    yield
    index_service.clear()


# --- The ladder --------------------------------------------------------------
async def test_git_is_tried_first_and_wins_where_it_works():
    """`--exclude-standard` is the whole reason. A repository already carries
    somebody's considered list of what is not part of the project, and
    reproducing it by hand is how an index ends up mostly build output."""
    executor = _Fake(answers={"git ls-files": _ok("README.md\nsrc/main.py\n")})

    found = await index_service.build(executor, "/work")

    assert found.source == "git"
    assert found.paths == ("README.md", "src/main.py")
    assert executor.commands == [executor.commands[0]]
    assert "--exclude-standard" in executor.commands[0]


async def test_find_takes_over_when_the_directory_is_not_a_repository():
    executor = _Fake(answers={"find .": _ok("./README.md\n./src/main.py\n")})

    found = await index_service.build(executor, "/work")

    assert found.source == "find"
    assert found.paths == ("README.md", "src/main.py")
    assert len(executor.commands) == 2  # git was tried, and fell through


async def test_find_prunes_the_usual_noise():
    """Not applied to the git rung, which has already applied the repository's
    own rules -- a checked-in `vendor/` is checked in on purpose."""
    executor = _Fake(answers={"find .": _ok("./a\n")})

    await index_service.build(executor, "/work")

    assert "node_modules" in executor.commands[1]
    assert "'.git'" in executor.commands[1]


async def test_sftp_is_the_last_resort_and_always_works():
    executor = _Fake(
        tree={
            "/work": [RemoteEntry("src", True), RemoteEntry("README.md", False)],
            "src": [RemoteEntry("main.py", False)],
        }
    )

    found = await index_service.build(executor, "/work")

    assert found.source == "sftp"
    assert "README.md" in found.paths
    assert "src/main.py" in found.paths


async def test_a_directory_that_cannot_be_read_at_all_is_empty_not_an_error():
    """A reply must not fail because a listing did. The fragment carrying it
    disappears instead, which is what `requires` is for."""
    executor = _Fake(tree={})

    found = await index_service.build(executor, "/work")

    assert found.paths == ()
    assert not found.ok


async def test_control_characters_are_stripped_from_a_filename():
    """These end up inside a system prompt. An escape sequence in a filename
    could otherwise repaint the transcript it is quoted in."""
    executor = _Fake(answers={"git ls-files": _ok("ok.py\nevil\x1b[31m.py\n")})

    found = await index_service.build(executor, "/work")

    assert "evil[31m.py" in found.paths
    assert "\x1b" not in "".join(found.paths)


async def test_the_number_of_paths_is_capped_and_says_so():
    """"There is nothing else here" and "I stopped looking" are different
    answers, and it must not give the first for the second."""
    many = "\n".join(f"f{i}.py" for i in range(index_service.MAX_ENTRIES + 50))
    executor = _Fake(answers={"git ls-files": _ok(many)})

    found = await index_service.build(executor, "/work")

    assert len(found.paths) == index_service.MAX_ENTRIES
    assert found.truncated


# --- The cache ---------------------------------------------------------------
async def test_a_second_ask_does_not_walk_again():
    executor = _Fake(answers={"git ls-files": _ok("a.py\n")})

    await index_service.ensure(executor, "profile-1", "/work")
    await index_service.ensure(executor, "profile-1", "/work")

    assert len(executor.commands) == 1


async def test_refreshing_walks_again():
    executor = _Fake(answers={"git ls-files": _ok("a.py\n")})

    await index_service.ensure(executor, "profile-1", "/work")
    await index_service.ensure(executor, "profile-1", "/work", refresh=True)

    assert len(executor.commands) == 2


async def test_a_stale_index_is_not_returned():
    executor = _Fake(answers={"git ls-files": _ok("a.py\n")})
    await index_service.ensure(executor, "profile-1", "/work")

    index_service._CACHE[("profile-1", "/work")] = index_service.ProjectIndex(
        paths=("a.py",), total=1, source="git", built_at=time.monotonic() - index_service.TTL - 1
    )

    assert index_service.cached("profile-1", "/work") is None


async def test_forgetting_a_connection_drops_its_listings():
    """A connection somebody has just revoked must not leave a listing of the
    machine behind it. Called from the same three places that close its
    terminals."""
    executor = _Fake(answers={"git ls-files": _ok("a.py\n")})
    await index_service.ensure(executor, "profile-1", "/work")
    await index_service.ensure(executor, "profile-2", "/work")

    assert index_service.forget("profile-1") == 1
    assert index_service.cached("profile-1", "/work") is None
    assert index_service.cached("profile-2", "/work") is not None


def test_reading_the_cache_never_does_work():
    """`harness` calls this synchronously while assembling the system message,
    so it must never be the thing that opens a connection."""
    assert index_service.cached("nobody", "/nowhere") is None


# --- The budget --------------------------------------------------------------
def _index(paths) -> index_service.ProjectIndex:
    return index_service.ProjectIndex(
        paths=tuple(sorted(paths)), total=len(paths), source="git", built_at=time.monotonic()
    )


def test_a_small_tree_is_shown_whole():
    text = index_service.render(_index(["README.md", "src/main.py"]), 2000)

    assert "README.md" in text
    assert "main.py" in text
    assert "files)" not in text


def test_a_big_directory_becomes_a_count():
    paths = ["README.md"] + [f"vendor/f{i}.js" for i in range(400)]

    text = index_service.render(_index(paths), 300)

    assert "vendor/ (400 files)" in text
    assert "README.md" in text
    assert "file_list" in text


def test_the_deepest_big_directory_is_collapsed_first():
    """Collapsing by size alone takes `src/` before `src/.../vendor/` -- it is
    bigger because it *contains* it -- and loses every name worth having in
    order to fold away one directory of third-party files."""
    paths = [f"src/api/{n}.py" for n in "abcde"] + [
        f"src/web/vendor/f{i}.js" for i in range(300)
    ]

    text = index_service.render(_index(paths), 400)

    assert "vendor/ (300 files)" in text
    assert "a.py" in text  # src/api survived


def test_a_flat_directory_of_thousands_is_still_bounded():
    """The one shape collapsing cannot help with: no directory to fold them
    into, so the running budget has to bite instead."""
    text = index_service.render(_index([f"dump{i:05d}.log" for i in range(5000)]), 400)

    assert len(text) < 1200
    assert "more files" in text


def test_a_budget_of_zero_renders_nothing():
    """Which is how "index it for the picker, but say nothing to the model" is
    expressed. The fragment vanishes rather than appearing empty."""
    assert index_service.render(_index(["a.py"]), 0) == ""


def test_an_empty_index_renders_nothing():
    assert index_service.render(index_service.ProjectIndex(), 2000) == ""


def test_a_truncated_index_says_it_is_a_sample():
    sample = index_service.ProjectIndex(
        paths=("a.py", "b.py"), total=9000, truncated=True, source="find", built_at=time.monotonic()
    )

    assert "sample" in index_service.render(sample, 2000)