test: trim memory write-bridge regression tests to the invariants
Keep the two tests that are red on unchanged main: previous_content comes from the locked native-store result (single + batch replace/remove, batch ordering) and never from caller arguments or build_metadata. Drop the uncommitted-batch cases (they pass on main -- the no-notification gate on a failed write predates this change) and the memory/user target axis (same store code path). Salvage of #118903 by @ehz0ah.
This commit is contained in:
@@ -104,7 +104,6 @@ def test_build_metadata_callback_is_merged_per_op():
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('target', ['memory', 'user'])
|
||||
@pytest.mark.parametrize('batch,operations,previous', [
|
||||
(False, [{'action': 'remove', 'old_text': 'Prefers tea'}], ['Prefers tea']),
|
||||
(False, [{'action': 'replace', 'old_text': 'Prefers tea', 'content': 'Prefers coffee'}], ['Prefers tea']),
|
||||
@@ -117,8 +116,9 @@ def test_build_metadata_callback_is_merged_per_op():
|
||||
], [None, 'Uses the blue notebook', 'Uses the green notebook']),
|
||||
])
|
||||
def test_committed_entry_identity_comes_from_locked_store(
|
||||
tmp_path, monkeypatch, target, batch, operations, previous
|
||||
tmp_path, monkeypatch, batch, operations, previous
|
||||
):
|
||||
target = 'memory'
|
||||
from contextlib import contextmanager
|
||||
from tools import memory_tool_store
|
||||
from tools.memory_tool import MemoryStore, memory_tool
|
||||
@@ -166,43 +166,6 @@ def test_committed_entry_identity_comes_from_locked_store(
|
||||
assert 'Prefers tea with milk' in store._entries_for(target)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('failure', ['invalid_operation', 'over_budget', 'empty_store', 'disk_write'])
|
||||
def test_uncommitted_batch_emits_no_entry_metadata_or_notifications(tmp_path, monkeypatch, failure):
|
||||
from tools.memory_tool import MemoryStore, memory_tool
|
||||
|
||||
monkeypatch.setattr('tools.memory_tool.get_memory_dir', lambda: tmp_path)
|
||||
store = MemoryStore(memory_char_limit=100)
|
||||
store.load_from_disk()
|
||||
store.add('memory', 'Keep this entry')
|
||||
path = store._path_for('memory')
|
||||
before = path.read_bytes()
|
||||
manager, provider = _manager_with_provider()
|
||||
operations = [{'action': 'replace', 'old_text': 'Keep this', 'content': 'Changed entry'}]
|
||||
if failure == 'invalid_operation':
|
||||
operations.append({'action': 'remove', 'old_text': 'Absent entry'})
|
||||
elif failure == 'over_budget':
|
||||
operations.append({'action': 'add', 'content': 'x' * 101})
|
||||
elif failure == 'empty_store':
|
||||
operations.append({'action': 'remove', 'old_text': 'Changed entry'})
|
||||
else:
|
||||
def fail_write(*_args):
|
||||
raise OSError('disk unavailable')
|
||||
monkeypatch.setattr(store, '_write_file', fail_write)
|
||||
args = {'target': 'memory', 'operations': operations}
|
||||
if failure == 'disk_write':
|
||||
with pytest.raises(OSError, match='disk unavailable'):
|
||||
result = memory_tool(store=store, **args)
|
||||
manager.notify_memory_tool_write(result, args)
|
||||
else:
|
||||
result = memory_tool(store=store, **args)
|
||||
payload = json.loads(result)
|
||||
assert payload['success'] is False
|
||||
assert not any(key in payload for key in ('replaced_entries', 'removed_entries'))
|
||||
manager.notify_memory_tool_write(result, args)
|
||||
assert provider.calls == []
|
||||
assert path.read_bytes() == before
|
||||
|
||||
|
||||
def test_previous_content_cannot_come_from_uncommitted_arguments():
|
||||
manager, provider = _manager_with_provider()
|
||||
args = {'action': 'remove', 'old_text': 'partial', 'previous_content': 'Untrusted argument'}
|
||||
|
||||
Reference in New Issue
Block a user