wehub-resource-sync
c3bf08ac8d
K8s Workspace Integration Tests / k8s-workspace-tests (push) Waiting to run
Pre-commit / run (ubuntu-latest) (push) Waiting to run
Python Unittest Coverage / test (macos-15, 3.11) (push) Waiting to run
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Waiting to run
Python Unittest Coverage / test (windows-latest, 3.11) (push) Waiting to run
Web UI / check (push) Waiting to run
433 lines
15 KiB
Python
433 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""File cache test case for Read/Write/Edit tools."""
|
|
import os
|
|
import tempfile
|
|
from unittest.async_case import IsolatedAsyncioTestCase
|
|
|
|
from agentscope.state import AgentState
|
|
from agentscope.tool import Read, Write, Edit
|
|
|
|
|
|
class FileCacheTest(IsolatedAsyncioTestCase):
|
|
"""Test file cache functionality for Read/Write/Edit tools."""
|
|
|
|
async def asyncSetUp(self) -> None:
|
|
"""The async setup method."""
|
|
self.read_tool = Read()
|
|
self.write_tool = Write()
|
|
self.edit_tool = Edit()
|
|
self.state = AgentState()
|
|
|
|
# Create a temporary directory
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.test_file = os.path.join(self.temp_dir, "test.txt")
|
|
|
|
async def asyncTearDown(self) -> None:
|
|
"""Clean up temporary files."""
|
|
import shutil
|
|
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
async def test_edit_without_read(self) -> None:
|
|
"""Test Edit fails when file not read first."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Hello World\n")
|
|
|
|
# Try to edit without reading first
|
|
chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="Hello",
|
|
new_string="Hi",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Should fail with error
|
|
self.assertEqual(chunk.state, "error")
|
|
self.assertIn("must first read", chunk.content[0].text)
|
|
|
|
async def test_write_without_read(self) -> None:
|
|
"""Test Write fails when existing file not read first."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Existing content\n")
|
|
|
|
# Try to write without reading first
|
|
chunk = await self.write_tool(
|
|
file_path=self.test_file,
|
|
content="New content\n",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Should fail with error
|
|
self.assertEqual(chunk.state, "error")
|
|
self.assertIn("has not been read yet", chunk.content[0].text)
|
|
|
|
async def test_write_new_file_without_read(self) -> None:
|
|
"""Test Write succeeds for new file without reading."""
|
|
new_file = os.path.join(self.temp_dir, "new_file.txt")
|
|
|
|
# Write to a new file (doesn't exist yet)
|
|
chunk = await self.write_tool(
|
|
file_path=new_file,
|
|
content="New file content\n",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Should succeed
|
|
self.assertEqual(chunk.state, "running")
|
|
self.assertTrue(os.path.exists(new_file))
|
|
|
|
with open(new_file, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.assertEqual(content, "New file content\n")
|
|
|
|
async def test_edit_after_read(self) -> None:
|
|
"""Test Edit succeeds after reading file."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Hello World\n")
|
|
|
|
# Read the file first
|
|
read_chunk = await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(read_chunk.state, "running")
|
|
|
|
# Now edit should succeed
|
|
edit_chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="Hello",
|
|
new_string="Hi",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
self.assertEqual(edit_chunk.state, "running")
|
|
|
|
# Verify file was edited
|
|
with open(self.test_file, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.assertEqual(content, "Hi World\n")
|
|
|
|
async def test_write_after_read(self) -> None:
|
|
"""Test Write succeeds after reading file."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Old content\n")
|
|
|
|
# Read the file first
|
|
read_chunk = await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(read_chunk.state, "running")
|
|
|
|
# Now write should succeed
|
|
write_chunk = await self.write_tool(
|
|
file_path=self.test_file,
|
|
content="New content\n",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
self.assertEqual(write_chunk.state, "running")
|
|
|
|
# Verify file was overwritten
|
|
with open(self.test_file, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.assertEqual(content, "New content\n")
|
|
|
|
async def test_cache_invalidation_after_file_deletion(self) -> None:
|
|
"""Test cache handles file deletion gracefully."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Test content\n")
|
|
|
|
# Read the file to cache it
|
|
read_chunk = await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(read_chunk.state, "running")
|
|
|
|
# Verify cache exists
|
|
self.assertEqual(len(self.state.tool_context.read_file_cache), 1)
|
|
self.assertEqual(
|
|
self.state.tool_context.read_file_cache[0].file_path,
|
|
self.test_file,
|
|
)
|
|
|
|
# Delete the file
|
|
os.unlink(self.test_file)
|
|
|
|
# Try to edit - should fail with "File not found" error
|
|
edit_chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="Test",
|
|
new_string="New",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Should fail with "File not found" error
|
|
self.assertEqual(edit_chunk.state, "error")
|
|
self.assertIn("not found", edit_chunk.content[0].text.lower())
|
|
|
|
# Try to read again - should also fail
|
|
read_chunk2 = await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(read_chunk2.state, "error")
|
|
self.assertIn("does not exist", read_chunk2.content[0].text)
|
|
|
|
async def test_cache_invalidation_after_file_modification(self) -> None:
|
|
"""Test cache detects file modification."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Original content\n")
|
|
|
|
# Read the file to cache it
|
|
read_chunk = await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(read_chunk.state, "running")
|
|
|
|
# Modify the file externally
|
|
import time
|
|
|
|
time.sleep(0.1) # Ensure mtime changes
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Modified content\n")
|
|
|
|
# Try to edit - should fail because cache is stale
|
|
edit_chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="Original",
|
|
new_string="New",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Should fail with error
|
|
self.assertEqual(edit_chunk.state, "error")
|
|
self.assertIn("must first read", edit_chunk.content[0].text)
|
|
|
|
async def test_cache_lru_eviction(self) -> None:
|
|
"""Test LRU cache eviction when max_cache_files is exceeded."""
|
|
# Set a small cache limit
|
|
self.state.tool_context.max_cache_files = 3
|
|
|
|
# Create and read 4 files
|
|
files = []
|
|
for i in range(4):
|
|
file_path = os.path.join(self.temp_dir, f"file{i}.txt")
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
f.write(f"Content {i}\n")
|
|
files.append(file_path)
|
|
|
|
# Read each file
|
|
await self.read_tool(
|
|
file_path=file_path,
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Cache should only have 3 files (oldest evicted)
|
|
self.assertEqual(len(self.state.tool_context.read_file_cache), 3)
|
|
|
|
# The first file should have been evicted
|
|
cached_paths = [
|
|
entry.file_path
|
|
for entry in self.state.tool_context.read_file_cache
|
|
]
|
|
self.assertNotIn(files[0], cached_paths)
|
|
self.assertIn(files[1], cached_paths)
|
|
self.assertIn(files[2], cached_paths)
|
|
self.assertIn(files[3], cached_paths)
|
|
|
|
async def test_cache_without_state(self) -> None:
|
|
"""Test tools work without state (fallback mode)."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Hello World\n")
|
|
|
|
# Edit without state should work (fallback to reading from disk)
|
|
edit_chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="Hello",
|
|
new_string="Hi",
|
|
_agent_state=None,
|
|
)
|
|
|
|
self.assertEqual(edit_chunk.state, "running")
|
|
|
|
# Verify file was edited
|
|
with open(self.test_file, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.assertEqual(content, "Hi World\n")
|
|
|
|
async def test_multiple_reads_update_cache(self) -> None:
|
|
"""Test reading same file multiple times updates cache."""
|
|
# Create a file
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Content v1\n")
|
|
|
|
# Read the file
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(len(self.state.tool_context.read_file_cache), 1)
|
|
|
|
# Read again - should update cache, not add duplicate
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(len(self.state.tool_context.read_file_cache), 1)
|
|
|
|
async def test_read_cache_lines_contain_newlines(self) -> None:
|
|
"""Test that cached lines from Read retain trailing newlines.
|
|
|
|
readlines() includes the newline character in each line. The cache
|
|
must store them as-is so that "".join(lines) reconstructs the exact
|
|
original file content.
|
|
"""
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("line1\nline2\nline3\n")
|
|
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
cache = await self.state.tool_context.get_cache(self.test_file)
|
|
self.assertIsNotNone(cache)
|
|
# Each line from readlines() ends with \n
|
|
self.assertEqual(cache.lines, ["line1\n", "line2\n", "line3\n"])
|
|
# "".join reconstructs the exact original content
|
|
self.assertEqual("".join(cache.lines), "line1\nline2\nline3\n")
|
|
|
|
async def test_edit_multiline_match_from_cache(self) -> None:
|
|
"""Test Edit correctly matches multi-line old_string from
|
|
cached content.
|
|
|
|
Regression test for the bug where "\n".join(cache.lines) doubled the
|
|
newlines (e.g. "line1\n\nline2\n" instead of "line1\nline2\n"),
|
|
making multi-line old_string matching fail.
|
|
"""
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("line1\nline2\nline3\n")
|
|
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# This multi-line old_string must match exactly in the reconstructed
|
|
# content; with the bug it would not be found.
|
|
chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="line1\nline2",
|
|
new_string="replaced",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
self.assertEqual(chunk.state, "running")
|
|
|
|
with open(self.test_file, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.assertEqual(content, "replaced\nline3\n")
|
|
|
|
async def test_edit_single_line_match_from_cache(self) -> None:
|
|
"""Test Edit correctly matches single-line old_string from cache."""
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("Hello World\nThis is a test\n")
|
|
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="Hello World",
|
|
new_string="Hello Python",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
self.assertEqual(chunk.state, "running")
|
|
|
|
with open(self.test_file, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.assertEqual(content, "Hello Python\nThis is a test\n")
|
|
|
|
async def test_write_invalidates_cache(self) -> None:
|
|
"""Test that Write updates the cache so Edit can use it afterwards."""
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("original content\n")
|
|
|
|
# Read to populate cache
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Overwrite with Write (mtime changes, old cache becomes stale)
|
|
write_chunk = await self.write_tool(
|
|
file_path=self.test_file,
|
|
content="new content\n",
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(write_chunk.state, "running")
|
|
|
|
# The old cache entry is now stale; Edit should require a new Read
|
|
edit_chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="new content",
|
|
new_string="updated content",
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(edit_chunk.state, "error")
|
|
self.assertIn("must first read", edit_chunk.content[0].text)
|
|
|
|
async def test_write_cache_stale_then_reread(self) -> None:
|
|
"""Test workflow: Read -> Write -> Read -> Edit works correctly."""
|
|
with open(self.test_file, "w", encoding="utf-8") as f:
|
|
f.write("original\n")
|
|
|
|
# First read
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Overwrite
|
|
import time
|
|
|
|
time.sleep(0.01) # ensure mtime changes
|
|
await self.write_tool(
|
|
file_path=self.test_file,
|
|
content="rewritten\n",
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Re-read to refresh cache
|
|
await self.read_tool(
|
|
file_path=self.test_file,
|
|
_agent_state=self.state,
|
|
)
|
|
|
|
# Now Edit should succeed against the new content
|
|
edit_chunk = await self.edit_tool(
|
|
file_path=self.test_file,
|
|
old_string="rewritten",
|
|
new_string="final",
|
|
_agent_state=self.state,
|
|
)
|
|
self.assertEqual(edit_chunk.state, "running")
|
|
|
|
with open(self.test_file, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
self.assertEqual(content, "final\n")
|