Phillip Tarrant
d829e6553c
Introduces FileCache (OrderedDict LRU, st_mtime_ns validation) to avoid redundant disk reads and duplicate content in conversation context. Read tools return a short "[Cached]" message on cache hit instead of resending unchanged file content, saving tokens. Write/edit/delete tools invalidate affected paths; str_replace pre-warms the cache after edits. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
315 lines
10 KiB
Python
315 lines
10 KiB
Python
"""Tests for the file cache with LRU eviction and mtime invalidation."""
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.models.config import AppConfig, load_config
|
|
from app.models.tool_call import ToolResultStatus
|
|
from app.tools.filesystem import ReadFileTool, ReadManyFilesTool
|
|
from app.utils.file_cache import CacheStats, FileCache, cached_read_file
|
|
from app.utils.file_helpers import BinaryFileError, FileSizeError, PathSecurityError
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# FileCache unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFileCache:
|
|
def test_put_and_get_roundtrip(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
f = tmp_path / "hello.txt"
|
|
f.write_text("hello world")
|
|
|
|
cache.put(f, "hello world")
|
|
assert cache.get(f) == "hello world"
|
|
|
|
def test_get_returns_none_for_missing_key(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
assert cache.get(tmp_path / "nope.txt") is None
|
|
|
|
def test_mtime_change_causes_miss(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
f = tmp_path / "data.txt"
|
|
f.write_text("v1")
|
|
cache.put(f, "v1")
|
|
|
|
# Mutate the file so mtime changes
|
|
time.sleep(0.05) # ensure mtime differs
|
|
f.write_text("v2")
|
|
|
|
assert cache.get(f) is None # stale → miss
|
|
assert cache.stats.invalidations == 1
|
|
|
|
def test_lru_eviction_at_capacity(self, tmp_path: Path) -> None:
|
|
cache = FileCache(max_entries=3)
|
|
files = []
|
|
for i in range(4):
|
|
f = tmp_path / f"f{i}.txt"
|
|
f.write_text(f"content-{i}")
|
|
files.append(f)
|
|
|
|
# Fill cache to capacity
|
|
for f in files[:3]:
|
|
cache.put(f, f.read_text())
|
|
assert len(cache) == 3
|
|
|
|
# Adding a 4th evicts the LRU (files[0])
|
|
cache.put(files[3], files[3].read_text())
|
|
assert len(cache) == 3
|
|
assert cache.get(files[0]) is None # evicted
|
|
assert cache.stats.evictions == 1
|
|
|
|
# files[1..3] still present
|
|
for f in files[1:]:
|
|
assert cache.get(f) is not None
|
|
|
|
def test_invalidate_removes_entry(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
f = tmp_path / "rm.txt"
|
|
f.write_text("bye")
|
|
cache.put(f, "bye")
|
|
assert len(cache) == 1
|
|
|
|
cache.invalidate(f)
|
|
assert len(cache) == 0
|
|
assert cache.get(f) is None
|
|
assert cache.stats.invalidations == 1
|
|
|
|
def test_invalidate_noop_for_missing(self) -> None:
|
|
cache = FileCache()
|
|
cache.invalidate(Path("/nonexistent"))
|
|
assert cache.stats.invalidations == 0
|
|
|
|
def test_clear_empties_cache(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
for i in range(5):
|
|
f = tmp_path / f"c{i}.txt"
|
|
f.write_text(str(i))
|
|
cache.put(f, str(i))
|
|
assert len(cache) == 5
|
|
|
|
cache.clear()
|
|
assert len(cache) == 0
|
|
|
|
def test_stats_accuracy(self, tmp_path: Path) -> None:
|
|
cache = FileCache(max_entries=2)
|
|
a = tmp_path / "a.txt"
|
|
b = tmp_path / "b.txt"
|
|
c = tmp_path / "c.txt"
|
|
a.write_text("a")
|
|
b.write_text("b")
|
|
c.write_text("c")
|
|
|
|
# Miss
|
|
cache.get(a)
|
|
assert cache.stats.misses == 1
|
|
assert cache.stats.hits == 0
|
|
|
|
# Put + hit
|
|
cache.put(a, "a")
|
|
cache.get(a)
|
|
assert cache.stats.hits == 1
|
|
|
|
# Fill + evict
|
|
cache.put(b, "b")
|
|
cache.put(c, "c") # evicts a
|
|
assert cache.stats.evictions == 1
|
|
|
|
def test_hit_rate(self) -> None:
|
|
stats = CacheStats(hits=3, misses=1)
|
|
assert stats.hit_rate == pytest.approx(0.75)
|
|
|
|
def test_hit_rate_zero_total(self) -> None:
|
|
stats = CacheStats()
|
|
assert stats.hit_rate == 0.0
|
|
|
|
def test_file_deleted_after_caching(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
f = tmp_path / "gone.txt"
|
|
f.write_text("here")
|
|
cache.put(f, "here")
|
|
|
|
f.unlink()
|
|
assert cache.get(f) is None # stat fails → miss
|
|
|
|
def test_put_skips_when_stat_fails(self) -> None:
|
|
cache = FileCache()
|
|
cache.put(Path("/totally/nonexistent"), "data")
|
|
assert len(cache) == 0
|
|
|
|
def test_get_moves_to_end(self, tmp_path: Path) -> None:
|
|
"""Accessing an entry makes it most-recently-used, protecting from eviction."""
|
|
cache = FileCache(max_entries=3)
|
|
files = []
|
|
for i in range(3):
|
|
f = tmp_path / f"lru{i}.txt"
|
|
f.write_text(f"c{i}")
|
|
files.append(f)
|
|
cache.put(f, f"c{i}")
|
|
|
|
# Touch files[0] to make it MRU
|
|
cache.get(files[0])
|
|
|
|
# Add a new entry — files[1] (LRU) should be evicted, not files[0]
|
|
extra = tmp_path / "extra.txt"
|
|
extra.write_text("x")
|
|
cache.put(extra, "x")
|
|
|
|
assert cache.get(files[0]) is not None # protected by access
|
|
assert cache.get(files[1]) is None # evicted
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cached_read_file tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCachedReadFile:
|
|
def test_without_cache_matches_safe_read(self, tmp_path: Path) -> None:
|
|
f = tmp_path / "plain.txt"
|
|
f.write_text("hello")
|
|
content = cached_read_file(f, tmp_path, cache=None)
|
|
assert content == "hello"
|
|
|
|
def test_populates_on_miss_returns_on_hit(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
f = tmp_path / "cached.txt"
|
|
f.write_text("data")
|
|
|
|
# First call: miss → read from disk → populate cache
|
|
content1 = cached_read_file(f, tmp_path, cache=cache)
|
|
assert content1 == "data"
|
|
assert cache.stats.misses == 1
|
|
assert cache.stats.hits == 0
|
|
|
|
# Second call: hit → from cache
|
|
content2 = cached_read_file(f, tmp_path, cache=cache)
|
|
assert content2 == "data"
|
|
assert cache.stats.hits == 1
|
|
|
|
def test_security_checks_run_on_cached_path(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
with pytest.raises(PathSecurityError):
|
|
cached_read_file("/etc/passwd", tmp_path, cache=cache)
|
|
|
|
def test_binary_check_runs_on_cached_path(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
f = tmp_path / "bin.dat"
|
|
f.write_bytes(b"\x00binary\x00")
|
|
with pytest.raises(BinaryFileError):
|
|
cached_read_file(f, tmp_path, cache=cache)
|
|
|
|
def test_size_check_runs_on_cached_path(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
f = tmp_path / "big.txt"
|
|
f.write_text("x" * 200)
|
|
|
|
# First read populates cache
|
|
cached_read_file(f, tmp_path, max_size_bytes=1000, cache=cache)
|
|
|
|
# Now make file too big on disk — security check should catch it
|
|
# even though content is cached
|
|
f.write_text("x" * 2000)
|
|
with pytest.raises(FileSizeError):
|
|
cached_read_file(f, tmp_path, max_size_bytes=1000, cache=cache)
|
|
|
|
def test_file_not_found(self, tmp_path: Path) -> None:
|
|
cache = FileCache()
|
|
with pytest.raises(FileNotFoundError):
|
|
cached_read_file(tmp_path / "nope.txt", tmp_path, cache=cache)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool-level cache-hit dedup tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def config() -> AppConfig:
|
|
return load_config()
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_workspace(tmp_path: Path, config: AppConfig) -> tuple[Path, AppConfig]:
|
|
config.agent.workspace_root = tmp_path
|
|
return tmp_path, config
|
|
|
|
|
|
class TestReadFileToolCacheHit:
|
|
def test_first_read_returns_full_content(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
|
ws, cfg = tmp_workspace
|
|
cache = FileCache()
|
|
(ws / "hello.txt").write_text("hello world")
|
|
|
|
tool = ReadFileTool(ws, cfg, file_cache=cache)
|
|
result = tool.run("tc-1", {"file_path": "hello.txt"})
|
|
assert result.status == ToolResultStatus.SUCCESS
|
|
assert result.output == "hello world"
|
|
|
|
def test_second_read_returns_cached_message(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
|
ws, cfg = tmp_workspace
|
|
cache = FileCache()
|
|
(ws / "hello.txt").write_text("hello world")
|
|
|
|
tool = ReadFileTool(ws, cfg, file_cache=cache)
|
|
tool.run("tc-1", {"file_path": "hello.txt"})
|
|
|
|
result2 = tool.run("tc-2", {"file_path": "hello.txt"})
|
|
assert result2.status == ToolResultStatus.SUCCESS
|
|
assert "[Cached]" in result2.output
|
|
assert "hello.txt" in result2.output
|
|
assert "hello world" not in result2.output
|
|
|
|
def test_changed_file_returns_full_content_again(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
|
ws, cfg = tmp_workspace
|
|
cache = FileCache()
|
|
f = ws / "data.txt"
|
|
f.write_text("v1")
|
|
|
|
tool = ReadFileTool(ws, cfg, file_cache=cache)
|
|
tool.run("tc-1", {"file_path": "data.txt"})
|
|
|
|
# Mutate file so mtime changes
|
|
time.sleep(0.05)
|
|
f.write_text("v2")
|
|
|
|
result2 = tool.run("tc-2", {"file_path": "data.txt"})
|
|
assert result2.status == ToolResultStatus.SUCCESS
|
|
assert result2.output == "v2"
|
|
assert "[Cached]" not in result2.output
|
|
|
|
def test_no_cache_always_returns_content(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
|
ws, cfg = tmp_workspace
|
|
(ws / "hello.txt").write_text("hello")
|
|
|
|
tool = ReadFileTool(ws, cfg, file_cache=None)
|
|
r1 = tool.run("tc-1", {"file_path": "hello.txt"})
|
|
r2 = tool.run("tc-2", {"file_path": "hello.txt"})
|
|
assert r1.output == "hello"
|
|
assert r2.output == "hello"
|
|
|
|
|
|
class TestReadManyFilesToolCacheHit:
|
|
def test_cached_files_get_short_message(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
|
ws, cfg = tmp_workspace
|
|
cache = FileCache()
|
|
(ws / "a.txt").write_text("alpha")
|
|
(ws / "b.txt").write_text("bravo")
|
|
|
|
tool = ReadManyFilesTool(ws, cfg, file_cache=cache)
|
|
|
|
# First read — full content
|
|
r1 = tool.run("tc-1", {"file_paths": ["a.txt", "b.txt"]})
|
|
assert "alpha" in r1.output
|
|
assert "bravo" in r1.output
|
|
|
|
# Second read — cached messages
|
|
r2 = tool.run("tc-2", {"file_paths": ["a.txt", "b.txt"]})
|
|
assert "[Cached]" in r2.output
|
|
assert "alpha" not in r2.output
|
|
assert "bravo" not in r2.output
|