pplate
42ffc85f0b
BUG-1: fix test_server_tools.py assert "ALWAYS" → "Always" (case mismatch)
BUG-2: export_memory() now includes hypotheses, upgrade_requests, token_saves,
people tables; renamed bigmind_version → bigmind_schema_version (int)
BUG-3: auto_close.py replaced CURRENT_TIMESTAMP (SQLite) with Python
datetime.now(timezone.utc).isoformat() for consistent UTC timestamps
PERF-1: context_builder.py caps get_facts() at _MAX_CONTEXT_FACTS=50 with
overflow hint to prevent unbounded context growth
All 297 tests passing. Upgrade requests #6-9 resolved.
Health report: plans/BIGMIND_HEALTH_REPORT_2026-04-04.md
866 lines
37 KiB
Python
866 lines
37 KiB
Python
"""Tests for BigMind MCP server tools (src/server.py).
|
|
|
|
All tests run against an isolated temp database (autouse fixture in conftest.py).
|
|
Server functions are called directly — no MCP transport layer needed for unit tests.
|
|
The module-level init_db() in server.py runs once at import time (harmless, idempotent).
|
|
All DATA operations are redirected to the temp DB by the monkeypatched BIGMIND_DB_PATH.
|
|
"""
|
|
import json
|
|
import re
|
|
import pytest
|
|
from datetime import datetime, timezone, timedelta
|
|
from pathlib import Path
|
|
|
|
from server import (
|
|
memory_start_session,
|
|
memory_end_session,
|
|
memory_flag_important,
|
|
memory_get_context,
|
|
memory_get_session_detail,
|
|
memory_search_chunks,
|
|
memory_list_sessions,
|
|
memory_store_fact,
|
|
memory_update_profile,
|
|
memory_append_chunk,
|
|
memory_get_stats,
|
|
memory_vacuum,
|
|
memory_get_instructions,
|
|
memory_health_check,
|
|
memory_export,
|
|
memory_deprecate_fact,
|
|
memory_add_hypothesis,
|
|
memory_resolve_hypothesis,
|
|
memory_list_hypotheses,
|
|
)
|
|
from bigmind import memory_store
|
|
from bigmind.db import db
|
|
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────────
|
|
|
|
def _start_and_get_id() -> str:
|
|
"""Call memory_start_session and return the session UUID."""
|
|
result = memory_start_session()
|
|
match = re.search(r"id: `([^`]+)`", result)
|
|
assert match, f"Could not extract session id from:\n{result}"
|
|
return match.group(1)
|
|
|
|
|
|
def _close(sid: str, headline: str = "Test session", topics: str = "test") -> str:
|
|
"""Convenience wrapper for memory_end_session."""
|
|
return memory_end_session(
|
|
session_id=sid,
|
|
one_liner=headline,
|
|
topics=topics,
|
|
outcome="Test outcome",
|
|
summary="Test summary narrative.",
|
|
)
|
|
|
|
|
|
# ── memory_start_session ───────────────────────────────────────────────────────
|
|
|
|
class TestMemoryStartSession:
|
|
def test_returns_started_confirmation(self, temp_db):
|
|
result = memory_start_session()
|
|
assert "BigMind session started" in result
|
|
|
|
def test_returns_valid_session_id(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
# UUIDs are 36 chars with 4 hyphens
|
|
assert len(sid) == 36
|
|
assert sid.count("-") == 4
|
|
|
|
def test_returns_context_markdown(self, temp_db):
|
|
result = memory_start_session()
|
|
assert "🧠 BigMind Context" in result
|
|
|
|
def test_creates_open_session_in_db(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
before = memory_store.get_open_sessions(user["id"])
|
|
memory_start_session()
|
|
after = memory_store.get_open_sessions(user["id"])
|
|
assert len(after) == len(before) + 1
|
|
|
|
def test_multiple_starts_create_multiple_sessions(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
memory_start_session()
|
|
memory_start_session()
|
|
open_sessions = memory_store.get_open_sessions(user["id"])
|
|
assert len(open_sessions) == 2
|
|
|
|
|
|
# ── memory_end_session ─────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryEndSession:
|
|
def test_returns_confirmation_with_headline(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result = _close(sid, headline="My important session")
|
|
assert "✅ Session closed" in result
|
|
assert "My important session" in result
|
|
|
|
def test_session_no_longer_open(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
sid = _start_and_get_id()
|
|
_close(sid)
|
|
open_sessions = memory_store.get_open_sessions(user["id"])
|
|
assert all(s["id"] != sid for s in open_sessions)
|
|
|
|
def test_session_appears_in_closed_list(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
_close(sid, headline="Findable session")
|
|
result = memory_list_sessions()
|
|
assert "Findable session" in result
|
|
|
|
def test_tier2_summary_saved(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_end_session(
|
|
session_id=sid,
|
|
one_liner="Narrative session",
|
|
topics="test",
|
|
outcome="Done",
|
|
summary="The full story lives here.",
|
|
key_facts="- Key insight one",
|
|
code_refs="src/server.py",
|
|
)
|
|
detail = memory_store.get_session_detail(sid)
|
|
assert detail is not None
|
|
assert "full story" in detail["summary"]
|
|
assert "Key insight one" in detail["key_facts"]
|
|
assert "src/server.py" in detail["code_refs"]
|
|
|
|
def test_importance_stored_correctly(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_end_session(
|
|
session_id=sid, one_liner="High importance",
|
|
topics="test", outcome="Done", summary="Summary", importance=9,
|
|
)
|
|
with db() as conn:
|
|
row = conn.execute(
|
|
"SELECT importance FROM sessions WHERE id=?", (sid,)
|
|
).fetchone()
|
|
assert row["importance"] == 9
|
|
|
|
def test_topics_stored(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
_close(sid, topics="bigmind,sqlite,memory")
|
|
result = memory_list_sessions()
|
|
assert "bigmind" in result
|
|
|
|
|
|
# ── memory_flag_important ──────────────────────────────────────────────────────
|
|
|
|
class TestMemoryFlagImportant:
|
|
def test_returns_tier3_confirmation(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result = memory_flag_important(
|
|
session_id=sid,
|
|
content="We decided to use SQLite",
|
|
flag_reason="architectural decision",
|
|
)
|
|
assert "✅ Stored as Tier-3 memory chunk" in result
|
|
|
|
def test_chunk_id_increments(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
r1 = memory_flag_important(session_id=sid, content="First chunk")
|
|
r2 = memory_flag_important(session_id=sid, content="Second chunk")
|
|
id1 = int(re.search(r"id: (\d+)", r1).group(1))
|
|
id2 = int(re.search(r"id: (\d+)", r2).group(1))
|
|
assert id2 > id1
|
|
|
|
def test_chunk_is_searchable_after_flagging(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(
|
|
session_id=sid,
|
|
content="PostgreSQL migration plan discussed",
|
|
)
|
|
result = memory_search_chunks("PostgreSQL")
|
|
assert "PostgreSQL" in result
|
|
|
|
def test_flag_reason_stored(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(
|
|
session_id=sid,
|
|
content="Some content",
|
|
flag_reason="user preference",
|
|
)
|
|
user = memory_store.get_or_create_user("testuser")
|
|
chunks = memory_store.search_chunks(user["id"], "Some content")
|
|
assert chunks[0]["flag_reason"] == "user preference"
|
|
|
|
def test_default_role_is_assistant(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(session_id=sid, content="Default role check")
|
|
user = memory_store.get_or_create_user("testuser")
|
|
chunks = memory_store.search_chunks(user["id"], "Default role check")
|
|
assert chunks[0]["role"] == "assistant"
|
|
|
|
def test_custom_role_stored(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(session_id=sid, content="User said this", role="user")
|
|
user = memory_store.get_or_create_user("testuser")
|
|
chunks = memory_store.search_chunks(user["id"], "User said this")
|
|
assert chunks[0]["role"] == "user"
|
|
|
|
|
|
# ── memory_get_context ─────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryGetContext:
|
|
def test_returns_markdown(self, temp_db):
|
|
result = memory_get_context()
|
|
assert "🧠 BigMind Context" in result
|
|
|
|
def test_does_not_create_new_session(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
before = memory_store.get_open_sessions(user["id"])
|
|
memory_get_context()
|
|
after = memory_store.get_open_sessions(user["id"])
|
|
assert len(before) == len(after)
|
|
|
|
def test_reflects_profile_changes(self, temp_db):
|
|
memory_update_profile(role="Staff Engineer")
|
|
result = memory_get_context()
|
|
assert "Staff Engineer" in result
|
|
|
|
def test_reflects_stored_facts(self, temp_db):
|
|
memory_store_fact(category="codebase", fact="Uses FastMCP for all servers")
|
|
result = memory_get_context()
|
|
assert "Uses FastMCP for all servers" in result
|
|
|
|
|
|
# ── memory_get_session_detail ──────────────────────────────────────────────────
|
|
|
|
class TestMemoryGetSessionDetail:
|
|
def test_returns_detail_for_session_with_summary(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_end_session(
|
|
session_id=sid, one_liner="Detailed one",
|
|
topics="test", outcome="Done",
|
|
summary="The complete story is stored here.",
|
|
)
|
|
result = memory_get_session_detail(sid)
|
|
assert "complete story" in result
|
|
|
|
def test_returns_key_facts_when_present(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_end_session(
|
|
session_id=sid, one_liner="With facts",
|
|
topics="test", outcome="Done", summary="Summary",
|
|
key_facts="- Decided on Python\n- Chose SQLite",
|
|
)
|
|
result = memory_get_session_detail(sid)
|
|
assert "Decided on Python" in result
|
|
|
|
def test_returns_code_refs_when_present(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_end_session(
|
|
session_id=sid, one_liner="With refs",
|
|
topics="test", outcome="Done", summary="Summary",
|
|
code_refs="bigmind/db.py, src/server.py",
|
|
)
|
|
result = memory_get_session_detail(sid)
|
|
assert "bigmind/db.py" in result
|
|
|
|
def test_returns_error_for_nonexistent_session(self, temp_db):
|
|
result = memory_get_session_detail("00000000-0000-0000-0000-000000000000")
|
|
assert "No detailed summary found" in result
|
|
|
|
def test_returns_error_for_session_without_tier2(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
sid = memory_store.create_session(user["id"])
|
|
result = memory_get_session_detail(sid)
|
|
assert "No detailed summary found" in result
|
|
|
|
|
|
# ── memory_search_chunks ───────────────────────────────────────────────────────
|
|
|
|
class TestMemorySearchChunks:
|
|
def test_returns_matching_results(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(session_id=sid, content="Chose WAL mode for SQLite concurrency")
|
|
result = memory_search_chunks("WAL mode")
|
|
assert "WAL mode" in result
|
|
assert "Result 1" in result
|
|
|
|
def test_returns_no_results_message_when_empty(self, temp_db):
|
|
result = memory_search_chunks("xyzzy_term_that_will_never_exist_42")
|
|
assert "No memory chunks found" in result
|
|
|
|
def test_result_count_in_header(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(session_id=sid, content="alpha beta gamma")
|
|
result = memory_search_chunks("alpha beta")
|
|
assert "1 result" in result
|
|
|
|
def test_respects_limit(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
for i in range(5):
|
|
memory_flag_important(session_id=sid, content=f"unique chunk item {i}")
|
|
result = memory_search_chunks("unique chunk item", limit=2)
|
|
assert "2 results" in result
|
|
|
|
def test_isolated_to_current_user(self, temp_db):
|
|
"""Chunks from a different user must not appear in search results."""
|
|
user = memory_store.get_or_create_user("testuser")
|
|
other_user = memory_store.get_or_create_user("otheruser")
|
|
other_sid = memory_store.create_session(other_user["id"])
|
|
memory_store.append_chunk(other_sid, other_user["id"], "user", "secret other data")
|
|
result = memory_search_chunks("secret other data")
|
|
assert "No memory chunks found" in result
|
|
|
|
|
|
# ── memory_list_sessions ───────────────────────────────────────────────────────
|
|
|
|
class TestMemoryListSessions:
|
|
def test_lists_closed_sessions(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
_close(sid, headline="Listed session headline")
|
|
result = memory_list_sessions()
|
|
assert "Listed session headline" in result
|
|
|
|
def test_no_sessions_message_when_empty(self, temp_db):
|
|
result = memory_list_sessions()
|
|
assert "No sessions found" in result
|
|
|
|
def test_open_sessions_not_in_list(self, temp_db):
|
|
_start_and_get_id() # open session — should NOT appear
|
|
result = memory_list_sessions()
|
|
assert "No sessions found" in result
|
|
|
|
def test_topics_filter_includes_match(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
_close(sid, headline="Python backend work", topics="python,backend")
|
|
result = memory_list_sessions(topics_filter="python")
|
|
assert "Python backend work" in result
|
|
|
|
def test_topics_filter_excludes_non_match(self, temp_db):
|
|
sid1 = _start_and_get_id()
|
|
_close(sid1, headline="Python session", topics="python")
|
|
sid2 = _start_and_get_id()
|
|
_close(sid2, headline="Design session", topics="design,frontend")
|
|
result = memory_list_sessions(topics_filter="python")
|
|
assert "Python session" in result
|
|
assert "Design session" not in result
|
|
|
|
def test_topics_filter_no_match_returns_message(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
_close(sid, topics="test")
|
|
result = memory_list_sessions(topics_filter="nonexistenttopic")
|
|
assert "No sessions found" in result
|
|
|
|
def test_tier2_indicator_shown(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_end_session(
|
|
session_id=sid, one_liner="With detail",
|
|
topics="test", outcome="Done", summary="Full narrative.",
|
|
)
|
|
result = memory_list_sessions()
|
|
assert "📄" in result
|
|
|
|
|
|
# ── memory_store_fact ──────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryStoreFact:
|
|
def test_returns_confirmation(self, temp_db):
|
|
result = memory_store_fact(category="preference", fact="Prefers dark mode always")
|
|
assert "✅ Fact stored" in result
|
|
assert "preference" in result
|
|
assert "Prefers dark mode always" in result
|
|
|
|
def test_fact_appears_in_context(self, temp_db):
|
|
memory_store_fact(category="codebase", fact="All servers use FastMCP pattern")
|
|
result = memory_get_context()
|
|
assert "All servers use FastMCP pattern" in result
|
|
|
|
def test_category_shown_in_context(self, temp_db):
|
|
memory_store_fact(category="constraint", fact="Must support Python 3.12+")
|
|
result = memory_get_context()
|
|
assert "[constraint]" in result
|
|
|
|
def test_multiple_facts_all_shown(self, temp_db):
|
|
memory_store_fact(category="preference", fact="Uses uv for packaging")
|
|
memory_store_fact(category="decision", fact="Chose SQLite over DuckDB")
|
|
result = memory_get_context()
|
|
assert "Uses uv for packaging" in result
|
|
assert "Chose SQLite over DuckDB" in result
|
|
|
|
|
|
# ── memory_update_profile ──────────────────────────────────────────────────────
|
|
|
|
class TestMemoryUpdateProfile:
|
|
def test_returns_confirmation(self, temp_db):
|
|
result = memory_update_profile(role="Senior Engineer")
|
|
assert "✅ Identity profile updated" in result
|
|
|
|
def test_role_appears_in_context(self, temp_db):
|
|
memory_update_profile(role="Principal Engineer")
|
|
result = memory_get_context()
|
|
assert "Principal Engineer" in result
|
|
|
|
def test_preferences_appear_in_context(self, temp_db):
|
|
memory_update_profile(preferences="Python first, no unnecessary abstractions")
|
|
result = memory_get_context()
|
|
assert "Python first" in result
|
|
|
|
def test_pinned_facts_appear_in_context(self, temp_db):
|
|
memory_update_profile(pinned_facts="- Always uses uv\n- macOS only")
|
|
result = memory_get_context()
|
|
assert "Always uses uv" in result
|
|
|
|
def test_partial_update_preserves_existing_fields(self, temp_db):
|
|
memory_update_profile(role="Engineer", preferences="Concise code")
|
|
memory_update_profile(pinned_facts="- New pinned fact") # only update pinned_facts
|
|
result = memory_get_context()
|
|
assert "Engineer" in result # role preserved
|
|
assert "Concise code" in result # preferences preserved
|
|
assert "New pinned fact" in result # new field added
|
|
|
|
|
|
# ── memory_append_chunk ────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryAppendChunk:
|
|
def test_returns_confirmation(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result = memory_append_chunk(session_id=sid, content="Manually saved content")
|
|
assert "✅ Chunk stored" in result
|
|
|
|
def test_chunk_is_searchable(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_append_chunk(session_id=sid, content="Manually appended important data")
|
|
result = memory_search_chunks("Manually appended")
|
|
assert "Manually appended" in result
|
|
|
|
def test_flag_reason_stored(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_append_chunk(
|
|
session_id=sid, content="Some important note",
|
|
flag_reason="manual save by user",
|
|
)
|
|
user = memory_store.get_or_create_user("testuser")
|
|
chunks = memory_store.search_chunks(user["id"], "important note")
|
|
assert chunks[0]["flag_reason"] == "manual save by user"
|
|
|
|
|
|
# ── memory_get_stats ───────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryGetStats:
|
|
def test_returns_stats_markdown(self, temp_db):
|
|
result = memory_get_stats()
|
|
assert "📊 BigMind Stats" in result
|
|
assert "Sessions" in result
|
|
assert "Facts" in result
|
|
assert "Database size" in result
|
|
assert "Database path" in result
|
|
|
|
def test_session_count_is_accurate(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
_close(sid)
|
|
result = memory_get_stats()
|
|
assert "| Sessions | 1 |" in result
|
|
|
|
def test_chunk_count_is_accurate(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(session_id=sid, content="Counted chunk")
|
|
result = memory_get_stats()
|
|
assert "| Memory chunks (Tier 3) | 1 |" in result
|
|
|
|
def test_fact_count_is_accurate(self, temp_db):
|
|
memory_store_fact(category="test", fact="A counted fact")
|
|
result = memory_get_stats()
|
|
assert "| Facts | 1 |" in result
|
|
|
|
|
|
# ── memory_vacuum ──────────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryVacuum:
|
|
def test_returns_confirmation(self, temp_db):
|
|
result = memory_vacuum(older_than_days=90)
|
|
assert "✅ Removed" in result
|
|
assert "chunk(s)" in result
|
|
|
|
def test_removes_old_chunks(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
sid = memory_store.create_session(user["id"])
|
|
old_date = (datetime.now(timezone.utc) - timedelta(days=100)).isoformat()
|
|
with db() as conn:
|
|
cur = conn.execute(
|
|
"""INSERT INTO conversation_chunks
|
|
(session_id, user_id, role, content, flag_reason, seq, created_at)
|
|
VALUES (?,?,?,?,?,?,?)""",
|
|
(sid, user["id"], "user", "old stale content", "old", 1, old_date),
|
|
)
|
|
chunk_id = cur.lastrowid
|
|
conn.execute(
|
|
"INSERT INTO conversation_chunks_fts(rowid, content, flag_reason) VALUES(?,?,?)",
|
|
(chunk_id, "old stale content", "old"),
|
|
)
|
|
result = memory_vacuum(older_than_days=90)
|
|
assert "Removed 1 chunk(s)" in result
|
|
|
|
def test_preserves_recent_chunks(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(session_id=sid, content="Very recent important thing")
|
|
result = memory_vacuum(older_than_days=90)
|
|
assert "Removed 0 chunk(s)" in result
|
|
|
|
def test_old_chunk_not_searchable_after_vacuum(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
sid = memory_store.create_session(user["id"])
|
|
old_date = (datetime.now(timezone.utc) - timedelta(days=100)).isoformat()
|
|
with db() as conn:
|
|
cur = conn.execute(
|
|
"""INSERT INTO conversation_chunks
|
|
(session_id, user_id, role, content, flag_reason, seq, created_at)
|
|
VALUES (?,?,?,?,?,?,?)""",
|
|
(sid, user["id"], "user", "ancient secret data", "old", 1, old_date),
|
|
)
|
|
chunk_id = cur.lastrowid
|
|
conn.execute(
|
|
"INSERT INTO conversation_chunks_fts(rowid, content, flag_reason) VALUES(?,?,?)",
|
|
(chunk_id, "ancient secret data", "old"),
|
|
)
|
|
memory_vacuum(older_than_days=90)
|
|
result = memory_search_chunks("ancient secret data")
|
|
assert "No memory chunks found" in result
|
|
|
|
|
|
# ── memory_get_instructions ────────────────────────────────────────────────────
|
|
|
|
class TestMemoryGetInstructions:
|
|
def test_returns_string(self, temp_db):
|
|
result = memory_get_instructions()
|
|
assert isinstance(result, str)
|
|
assert len(result) > 0
|
|
|
|
def test_contains_start_session_directive(self, temp_db):
|
|
result = memory_get_instructions()
|
|
assert "memory_start_session" in result
|
|
|
|
def test_contains_end_session_directive(self, temp_db):
|
|
result = memory_get_instructions()
|
|
assert "memory_end_session" in result
|
|
|
|
def test_contains_mandatory_language(self, temp_db):
|
|
result = memory_get_instructions()
|
|
assert "Always" in result
|
|
|
|
|
|
# ── memory_health_check ────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryHealthCheck:
|
|
def test_returns_health_report_header(self, temp_db):
|
|
result = memory_health_check()
|
|
assert "🩺 BigMind Health Check" in result
|
|
|
|
def test_fts_in_sync_shown_on_clean_db(self, temp_db):
|
|
result = memory_health_check()
|
|
assert "FTS index" in result
|
|
assert "✅" in result
|
|
|
|
def test_no_warnings_on_clean_db(self, temp_db):
|
|
result = memory_health_check()
|
|
assert "⚠️" not in result
|
|
|
|
def test_flags_stale_facts(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
fid = memory_store.store_fact(user["id"], "codebase", "Stale old technology note")
|
|
old_date = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat()
|
|
with db() as conn:
|
|
conn.execute("UPDATE facts SET updated_at=? WHERE id=?", (old_date, fid))
|
|
result = memory_health_check(stale_days=30)
|
|
assert "Stale facts" in result
|
|
assert "Stale old technology note" in result
|
|
|
|
def test_fresh_facts_not_flagged_as_stale(self, temp_db):
|
|
memory_store_fact(category="preference", fact="Very fresh preference")
|
|
result = memory_health_check(stale_days=30)
|
|
assert "Stale facts: 0" not in result
|
|
# The "✅ Facts freshness" line should appear
|
|
assert "Facts freshness" in result
|
|
|
|
def test_flags_sessions_without_summary(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
sid = memory_store.create_session(user["id"])
|
|
memory_store.close_session(sid, "Session with no narrative")
|
|
result = memory_health_check()
|
|
assert "Sessions without Tier-2 summary" in result
|
|
|
|
def test_no_warning_when_all_sessions_have_summary(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
_close(sid) # _close calls memory_end_session which saves a Tier-2 summary
|
|
result = memory_health_check()
|
|
assert "Sessions without Tier-2 summary" not in result
|
|
|
|
def test_flags_low_confidence_facts(self, temp_db):
|
|
user = memory_store.get_or_create_user("testuser")
|
|
memory_store.store_fact(user["id"], "codebase", "Uncertain technology choice", confidence=0.4)
|
|
result = memory_health_check()
|
|
assert "Low-confidence facts" in result
|
|
assert "Uncertain technology choice" in result
|
|
|
|
def test_open_sessions_listed(self, temp_db):
|
|
_start_and_get_id() # creates an open session
|
|
result = memory_health_check()
|
|
assert "Open sessions" in result
|
|
|
|
def test_default_stale_days_is_30(self, temp_db):
|
|
result = memory_health_check()
|
|
# Either "30 days" in the stale line or the clean version
|
|
assert "30" in result
|
|
|
|
|
|
# ── memory_export ──────────────────────────────────────────────────────────────
|
|
|
|
class TestMemoryExport:
|
|
def test_returns_confirmation(self, temp_db, tmp_path):
|
|
out = str(tmp_path / "test_export.json")
|
|
result = memory_export(output_path=out)
|
|
assert "✅" in result
|
|
assert "BigMind memory exported" in result
|
|
|
|
def test_shows_file_path_in_result(self, temp_db, tmp_path):
|
|
out = str(tmp_path / "test_export.json")
|
|
result = memory_export(output_path=out)
|
|
assert out in result
|
|
|
|
def test_file_created_on_disk(self, temp_db, tmp_path):
|
|
out = str(tmp_path / "test_export.json")
|
|
memory_export(output_path=out)
|
|
assert Path(out).exists()
|
|
|
|
def test_result_contains_count_rows(self, temp_db, tmp_path):
|
|
out = str(tmp_path / "test_export.json")
|
|
result = memory_export(output_path=out)
|
|
assert "| **Facts** |" in result
|
|
assert "| **Sessions** |" in result
|
|
assert "| **Chunks (Tier 3)** |" in result
|
|
assert "| **File size** |" in result
|
|
|
|
def test_exports_facts(self, temp_db, tmp_path):
|
|
memory_store_fact(category="preference", fact="Exported preference via tool")
|
|
out = str(tmp_path / "test_export.json")
|
|
memory_export(output_path=out)
|
|
data = json.loads(Path(out).read_text())
|
|
assert data["stats"]["facts_count"] == 1
|
|
assert any("Exported preference via tool" in f["fact"] for f in data["facts"])
|
|
|
|
def test_exports_sessions_and_summaries(self, temp_db, tmp_path):
|
|
sid = _start_and_get_id()
|
|
_close(sid, headline="Session for export")
|
|
out = str(tmp_path / "test_export.json")
|
|
memory_export(output_path=out)
|
|
data = json.loads(Path(out).read_text())
|
|
assert data["stats"]["sessions_count"] >= 1
|
|
matches = [s for s in data["sessions"] if s.get("one_liner") == "Session for export"]
|
|
assert len(matches) == 1
|
|
|
|
def test_exports_chunks(self, temp_db, tmp_path):
|
|
sid = _start_and_get_id()
|
|
memory_flag_important(session_id=sid, content="Chunk to be exported")
|
|
out = str(tmp_path / "test_export.json")
|
|
memory_export(output_path=out)
|
|
data = json.loads(Path(out).read_text())
|
|
assert data["stats"]["chunks_count"] == 1
|
|
assert any("Chunk to be exported" in c["content"] for c in data["conversation_chunks"])
|
|
|
|
def test_valid_json_output(self, temp_db, tmp_path):
|
|
out = str(tmp_path / "test_export.json")
|
|
memory_export(output_path=out)
|
|
# Should not raise
|
|
data = json.loads(Path(out).read_text())
|
|
assert isinstance(data, dict)
|
|
|
|
def test_export_date_present(self, temp_db, tmp_path):
|
|
out = str(tmp_path / "test_export.json")
|
|
memory_export(output_path=out)
|
|
data = json.loads(Path(out).read_text())
|
|
assert "export_date" in data
|
|
assert "bigmind_schema_version" in data
|
|
assert isinstance(data["bigmind_schema_version"], int)
|
|
|
|
|
|
# ── memory_deprecate_fact ──────────────────────────────────────────────────────
|
|
|
|
class TestMemoryDeprecateFact:
|
|
def test_returns_confirmation(self, temp_db):
|
|
result_store = memory_store_fact(category="codebase", fact="Old stack fact")
|
|
fid = int(re.search(r"id: (\d+)", result_store).group(1))
|
|
result = memory_deprecate_fact(fact_id=fid, reason="Technology replaced")
|
|
assert "✅" in result
|
|
assert str(fid) in result
|
|
|
|
def test_reason_shown_in_confirmation(self, temp_db):
|
|
result_store = memory_store_fact(category="decision", fact="Used Gradle")
|
|
fid = int(re.search(r"id: (\d+)", result_store).group(1))
|
|
result = memory_deprecate_fact(fact_id=fid, reason="Switched to Maven")
|
|
assert "Switched to Maven" in result
|
|
|
|
def test_no_reason_still_succeeds(self, temp_db):
|
|
result_store = memory_store_fact(category="preference", fact="Some old pref")
|
|
fid = int(re.search(r"id: (\d+)", result_store).group(1))
|
|
result = memory_deprecate_fact(fact_id=fid)
|
|
assert "✅" in result
|
|
|
|
def test_deprecated_fact_absent_from_context(self, temp_db):
|
|
result_store = memory_store_fact(category="codebase", fact="Outdated deployment detail")
|
|
fid = int(re.search(r"id: (\d+)", result_store).group(1))
|
|
# Confirm it's in context before deprecation
|
|
assert "Outdated deployment detail" in memory_get_context()
|
|
# Deprecate it
|
|
memory_deprecate_fact(fact_id=fid, reason="No longer true")
|
|
# Must be gone from context
|
|
assert "Outdated deployment detail" not in memory_get_context()
|
|
|
|
def test_other_facts_unaffected_by_deprecation(self, temp_db):
|
|
r1 = memory_store_fact(category="preference", fact="Keep this preference")
|
|
r2 = memory_store_fact(category="codebase", fact="Drop this codebase note")
|
|
fid_drop = int(re.search(r"id: (\d+)", r2).group(1))
|
|
memory_deprecate_fact(fact_id=fid_drop)
|
|
ctx = memory_get_context()
|
|
assert "Keep this preference" in ctx
|
|
assert "Drop this codebase note" not in ctx
|
|
|
|
def test_nonexistent_fact_returns_error(self, temp_db):
|
|
result = memory_deprecate_fact(fact_id=99999)
|
|
assert "❌" in result
|
|
assert "99999" in result
|
|
|
|
def test_health_check_does_not_flag_deprecated_facts_as_stale(self, temp_db):
|
|
"""Deprecated facts should not surface as actionable stale warnings."""
|
|
user = memory_store.get_or_create_user("testuser")
|
|
fid = memory_store.store_fact(user["id"], "codebase", "Will be deprecated")
|
|
from datetime import timedelta
|
|
old_date = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat()
|
|
from bigmind.db import db as _db
|
|
with _db() as conn:
|
|
conn.execute("UPDATE facts SET updated_at=? WHERE id=?", (old_date, fid))
|
|
memory_deprecate_fact(fact_id=fid, reason="Removed intentionally")
|
|
result = memory_health_check(stale_days=30)
|
|
assert "Will be deprecated" not in result
|
|
|
|
|
|
# ── memory_add_hypothesis / resolve / list ─────────────────────────────────────
|
|
|
|
class TestMemoryHypotheses:
|
|
def test_add_returns_confirmation(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result = memory_add_hypothesis(
|
|
session_id=sid,
|
|
hypothesis="I believe the issue is in the connection pool",
|
|
)
|
|
assert "💭 Hypothesis recorded" in result
|
|
assert "connection pool" in result
|
|
|
|
def test_add_shows_confidence_percent(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result = memory_add_hypothesis(
|
|
session_id=sid,
|
|
hypothesis="High confidence belief",
|
|
confidence=0.9,
|
|
)
|
|
assert "90%" in result
|
|
|
|
def test_add_returns_id(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result = memory_add_hypothesis(session_id=sid, hypothesis="Some thought")
|
|
assert re.search(r"id: \d+", result)
|
|
|
|
def test_list_empty_returns_message(self, temp_db):
|
|
result = memory_list_hypotheses()
|
|
assert "No hypotheses found" in result
|
|
|
|
def test_list_shows_open_hypotheses(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_add_hypothesis(session_id=sid, hypothesis="The cache is stale")
|
|
result = memory_list_hypotheses()
|
|
assert "The cache is stale" in result
|
|
assert "💭" in result
|
|
|
|
def test_list_filter_by_status(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_add_hypothesis(session_id=sid, hypothesis="Will be confirmed soon")
|
|
memory_add_hypothesis(session_id=sid, hypothesis="Will stay open")
|
|
user = memory_store.get_or_create_user("testuser")
|
|
hyps = memory_store.list_hypotheses(user["id"])
|
|
hid = next(h["id"] for h in hyps if "confirmed" in h["hypothesis"])
|
|
memory_resolve_hypothesis(hypothesis_id=hid, status="confirmed", resolution="Yes it was true")
|
|
open_result = memory_list_hypotheses(status="open")
|
|
confirmed_result = memory_list_hypotheses(status="confirmed")
|
|
assert "Will stay open" in open_result
|
|
assert "Will be confirmed soon" not in open_result
|
|
assert "Will be confirmed soon" in confirmed_result
|
|
|
|
def test_resolve_confirmed_shows_checkmark(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result_add = memory_add_hypothesis(session_id=sid, hypothesis="Bug in serializer")
|
|
hid = int(re.search(r"id: (\d+)", result_add).group(1))
|
|
result = memory_resolve_hypothesis(
|
|
hypothesis_id=hid,
|
|
status="confirmed",
|
|
resolution="Confirmed — null check was missing in BVV serializer",
|
|
)
|
|
assert "✅" in result
|
|
assert "confirmed" in result
|
|
|
|
def test_resolve_refuted_shows_cross(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result_add = memory_add_hypothesis(session_id=sid, hypothesis="Network latency")
|
|
hid = int(re.search(r"id: (\d+)", result_add).group(1))
|
|
result = memory_resolve_hypothesis(
|
|
hypothesis_id=hid,
|
|
status="refuted",
|
|
resolution="Was a race condition, not network",
|
|
)
|
|
assert "❌" in result
|
|
assert "refuted" in result
|
|
|
|
def test_resolve_abandoned_shows_icon(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result_add = memory_add_hypothesis(session_id=sid, hypothesis="Might be cache")
|
|
hid = int(re.search(r"id: (\d+)", result_add).group(1))
|
|
result = memory_resolve_hypothesis(hypothesis_id=hid, status="abandoned")
|
|
assert "🚫" in result
|
|
assert "abandoned" in result
|
|
|
|
def test_resolve_shows_resolution_text(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result_add = memory_add_hypothesis(session_id=sid, hypothesis="A theory")
|
|
hid = int(re.search(r"id: (\d+)", result_add).group(1))
|
|
result = memory_resolve_hypothesis(
|
|
hypothesis_id=hid, status="confirmed", resolution="The theory held up"
|
|
)
|
|
assert "The theory held up" in result
|
|
|
|
def test_resolve_invalid_status_returns_error(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result_add = memory_add_hypothesis(session_id=sid, hypothesis="Some belief")
|
|
hid = int(re.search(r"id: (\d+)", result_add).group(1))
|
|
result = memory_resolve_hypothesis(hypothesis_id=hid, status="maybe")
|
|
assert "❌" in result
|
|
|
|
def test_resolve_nonexistent_id_returns_error(self, temp_db):
|
|
result = memory_resolve_hypothesis(hypothesis_id=99999, status="confirmed")
|
|
assert "❌" in result
|
|
assert "99999" in result
|
|
|
|
def test_list_shows_resolution_when_resolved(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
result_add = memory_add_hypothesis(session_id=sid, hypothesis="Root cause is threading")
|
|
hid = int(re.search(r"id: (\d+)", result_add).group(1))
|
|
memory_resolve_hypothesis(
|
|
hypothesis_id=hid, status="confirmed",
|
|
resolution="Thread-local storage was the culprit"
|
|
)
|
|
result = memory_list_hypotheses()
|
|
assert "Thread-local storage was the culprit" in result
|
|
|
|
def test_list_status_filter_no_match_returns_message(self, temp_db):
|
|
sid = _start_and_get_id()
|
|
memory_add_hypothesis(session_id=sid, hypothesis="Open thought")
|
|
result = memory_list_hypotheses(status="confirmed")
|
|
assert "No hypotheses found" in result
|
|
|
|
|
|
|
|
|
|
|