mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-04 08:28:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(memory): add regression tests for missing cursor key

Browse Source 
Cover read_unprocessed_history skipping cursorless entries and
_next_cursor safe fallback when last entry has no cursor.
This commit is contained in:
chengyongru 2026-04-16 10:36:49 +08:00 committed by Xubin Ren
parent 524c097f76
commit d64e963258
2 changed files with 24 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
nanobot/agent/memory.py
View File

@ -239,7 +239,7 @@ class MemoryStore:
pass
# Fallback: read last line's cursor from the JSONL file.
last = self._read_last_entry()
if last and last.get("cursor") is not None:
if last and last.get("cursor"):
return last["cursor"] + 1
return 1

23
tests/agent/test_memory_store.py
View File

@ -79,6 +79,29 @@ class TestHistoryWithCursor:
entries = store.read_unprocessed_history(since_cursor=0)
assert len(entries) == 2
def test_read_unprocessed_skips_entries_without_cursor(self, store):
"""Regression: entries missing the cursor key should be silently skipped."""
store.history_file.write_text(
'{"timestamp": "2026-04-01 10:00", "content": "no cursor"}\n'
'{"cursor": 2, "timestamp": "2026-04-01 10:01", "content": "valid"}\n'
'{"cursor": 3, "timestamp": "2026-04-01 10:02", "content": "also valid"}\n',
encoding="utf-8",
)
entries = store.read_unprocessed_history(since_cursor=0)
assert [e["cursor"] for e in entries] == [2, 3]
def test_next_cursor_falls_back_when_last_entry_has_no_cursor(self, store):
"""Regression: _next_cursor should not KeyError on entries without cursor."""
store.history_file.write_text(
'{"timestamp": "2026-04-01 10:01", "content": "no cursor"}\n',
encoding="utf-8",
)
# Delete .cursor file so _next_cursor falls back to reading JSONL
store._cursor_file.unlink(missing_ok=True)
# Last entry has no cursor — should safely return 1, not KeyError
cursor = store.append_history("new event")
assert cursor == 1
def test_compact_history_drops_oldest(self, tmp_path):
store = MemoryStore(tmp_path, max_history_entries=2)
store.append_history("event 1")