ef2ae3e48f
After a successful write_file or patch, update the stored read timestamp to match the file's new modification time. Without this, consecutive edits by the same task (read → write → write) would false-warn on the second write because the stored timestamp still reflected the original read, not the first write. Also renames the internal tracker key from 'file_mtimes' to 'read_timestamps' for clarity.
242 lines
8.2 KiB
Python
242 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tests for file staleness detection in write_file and patch.
|
|
|
|
When a file is modified externally between the agent's read and write,
|
|
the write should include a warning so the agent can re-read and verify.
|
|
|
|
Run with: python -m pytest tests/tools/test_file_staleness.py -v
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from tools.file_tools import (
|
|
read_file_tool,
|
|
write_file_tool,
|
|
patch_tool,
|
|
clear_read_tracker,
|
|
_check_file_staleness,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _FakeReadResult:
|
|
def __init__(self, content="line1\nline2\n", total_lines=2, file_size=100):
|
|
self.content = content
|
|
self._total_lines = total_lines
|
|
self._file_size = file_size
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"content": self.content,
|
|
"total_lines": self._total_lines,
|
|
"file_size": self._file_size,
|
|
}
|
|
|
|
|
|
class _FakeWriteResult:
|
|
def __init__(self):
|
|
self.bytes_written = 10
|
|
|
|
def to_dict(self):
|
|
return {"bytes_written": self.bytes_written}
|
|
|
|
|
|
class _FakePatchResult:
|
|
def __init__(self):
|
|
self.success = True
|
|
|
|
def to_dict(self):
|
|
return {"success": True, "diff": "--- a\n+++ b\n@@ ...\n"}
|
|
|
|
|
|
def _make_fake_ops(read_content="hello\n", file_size=6):
|
|
fake = MagicMock()
|
|
fake.read_file = lambda path, offset=1, limit=500: _FakeReadResult(
|
|
content=read_content, total_lines=1, file_size=file_size,
|
|
)
|
|
fake.write_file = lambda path, content: _FakeWriteResult()
|
|
fake.patch_replace = lambda path, old, new, replace_all=False: _FakePatchResult()
|
|
return fake
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Core staleness check
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStalenessCheck(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
clear_read_tracker()
|
|
self._tmpdir = tempfile.mkdtemp()
|
|
self._tmpfile = os.path.join(self._tmpdir, "stale_test.txt")
|
|
with open(self._tmpfile, "w") as f:
|
|
f.write("original content\n")
|
|
|
|
def tearDown(self):
|
|
clear_read_tracker()
|
|
try:
|
|
os.unlink(self._tmpfile)
|
|
os.rmdir(self._tmpdir)
|
|
except OSError:
|
|
pass
|
|
|
|
@patch("tools.file_tools._get_file_ops")
|
|
def test_no_warning_when_file_unchanged(self, mock_ops):
|
|
"""Read then write with no external modification — no warning."""
|
|
mock_ops.return_value = _make_fake_ops("original content\n", 18)
|
|
read_file_tool(self._tmpfile, task_id="t1")
|
|
|
|
result = json.loads(write_file_tool(self._tmpfile, "new content", task_id="t1"))
|
|
self.assertNotIn("_warning", result)
|
|
|
|
@patch("tools.file_tools._get_file_ops")
|
|
def test_warning_when_file_modified_externally(self, mock_ops):
|
|
"""Read, then external modify, then write — should warn."""
|
|
mock_ops.return_value = _make_fake_ops("original content\n", 18)
|
|
read_file_tool(self._tmpfile, task_id="t1")
|
|
|
|
# Simulate external modification
|
|
time.sleep(0.05)
|
|
with open(self._tmpfile, "w") as f:
|
|
f.write("someone else changed this\n")
|
|
|
|
result = json.loads(write_file_tool(self._tmpfile, "new content", task_id="t1"))
|
|
self.assertIn("_warning", result)
|
|
self.assertIn("modified since you last read", result["_warning"])
|
|
|
|
@patch("tools.file_tools._get_file_ops")
|
|
def test_no_warning_when_file_never_read(self, mock_ops):
|
|
"""Writing a file that was never read — no warning."""
|
|
mock_ops.return_value = _make_fake_ops()
|
|
result = json.loads(write_file_tool(self._tmpfile, "new content", task_id="t2"))
|
|
self.assertNotIn("_warning", result)
|
|
|
|
@patch("tools.file_tools._get_file_ops")
|
|
def test_no_warning_for_new_file(self, mock_ops):
|
|
"""Creating a new file — no warning."""
|
|
mock_ops.return_value = _make_fake_ops()
|
|
new_path = os.path.join(self._tmpdir, "brand_new.txt")
|
|
result = json.loads(write_file_tool(new_path, "content", task_id="t3"))
|
|
self.assertNotIn("_warning", result)
|
|
try:
|
|
os.unlink(new_path)
|
|
except OSError:
|
|
pass
|
|
|
|
@patch("tools.file_tools._get_file_ops")
|
|
def test_different_task_isolated(self, mock_ops):
|
|
"""Task A reads, file changes, Task B writes — no warning for B."""
|
|
mock_ops.return_value = _make_fake_ops("original content\n", 18)
|
|
read_file_tool(self._tmpfile, task_id="task_a")
|
|
|
|
time.sleep(0.05)
|
|
with open(self._tmpfile, "w") as f:
|
|
f.write("changed\n")
|
|
|
|
result = json.loads(write_file_tool(self._tmpfile, "new", task_id="task_b"))
|
|
self.assertNotIn("_warning", result)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Staleness in patch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPatchStaleness(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
clear_read_tracker()
|
|
self._tmpdir = tempfile.mkdtemp()
|
|
self._tmpfile = os.path.join(self._tmpdir, "patch_test.txt")
|
|
with open(self._tmpfile, "w") as f:
|
|
f.write("original line\n")
|
|
|
|
def tearDown(self):
|
|
clear_read_tracker()
|
|
try:
|
|
os.unlink(self._tmpfile)
|
|
os.rmdir(self._tmpdir)
|
|
except OSError:
|
|
pass
|
|
|
|
@patch("tools.file_tools._get_file_ops")
|
|
def test_patch_warns_on_stale_file(self, mock_ops):
|
|
"""Patch should warn if the target file changed since last read."""
|
|
mock_ops.return_value = _make_fake_ops("original line\n", 15)
|
|
read_file_tool(self._tmpfile, task_id="p1")
|
|
|
|
time.sleep(0.05)
|
|
with open(self._tmpfile, "w") as f:
|
|
f.write("externally modified\n")
|
|
|
|
result = json.loads(patch_tool(
|
|
mode="replace", path=self._tmpfile,
|
|
old_string="original", new_string="patched",
|
|
task_id="p1",
|
|
))
|
|
self.assertIn("_warning", result)
|
|
self.assertIn("modified since you last read", result["_warning"])
|
|
|
|
@patch("tools.file_tools._get_file_ops")
|
|
def test_patch_no_warning_when_fresh(self, mock_ops):
|
|
"""Patch with no external changes — no warning."""
|
|
mock_ops.return_value = _make_fake_ops("original line\n", 15)
|
|
read_file_tool(self._tmpfile, task_id="p2")
|
|
|
|
result = json.loads(patch_tool(
|
|
mode="replace", path=self._tmpfile,
|
|
old_string="original", new_string="patched",
|
|
task_id="p2",
|
|
))
|
|
self.assertNotIn("_warning", result)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unit test for the helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCheckFileStalenessHelper(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
clear_read_tracker()
|
|
|
|
def tearDown(self):
|
|
clear_read_tracker()
|
|
|
|
def test_returns_none_for_unknown_task(self):
|
|
self.assertIsNone(_check_file_staleness("/tmp/x.py", "nonexistent"))
|
|
|
|
def test_returns_none_for_unread_file(self):
|
|
# Populate tracker with a different file
|
|
from tools.file_tools import _read_tracker, _read_tracker_lock
|
|
with _read_tracker_lock:
|
|
_read_tracker["t1"] = {
|
|
"last_key": None, "consecutive": 0,
|
|
"read_history": set(), "dedup": {},
|
|
"read_timestamps": {"/tmp/other.py": 12345.0},
|
|
}
|
|
self.assertIsNone(_check_file_staleness("/tmp/x.py", "t1"))
|
|
|
|
def test_returns_none_when_stat_fails(self):
|
|
from tools.file_tools import _read_tracker, _read_tracker_lock
|
|
with _read_tracker_lock:
|
|
_read_tracker["t1"] = {
|
|
"last_key": None, "consecutive": 0,
|
|
"read_history": set(), "dedup": {},
|
|
"read_timestamps": {"/nonexistent/path": 99999.0},
|
|
}
|
|
# File doesn't exist → stat fails → returns None (let write handle it)
|
|
self.assertIsNone(_check_file_staleness("/nonexistent/path", "t1"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|