426 lines
16 KiB
Python
426 lines
16 KiB
Python
"""Tests for compact_memory_auto.py — change-set parsing, validation, applying and reporting.
|
|
|
|
The nanobot import is deferred inside _run, so importing the module needs no nanobot-ai install.
|
|
Nothing here touches the network or an LLM; every tested function is pure or writes into tmp_path.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
import compact_memory_auto as cma
|
|
|
|
MEMORY_TEXT = """# MEMORY
|
|
|
|
## Infra
|
|
|
|
- Runs as a systemd user service `nanobot.service`
|
|
- Model switching via `my` tool needs `tools.my.allow_set = true`
|
|
- Telegram bot polls every 2 seconds
|
|
|
|
## Projects
|
|
|
|
### compact-memory
|
|
|
|
- Nightly job driven by the crontab
|
|
- Debug run 2026-07-25: output-format test in progress
|
|
- Change-set validated by the script
|
|
|
|
### remind
|
|
|
|
- Reminders live in a SQLite database
|
|
- Cron sends due reminders every minute
|
|
"""
|
|
|
|
DEBUG_LINE = "- Debug run 2026-07-25: output-format test in progress"
|
|
SYSTEMD_LINE = "- Runs as a systemd user service `nanobot.service`"
|
|
PRESET_LINE = "- Model switching via `my` tool needs `tools.my.allow_set = true`"
|
|
|
|
NOW = datetime(2026, 7, 27, 2, 5)
|
|
|
|
|
|
def delete_item(original, reason="ephemeral marker", category="ephemeral", **overrides):
|
|
item = {"op": "delete", "category": category, "original": original, "reason": reason}
|
|
item.update(overrides)
|
|
return item
|
|
|
|
|
|
def merge_item(original, new_text, reason="same subsection, one topic", **overrides):
|
|
item = {"op": "merge", "original": original, "new_text": new_text, "reason": reason}
|
|
item.update(overrides)
|
|
return item
|
|
|
|
|
|
def answer(*changes):
|
|
"""Wrap changes the way the agent does — one fenced json block."""
|
|
return "```json\n" + json.dumps({"changes": list(changes)}) + "\n```"
|
|
|
|
|
|
@pytest.fixture
|
|
def workspace(tmp_path):
|
|
memory = tmp_path / "memory"
|
|
memory.mkdir()
|
|
(memory / "MEMORY.md").write_text(MEMORY_TEXT, encoding="utf-8")
|
|
return tmp_path
|
|
|
|
|
|
# --- _extract_json ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_extract_json_from_fenced_block():
|
|
assert cma._extract_json('```json\n{"changes": []}\n```') == {"changes": []}
|
|
|
|
|
|
def test_extract_json_from_bare_answer():
|
|
assert cma._extract_json('{"changes": []}') == {"changes": []}
|
|
|
|
|
|
def test_extract_json_takes_last_parseable_block():
|
|
content = '```json\n{"changes": [1]}\n```\ntext\n```json\n{"changes": [2]}\n```'
|
|
assert cma._extract_json(content) == {"changes": [2]}
|
|
|
|
|
|
def test_extract_json_without_json_raises():
|
|
with pytest.raises(cma.ChangeSetError, match="no parseable"):
|
|
cma._extract_json("I audited the file and found nothing.")
|
|
|
|
|
|
# --- parse_change_set: happy paths -----------------------------------------------------------
|
|
|
|
|
|
def test_parse_delete():
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE])), MEMORY_TEXT)
|
|
assert len(located) == 1
|
|
change = located[0].change
|
|
assert change.op == "delete"
|
|
assert change.category == "ephemeral"
|
|
assert change.original == (DEBUG_LINE,)
|
|
assert change.new_text == ()
|
|
|
|
|
|
def test_parse_merge():
|
|
merged = "- systemd service `nanobot.service`; model switching needs `tools.my.allow_set = true`"
|
|
located = cma.parse_change_set(answer(merge_item([SYSTEMD_LINE, PRESET_LINE], [merged])), MEMORY_TEXT)
|
|
assert len(located) == 1
|
|
assert located[0].change.op == "merge"
|
|
assert located[0].change.category == "merge"
|
|
assert located[0].change.new_text == (merged,)
|
|
|
|
|
|
def test_parse_empty_change_set():
|
|
assert cma.parse_change_set(answer(), MEMORY_TEXT) == []
|
|
|
|
|
|
def test_parse_strips_reason_whitespace():
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE], reason=" stale ")), MEMORY_TEXT)
|
|
assert located[0].change.reason == "stale"
|
|
|
|
|
|
def test_parse_reports_span_of_matched_block():
|
|
lines = MEMORY_TEXT.splitlines()
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE])), MEMORY_TEXT)
|
|
assert located[0].start == lines.index(DEBUG_LINE)
|
|
assert located[0].end == located[0].start + 1
|
|
|
|
|
|
# --- parse_change_set: payload shape ---------------------------------------------------------
|
|
|
|
|
|
def test_payload_must_be_object():
|
|
with pytest.raises(cma.ChangeSetError, match="must be an object"):
|
|
cma.parse_change_set("```json\n[]\n```", MEMORY_TEXT)
|
|
|
|
|
|
def test_payload_needs_changes_list():
|
|
with pytest.raises(cma.ChangeSetError, match='"changes" list'):
|
|
cma.parse_change_set('```json\n{"items": []}\n```', MEMORY_TEXT)
|
|
|
|
|
|
def test_item_must_be_object():
|
|
with pytest.raises(cma.ChangeSetError, match="must be a JSON object"):
|
|
cma.parse_change_set(answer("delete everything"), MEMORY_TEXT)
|
|
|
|
|
|
# --- parse_change_set: field validation ------------------------------------------------------
|
|
|
|
|
|
def test_unknown_op_rejected():
|
|
with pytest.raises(cma.ChangeSetError, match='"op" must be'):
|
|
cma.parse_change_set(answer(delete_item([DEBUG_LINE], op="rewrite")), MEMORY_TEXT)
|
|
|
|
|
|
def test_unknown_field_rejected():
|
|
with pytest.raises(cma.ChangeSetError, match="unknown fields"):
|
|
cma.parse_change_set(answer(delete_item([DEBUG_LINE], note="extra")), MEMORY_TEXT)
|
|
|
|
|
|
def test_merge_field_on_delete_rejected():
|
|
with pytest.raises(cma.ChangeSetError, match="unknown fields"):
|
|
cma.parse_change_set(answer(delete_item([DEBUG_LINE], new_text=["x"])), MEMORY_TEXT)
|
|
|
|
|
|
def test_empty_reason_rejected():
|
|
with pytest.raises(cma.ChangeSetError, match='"reason" must be a non-empty string'):
|
|
cma.parse_change_set(answer(delete_item([DEBUG_LINE], reason=" ")), MEMORY_TEXT)
|
|
|
|
|
|
def test_too_long_reason_rejected():
|
|
long_reason = "x" * (cma.MAX_REASON_CHARS + 1)
|
|
with pytest.raises(cma.ChangeSetError, match=f"the limit is {cma.MAX_REASON_CHARS}"):
|
|
cma.parse_change_set(answer(delete_item([DEBUG_LINE], reason=long_reason)), MEMORY_TEXT)
|
|
|
|
|
|
def test_original_must_be_non_empty_list():
|
|
with pytest.raises(cma.ChangeSetError, match='"original" must be a non-empty list'):
|
|
cma.parse_change_set(answer(delete_item([])), MEMORY_TEXT)
|
|
|
|
|
|
def test_original_must_be_list_of_strings():
|
|
with pytest.raises(cma.ChangeSetError, match='"original" must be a non-empty list'):
|
|
cma.parse_change_set(answer(delete_item([DEBUG_LINE, 42])), MEMORY_TEXT)
|
|
|
|
|
|
def test_too_many_original_lines_rejected():
|
|
block = [f"- line {i}" for i in range(cma.MAX_ORIGINAL_LINES + 1)]
|
|
with pytest.raises(cma.ChangeSetError, match=f"the limit is {cma.MAX_ORIGINAL_LINES}"):
|
|
cma.parse_change_set(answer(delete_item(block)), MEMORY_TEXT)
|
|
|
|
|
|
def test_unknown_delete_category_rejected():
|
|
with pytest.raises(cma.ChangeSetError, match='"category" must be one of'):
|
|
cma.parse_change_set(answer(delete_item([DEBUG_LINE], category="obsolete")), MEMORY_TEXT)
|
|
|
|
|
|
def test_too_many_new_text_lines_rejected():
|
|
original = [SYSTEMD_LINE, PRESET_LINE, "- Telegram bot polls every 2 seconds"]
|
|
new_text = [f"- merged {i}" for i in range(cma.MAX_NEW_TEXT_LINES + 1)]
|
|
with pytest.raises(cma.ChangeSetError, match=f"the limit is {cma.MAX_NEW_TEXT_LINES}"):
|
|
cma.parse_change_set(answer(merge_item(original, new_text)), MEMORY_TEXT)
|
|
|
|
|
|
def test_too_long_new_text_rejected():
|
|
new_text = ["- " + "x" * cma.MAX_NEW_TEXT_CHARS]
|
|
with pytest.raises(cma.ChangeSetError, match=f"the limit is {cma.MAX_NEW_TEXT_CHARS}"):
|
|
cma.parse_change_set(answer(merge_item([SYSTEMD_LINE, PRESET_LINE], new_text)), MEMORY_TEXT)
|
|
|
|
|
|
def test_merge_must_shrink():
|
|
new_text = ["- one", "- two"]
|
|
with pytest.raises(cma.ChangeSetError, match="fewer lines"):
|
|
cma.parse_change_set(answer(merge_item([SYSTEMD_LINE, PRESET_LINE], new_text)), MEMORY_TEXT)
|
|
|
|
|
|
def test_all_item_errors_reported_at_once():
|
|
bad_op = delete_item([DEBUG_LINE], op="rewrite")
|
|
bad_reason = delete_item([SYSTEMD_LINE], reason="")
|
|
with pytest.raises(cma.ChangeSetError) as excinfo:
|
|
cma.parse_change_set(answer(bad_op, bad_reason), MEMORY_TEXT)
|
|
message = str(excinfo.value)
|
|
assert "item 1:" in message
|
|
assert "item 2:" in message
|
|
|
|
|
|
# --- locating blocks in the file -------------------------------------------------------------
|
|
|
|
|
|
def test_missing_block_rejected():
|
|
with pytest.raises(cma.ChangeSetError, match=r"does not appear in MEMORY\.md"):
|
|
cma.parse_change_set(answer(delete_item(["- this line was never in the file"])), MEMORY_TEXT)
|
|
|
|
|
|
def test_ambiguous_block_rejected():
|
|
text = "- duplicated bullet\n- something else\n- duplicated bullet\n"
|
|
with pytest.raises(cma.ChangeSetError, match="appears 2 times"):
|
|
cma.parse_change_set(answer(delete_item(["- duplicated bullet"])), text)
|
|
|
|
|
|
def test_trailing_whitespace_is_ignored_when_matching():
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE + " "])), MEMORY_TEXT)
|
|
assert located[0].change.original == (DEBUG_LINE + " ",)
|
|
|
|
|
|
def test_located_changes_are_sorted_by_position():
|
|
later = delete_item([DEBUG_LINE])
|
|
earlier = delete_item([SYSTEMD_LINE], category="detail", reason="code reference")
|
|
located = cma.parse_change_set(answer(later, earlier), MEMORY_TEXT)
|
|
assert [item.change.original[0] for item in located] == [SYSTEMD_LINE, DEBUG_LINE]
|
|
|
|
|
|
def test_overlapping_blocks_rejected():
|
|
first = delete_item([SYSTEMD_LINE, PRESET_LINE])
|
|
second = delete_item([PRESET_LINE, "- Telegram bot polls every 2 seconds"])
|
|
with pytest.raises(cma.ChangeSetError, match="overlapping items"):
|
|
cma.parse_change_set(answer(first, second), MEMORY_TEXT)
|
|
|
|
|
|
def test_removing_more_than_half_the_file_rejected():
|
|
text = "- one\n- two\n- three\n- four\n"
|
|
block = ["- one", "- two", "- three"]
|
|
with pytest.raises(cma.ChangeSetError, match="more than 50% of the file"):
|
|
cma.parse_change_set(answer(delete_item(block)), text)
|
|
|
|
|
|
# --- apply_change_set ------------------------------------------------------------------------
|
|
|
|
|
|
def test_apply_delete_removes_the_block(workspace):
|
|
memory = workspace / "memory" / "MEMORY.md"
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE])), MEMORY_TEXT)
|
|
|
|
lines_after = cma.apply_change_set(memory, located, workspace, NOW)
|
|
|
|
text = memory.read_text(encoding="utf-8")
|
|
assert DEBUG_LINE not in text
|
|
assert SYSTEMD_LINE in text
|
|
assert text.endswith("\n")
|
|
assert lines_after == len(MEMORY_TEXT.splitlines()) - 1
|
|
assert lines_after == len(text.splitlines())
|
|
|
|
|
|
def test_apply_merge_replaces_the_block(workspace):
|
|
memory = workspace / "memory" / "MEMORY.md"
|
|
merged = "- systemd service; model switching needs `tools.my.allow_set = true`"
|
|
located = cma.parse_change_set(answer(merge_item([SYSTEMD_LINE, PRESET_LINE], [merged])), MEMORY_TEXT)
|
|
|
|
lines_after = cma.apply_change_set(memory, located, workspace, NOW)
|
|
|
|
text = memory.read_text(encoding="utf-8")
|
|
assert merged in text
|
|
assert SYSTEMD_LINE not in text
|
|
assert PRESET_LINE not in text
|
|
assert lines_after == len(MEMORY_TEXT.splitlines()) - 1
|
|
|
|
|
|
def test_apply_writes_backup_with_the_original_content(workspace):
|
|
memory = workspace / "memory" / "MEMORY.md"
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE])), MEMORY_TEXT)
|
|
|
|
cma.apply_change_set(memory, located, workspace, NOW)
|
|
|
|
backup = workspace / "backup" / "2026-07-27_0205_memory.backup.md"
|
|
assert backup.read_text(encoding="utf-8") == MEMORY_TEXT
|
|
|
|
|
|
def test_apply_appends_to_the_clean_log(workspace):
|
|
memory = workspace / "memory" / "MEMORY.md"
|
|
clean_log = workspace / cma.CLEAN_LOG_REL
|
|
clean_log.parent.mkdir(parents=True)
|
|
clean_log.write_text('2026-07-26 02:05 DELETED [detail] "old entry" — earlier run\n', encoding="utf-8")
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE])), MEMORY_TEXT)
|
|
|
|
cma.apply_change_set(memory, located, workspace, NOW)
|
|
|
|
lines = clean_log.read_text(encoding="utf-8").splitlines()
|
|
assert len(lines) == 2
|
|
assert lines[0].endswith("earlier run")
|
|
assert lines[1].startswith("2026-07-27 02:05 DELETED [ephemeral]")
|
|
|
|
|
|
def test_apply_logs_one_line_per_change(workspace):
|
|
memory = workspace / "memory" / "MEMORY.md"
|
|
merged = "- systemd service; model switching needs `tools.my.allow_set = true`"
|
|
located = cma.parse_change_set(
|
|
answer(delete_item([DEBUG_LINE]), merge_item([SYSTEMD_LINE, PRESET_LINE], [merged])),
|
|
MEMORY_TEXT,
|
|
)
|
|
|
|
cma.apply_change_set(memory, located, workspace, NOW)
|
|
|
|
lines = (workspace / cma.CLEAN_LOG_REL).read_text(encoding="utf-8").splitlines()
|
|
assert len(lines) == 2
|
|
assert any("MERGED [merge]" in line for line in lines)
|
|
assert any("DELETED [ephemeral]" in line for line in lines)
|
|
|
|
|
|
# --- reporting -------------------------------------------------------------------------------
|
|
|
|
|
|
def test_report_for_empty_change_set():
|
|
assert cma.format_report([], 42, 42) == "Memory compact: nothing to remove (MEMORY.md, 42 lines)."
|
|
|
|
|
|
def test_report_counts_deletes_and_merges():
|
|
merged = "- systemd service; model switching needs `tools.my.allow_set = true`"
|
|
located = cma.parse_change_set(
|
|
answer(delete_item([DEBUG_LINE]), merge_item([SYSTEMD_LINE, PRESET_LINE], [merged])),
|
|
MEMORY_TEXT,
|
|
)
|
|
|
|
rows = cma.format_report(located, 20, 18).splitlines()
|
|
|
|
assert rows[0] == "Memory compact: deleted 1, merged 1 (20 → 18 lines)."
|
|
assert len(rows) == 3
|
|
assert any(row.startswith("- [merge] ") for row in rows[1:])
|
|
assert any(row.startswith("- [ephemeral] ") for row in rows[1:])
|
|
|
|
|
|
def test_report_quotes_reason_verbatim():
|
|
located = cma.parse_change_set(answer(delete_item([DEBUG_LINE], reason="task finished")), MEMORY_TEXT)
|
|
assert cma.format_report(located, 20, 19).endswith("— task finished")
|
|
|
|
|
|
def test_log_line_for_delete():
|
|
change = cma.Change(op="delete", category="detail", original=("- a path",), new_text=(), reason="belongs in code")
|
|
assert cma._log_line(change, "2026-07-27 02:05") == '2026-07-27 02:05 DELETED [detail] "- a path" — belongs in code'
|
|
|
|
|
|
def test_log_line_for_merge():
|
|
change = cma.Change(op="merge", category="merge", original=("- a", "- b"), new_text=("- ab",), reason="one topic")
|
|
assert cma._log_line(change, "2026-07-27 02:05") == '2026-07-27 02:05 MERGED [merge] "- a - b" → "- ab" — one topic'
|
|
|
|
|
|
# --- helpers ---------------------------------------------------------------------------------
|
|
|
|
|
|
def test_shorten_keeps_short_text():
|
|
assert cma._shorten("short", 10) == "short"
|
|
|
|
|
|
def test_shorten_truncates_with_ellipsis():
|
|
assert cma._shorten("abcdefghij", 5) == "abcd…"
|
|
|
|
|
|
def test_shorten_strips_before_the_ellipsis():
|
|
assert cma._shorten("ab cdefgh", 4) == "ab…"
|
|
|
|
|
|
def test_quote_joins_stripped_lines():
|
|
assert cma._quote((" - first ", "- second")) == "- first - second"
|
|
|
|
|
|
def test_quote_truncates_long_blocks():
|
|
quoted = cma._quote(tuple(f"- line {i}" for i in range(30)))
|
|
assert len(quoted) == cma.QUOTE_CHARS
|
|
assert quoted.endswith("…")
|
|
|
|
|
|
# --- config ----------------------------------------------------------------------------------
|
|
|
|
|
|
def test_telegram_config_uses_first_allowed_chat():
|
|
config = {"channels": {"telegram": {"token": "tok", "allowFrom": [12345, 999]}}}
|
|
assert cma._telegram_config(config) == ("tok", "12345")
|
|
|
|
|
|
def test_telegram_config_falls_back_when_allow_from_is_empty():
|
|
config = {"channels": {"telegram": {"token": "tok", "allowFrom": []}}}
|
|
assert cma._telegram_config(config) == ("tok", cma.FALLBACK_CHAT_ID)
|
|
|
|
|
|
def test_telegram_config_falls_back_when_allow_from_is_missing():
|
|
config = {"channels": {"telegram": {"token": "tok"}}}
|
|
assert cma._telegram_config(config) == ("tok", cma.FALLBACK_CHAT_ID)
|
|
|
|
|
|
def test_workspace_from_config():
|
|
config = {"agents": {"defaults": {"workspace": "~/custom/workspace"}}}
|
|
assert cma._workspace(config) == Path.home() / "custom" / "workspace"
|
|
|
|
|
|
def test_workspace_falls_back_when_unset():
|
|
assert cma._workspace({}) == cma.WORKSPACE_FALLBACK
|