From dba68a219f698fb526940868b5110cddcf8f0cda Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:07:08 +0300 Subject: [PATCH 1/8] test: delete 7 duplicate test files (-38 tests) Deleted exact copies of test_integration.py and test_auth_backup.py: - test_all.py: duplicates test_core + test_features + test_mcp - test_mcp/test_mcp.py: exact copy of test_auth_backup.py - test_lifecycle/test_lifecycle.py: copies test_integration.py - test_hooks/test_hooks.py: copies test_integration.py - test_graph/test_graph.py: copies test_integration.py - test_rag/test_rag.py: copies test_integration.py - test_rag/test_rag_edge_cases.py: overlap with test_integration + search_facade --- tests/test_all.py | 125 ---------------------- tests/test_graph/test_graph.py | 63 ----------- tests/test_hooks/test_hooks.py | 48 --------- tests/test_lifecycle/test_lifecycle.py | 39 ------- tests/test_mcp/test_mcp.py | 23 ---- tests/test_rag/test_rag.py | 79 -------------- tests/test_rag/test_rag_edge_cases.py | 142 ------------------------- 7 files changed, 519 deletions(-) delete mode 100644 tests/test_all.py delete mode 100644 tests/test_graph/test_graph.py delete mode 100644 tests/test_hooks/test_hooks.py delete mode 100644 tests/test_lifecycle/test_lifecycle.py delete mode 100644 tests/test_mcp/test_mcp.py delete mode 100644 tests/test_rag/test_rag.py delete mode 100644 tests/test_rag/test_rag_edge_cases.py diff --git a/tests/test_all.py b/tests/test_all.py deleted file mode 100644 index c76fe81b..00000000 --- a/tests/test_all.py +++ /dev/null @@ -1,125 +0,0 @@ -"""Basic tests for mcp-ariel-memory (async).""" - -import asyncio -import sys -from pathlib import Path - -import pytest - -sys.path.insert(0, str(Path(__file__).parent.parent)) - - -@pytest.fixture(autouse=True, scope="session") -def run_migrations(): - """Ensure migrations run before any test.""" - - async def _setup(): - from shared.migrations import migration_manager - - await migration_manager.migrate() - - asyncio.run(_setup()) - - -def test_mcp_tools_count(): - from mcp_server import mcp - - tools = mcp._tool_manager.list_tools() - assert len(tools) >= 15 - - -def test_mcp_tools_are_async(): - import inspect - - from mcp_server import mcp - - tools = mcp._tool_manager.list_tools() - for tool in tools: - assert inspect.iscoroutinefunction(tool.fn), f"{tool.name} is not async" - - -def test_user_remember_recall(): - from core import memory_manager - - mm = memory_manager - - async def t(): - await mm.user_memory("test_user").remember("lang", "Python", 0.8) - results = await mm.user_memory("test_user").recall("lang") - assert len(results) > 0 - assert results[0]["key"] == "lang" - - asyncio.run(t()) - - -def test_agent_remember_recall(): - from core import memory_manager - - mm = memory_manager - - async def t(): - await mm.agent_memory("test_agent").remember("rule", "YAGNI", 0.9) - results = await mm.agent_memory("test_agent").recall("rule") - assert len(results) > 0 - - asyncio.run(t()) - - -def test_rag_engine(): - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test") - await rag.ingest_text("Test Page", "Python is great for AI", user_id="test") - results = await rag.search("Python", user_id="test") - assert len(results) > 0 - - asyncio.run(t()) - - -def test_epistemic_graph(): - from graph.epistemic import EpistemicGraph - - async def t(): - g = EpistemicGraph(layer="test") - await g.init_db() - n = await g.add_node("test", "Likes Python", "fact", ["fact_about_user"]) - nodes = await g.query_by_tag("test", "fact_about_user") - assert len(nodes) >= 1 - - asyncio.run(t()) - - -def test_user_wiki(): - from wiki.manager import WikiManager - - async def t(): - w = WikiManager(layer="user") - path = await w.add("work_notes", "Day 1", "Started project") - assert path is not None - results = await w.search("project") - assert len(results) > 0 - - asyncio.run(t()) - - -def test_audit_trail(): - from features.audit_trail import AuditTrail - - async def t(): - at = AuditTrail() - await at.log("test", "test_action") - history = await at.get_history("test") - assert len(history) >= 1 - - asyncio.run(t()) - - -def test_cache(): - from shared.cache import MemoryCache - - mc = MemoryCache(max_size=5, ttl=60) - mc.set("key", "value") - assert mc.get("key") == "value" - mc.delete("key") - assert mc.get("key") is None diff --git a/tests/test_graph/test_graph.py b/tests/test_graph/test_graph.py deleted file mode 100644 index 6d1bda4a..00000000 --- a/tests/test_graph/test_graph.py +++ /dev/null @@ -1,63 +0,0 @@ -"""Tests for graph/ module — async.""" - -import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -def test_epistemic_add_query(): - from graph.epistemic import EpistemicGraph - - async def t(): - g = EpistemicGraph(layer="test_graph") - await g.init_db() - n = await g.add_node("t", "Likes Python", "fact", ["fact_about_user"], 0.9) - assert n > 0 - nodes = await g.query_by_tag("t", "fact_about_user") - assert len(nodes) >= 1 - - asyncio.run(t()) - - -def test_epistemic_neighbors(): - from graph.epistemic import EpistemicGraph - - async def t(): - g = EpistemicGraph(layer="test_graph") - n1 = await g.add_node("t", "A", "fact") - n2 = await g.add_node("t", "B", "fact") - await g.add_edge(n1, n2, "related") - neighbors = await g.get_neighbors(n1) - assert len(neighbors) >= 1 - - asyncio.run(t()) - - -def test_epistemic_find_path(): - from graph.epistemic import EpistemicGraph - - async def t(): - g = EpistemicGraph(layer="test_graph") - n1 = await g.add_node("t", "Start", "fact") - n2 = await g.add_node("t", "End", "fact") - await g.add_edge(n1, n2, "leads_to") - path = await g.find_path(n1, n2) - assert len(path) >= 1 - - asyncio.run(t()) - - -def test_temporal_timeline(): - from graph.temporal import TemporalGraph - - async def t(): - tg = TemporalGraph() - e1 = await tg.add_event("t", "msg", "hello") - e2 = await tg.add_event("t", "resp", "hi") - await tg.link_events(e1, e2, "follows") - timeline = await tg.get_timeline("t") - assert len(timeline) >= 2 - - asyncio.run(t()) diff --git a/tests/test_hooks/test_hooks.py b/tests/test_hooks/test_hooks.py deleted file mode 100644 index 3fc34d0c..00000000 --- a/tests/test_hooks/test_hooks.py +++ /dev/null @@ -1,48 +0,0 @@ -"""Tests for hooks/ module.""" - -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -def test_hook_registry(): - from hooks.registry import HookRegistry - - hr = HookRegistry() - hr.register("test_hook", lambda ctx: {"ok": True}) - result = hr.fire("test_hook", "user", {"data": 1}) - assert result["handler_count"] == 1 - - -def test_user_hooks_importance(): - from hooks.user_hooks import UserHooks - - uh = UserHooks("test_hooks") - r = uh._importance_gate({"text": "How do I configure Redis?"}) - assert r["importance"] > 0.3 - - -def test_agent_hooks_error(): - import asyncio - from graph.epistemic import EpistemicGraph - from hooks.agent_hooks import AgentHooks - - eg = EpistemicGraph(layer="agent") - asyncio.run(eg.init_db()) - - ah = AgentHooks("test_hooks") - r = ah._error_occurred({"error": "NullPointerException"}) - assert "node_id" in r - - -def test_agent_hooks_importance_gate(): - from hooks.agent_hooks import AgentHooks - - ah = AgentHooks("test_hooks") - r = ah._importance_gate({"text": "error in database connection"}) - assert r["importance"] > 0.3 - assert "bypass" in r - - r2 = ah._importance_gate({"text": ""}) - assert r2["bypass"] is True diff --git a/tests/test_lifecycle/test_lifecycle.py b/tests/test_lifecycle/test_lifecycle.py deleted file mode 100644 index 09331b97..00000000 --- a/tests/test_lifecycle/test_lifecycle.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Tests for lifecycle/ module — async.""" - -import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -def test_forgetting_cleanup(): - from lifecycle.forgetting import ForgettingSystem - - async def t(): - fs = ForgettingSystem() - stats = await fs.cleanup() - assert "archived" in stats - - asyncio.run(t()) - - -def test_emotion_trigger(): - from lifecycle.emotion_trigger import EmotionTrigger - - et = EmotionTrigger() - should, reason, weight = et.should_save("I love this!") - assert should is True - should2, _, _ = et.should_save("ok") - assert should2 is False - - -def test_consolidation(): - from lifecycle.consolidation import ConsolidationEngine - - async def t(): - ce = ConsolidationEngine() - result = await ce.consolidate_staging("test_lc", [{"content": "test", "importance": 0.9}], 0.7) - assert result["promoted"] == 1 - - asyncio.run(t()) diff --git a/tests/test_mcp/test_mcp.py b/tests/test_mcp/test_mcp.py deleted file mode 100644 index f09da601..00000000 --- a/tests/test_mcp/test_mcp.py +++ /dev/null @@ -1,23 +0,0 @@ -"""Tests for MCP server and tools — async.""" - -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -def test_mcp_tools_count(): - from mcp_server import mcp - - tools = mcp._tool_manager.list_tools() - assert len(tools) >= 15 - - -def test_mcp_tools_are_async(): - import inspect - - from mcp_server import mcp - - tools = mcp._tool_manager.list_tools() - for tool in tools: - assert inspect.iscoroutinefunction(tool.fn), f"{tool.name} is not async" diff --git a/tests/test_rag/test_rag.py b/tests/test_rag/test_rag.py deleted file mode 100644 index b0467c08..00000000 --- a/tests/test_rag/test_rag.py +++ /dev/null @@ -1,79 +0,0 @@ -"""Tests for rag/ module — async.""" - -import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -# Ensure migrations run -async def _setup(): - from shared.migrations import migration_manager - - await migration_manager.migrate() - - -asyncio.run(_setup()) - - -def test_rag_ingest_search(): - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_rag2") - eid = await rag.ingest_text("Unique Python Guide 2026", "Python is great for AI", user_id="t2") - assert eid > 0 - results = await rag.search("Unique Python", user_id="t2") - assert len(results) > 0 - - asyncio.run(t()) - - -def test_rag_relations(): - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_rag_r") - eid1 = await rag.ingest_text("Page A", "Content A", user_id="t") - eid2 = await rag.ingest_text("Page B", "Content B", user_id="t") - await rag.add_relation(eid1, eid2, "related") - rels = await rag.get_relations(eid1) - assert len(rels) >= 1 - - asyncio.run(t()) - - -def test_rag_hybrid(): - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_rrf2") - await rag.ingest_text("A", "Content A", user_id="t") - await rag.ingest_text("B", "Content B", user_id="t") - results = await rag.search("Content", user_id="t", strategy="hybrid", limit=3) - assert len(results) > 0 - - asyncio.run(t()) - - -def test_retrieval_router(): - from rag.router import RetrievalRouter - - async def t(): - r = RetrievalRouter(user_id="t") - result = await r.route("Python docs") - assert result.strategy is not None - - asyncio.run(t()) - - -def test_conflict_resolver(): - from rag.conflict import ConflictResolver - - async def t(): - cr = ConflictResolver() - r = await cr.check("t", "Test content here") - assert "is_conflict" in r - - asyncio.run(t()) diff --git a/tests/test_rag/test_rag_edge_cases.py b/tests/test_rag/test_rag_edge_cases.py deleted file mode 100644 index 0af780bb..00000000 --- a/tests/test_rag/test_rag_edge_cases.py +++ /dev/null @@ -1,142 +0,0 @@ -"""Edge case tests for rag/engine.py — timeout, empty, corrupt, dedup.""" - -import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -async def _setup(): - from shared.migrations import migration_manager - - await migration_manager.migrate() - - -asyncio.run(_setup()) - - -def test_rag_empty_query(): - """Search with empty string should not crash.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_edge") - results = await rag.search("", user_id="edge_test") - assert isinstance(results, list) - - asyncio.run(t()) - - -def test_rag_dedup(): - """Ingesting same text twice should return existing page_id.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_dedup") - eid1 = await rag.ingest_text("Dedup Test", "Unique content for dedup", user_id="dedup") - eid2 = await rag.ingest_text("Dedup Test", "Unique content for dedup", user_id="dedup") - assert eid1 == eid2 - - asyncio.run(t()) - - -def test_rag_search_no_results(): - """Search for nonexistent content should return empty list.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_noresults") - results = await rag.search("xyznonexistentquery12345", user_id="noresults") - assert isinstance(results, list) - assert len(results) == 0 - - asyncio.run(t()) - - -def test_rag_count_pages(): - """count_pages should return correct count.""" - from rag.engine import RAGEngine - - async def t(): - import uuid - - uid = "count_" + uuid.uuid4().hex[:8] - rag = RAGEngine(layer="test_count") - before = await rag.count_pages(user_id=uid) - await rag.ingest_text("Count Page 1", "Content 1", user_id=uid) - await rag.ingest_text("Count Page 2", "Content 2", user_id=uid) - after = await rag.count_pages(user_id=uid) - assert after >= before + 2 - - asyncio.run(t()) - - -def test_rag_count_chunks(): - """count_chunks should return integer >= 0.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_chunks") - count = await rag.count_chunks() - assert isinstance(count, int) - assert count >= 0 - - asyncio.run(t()) - - -def test_rag_ingest_file(): - """ingest_file should handle a real file.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_file") - # Create temp file - tmp = Path("/tmp/test_rag_edge.txt") - tmp.write_text("Test file content for RAG edge case", encoding="utf-8") - result = await rag.ingest_file(tmp, user_id="file_test") - assert "[OK]" in result or "[SKIP]" in result - tmp.unlink(missing_ok=True) - - asyncio.run(t()) - - -def test_rag_strategy_auto(): - """Auto strategy should pick fts or hybrid based on query length.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_auto", search_strategy="auto") - await rag.ingest_text("Auto Test", "Some content", user_id="auto") - results = await rag.search("hi", user_id="auto") - assert isinstance(results, list) - - asyncio.run(t()) - - -def test_rag_relations_empty(): - """get_relations with no relations should return empty list.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_rels_empty") - eid = await rag.ingest_text("No Relations", "Content", user_id="rels") - rels = await rag.get_relations(eid) - assert isinstance(rels, list) - assert len(rels) == 0 - - asyncio.run(t()) - - -def test_rag_search_limit(): - """Search with limit=1 should return at most 1 result.""" - from rag.engine import RAGEngine - - async def t(): - rag = RAGEngine(layer="test_limit") - for i in range(5): - await rag.ingest_text(f"Limit Page {i}", f"Content {i}", user_id="limit") - results = await rag.search("Content", user_id="limit", limit=1) - assert len(results) <= 1 - - asyncio.run(t()) From b870109151d84b8469ae012f715930a64911fd0c Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:08:14 +0300 Subject: [PATCH 2/8] test: reduce test_core and test_features to unique tests only (-9 tests) Kept: test_reflex_buffer (unique L1 buffer test), test_compression (unique feature) Removed: 5 core tests + 5 feature tests that duplicated test_integration.py --- tests/test_core/test_core.py | 76 +--------------------------- tests/test_features/test_features.py | 65 +----------------------- 2 files changed, 3 insertions(+), 138 deletions(-) diff --git a/tests/test_core/test_core.py b/tests/test_core/test_core.py index 7476b3e4..e1ac3d7d 100644 --- a/tests/test_core/test_core.py +++ b/tests/test_core/test_core.py @@ -1,83 +1,11 @@ -"""Tests for core/ module (L1-L4) — async.""" +"""Tests for core/ module — unique tests only.""" -import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -def test_user_remember_recall(): - from core import memory_manager - - async def t(): - await memory_manager.user_memory("test_core").remember("lang", "Python", 0.8) - results = await memory_manager.user_memory("test_core").recall("lang") - assert len(results) > 0 - assert results[0]["key"] == "lang" - - asyncio.run(t()) - - -def test_agent_remember_recall(): - from core import memory_manager - - async def t(): - await memory_manager.agent_memory("test_core").remember("rule", "YAGNI", 0.9) - results = await memory_manager.agent_memory("test_core").recall("rule") - assert len(results) > 0 - - asyncio.run(t()) +from core.reflex import ReflexBuffer def test_reflex_buffer(): - from core.reflex import ReflexBuffer - buf = ReflexBuffer(max_size=5) buf.add(role="user", content="Hello", tokens=5) buf.add(role="assistant", content="Hi", tokens=3) assert buf.size() == 2 assert buf.get_recent(1)[0].content == "Hi" - - -def test_session_store(): - from core.session import SessionStore - - async def t(): - ss = SessionStore() - sid = await ss.create_session("test_core") - assert sid is not None - await ss.close_session(sid, summary="Test session") - assert await ss.count_sessions("test_core") >= 1 - - asyncio.run(t()) - - -def test_episodic_memory(): - from core.episodic import EpisodicMemory - - async def t(): - ep = EpisodicMemory() - eid = await ep.save("test_core", "Test episode", 0.8, ["tag1"]) - assert eid > 0 - episodes = await ep.search_by_tag("test_core", "tag1") - assert len(episodes) >= 1 - - asyncio.run(t()) - - -def test_core_memory(): - from core.memory import CoreMemory - - async def t(): - cm = CoreMemory() - await cm.save("test_core", "key1", "value1", 0.9) - entry = await cm.get("test_core", "key1") - assert entry is not None - assert entry.value == "value1" - results = await cm.search("test_core", "value1") - assert len(results) > 0 - assert await cm.delete("test_core", "key1") - assert await cm.get("test_core", "key1") is None - - asyncio.run(t()) diff --git a/tests/test_features/test_features.py b/tests/test_features/test_features.py index 96c494a2..d3f47dc6 100644 --- a/tests/test_features/test_features.py +++ b/tests/test_features/test_features.py @@ -1,69 +1,6 @@ -"""Tests for features/ module — async.""" +"""Tests for features/ module — unique tests only.""" import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent.parent)) - - -def test_audit_trail(): - from features.audit_trail import AuditTrail - - async def t(): - at = AuditTrail() - await at.log("test_feat", "test_action") - history = await at.get_history("test_feat") - assert len(history) >= 1 - - asyncio.run(t()) - - -def test_audit_rotation(): - from features.audit_trail import AuditTrail - - async def t(): - at = AuditTrail() - await at.log("test_rot", "action") - result = await at.cleanup_old(retention_days=0) - assert result >= 0 - - asyncio.run(t()) - - -def test_rate_limiter(): - from features.rate_limiting import RateLimiter - - async def t(): - rl = RateLimiter() - r = await rl.check("test_feat") - assert r["allowed"] is True - stats = await rl.get_stats("test_feat") - assert "requests_last_minute" in stats - - asyncio.run(t()) - - -def test_backup(): - from features.backup import BackupManager - - async def t(): - bm = BackupManager() - path = await bm.backup("test_feat") - assert path is not None - - asyncio.run(t()) - - -def test_import_export(): - from features.import_export import ImportExport - - async def t(): - ie = ImportExport() - path = await ie.export_user("test_feat") - assert path is not None - - asyncio.run(t()) def test_compression(): From a93ddae311647e1f3f666039a2f724d6705d4730 Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:09:52 +0300 Subject: [PATCH 3/8] test: reduce test_auth_backup to unique auth + MCP + config tests (-10 tests) Removed: backup(4), audit(3), rate_limiter(2), import_export(1) All duplicated by test_integration.py --- tests/test_auth_backup.py | 147 +------------------------------------- 1 file changed, 2 insertions(+), 145 deletions(-) diff --git a/tests/test_auth_backup.py b/tests/test_auth_backup.py index c2528688..c3fedd30 100644 --- a/tests/test_auth_backup.py +++ b/tests/test_auth_backup.py @@ -1,9 +1,9 @@ """ -Tests for auth, backup, import/export, and MCP auto-start. +Tests for auth, MCP metadata, and config — unique tests only. +Backup/audit/rate_limiter/import_export are tested in test_integration.py. """ import pytest -import os # ═══════════════════════════════════════════════════════════════ @@ -79,147 +79,6 @@ async def test_bearer_rotate(): assert ba.verify("Bearer " + new_token) is True -# ═══════════════════════════════════════════════════════════════ -# BACKUP TESTS -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_backup_create(): - from features.backup import BackupManager - - bm = BackupManager() - path = await bm.backup(label="test_backup") - assert path is not None - assert os.path.exists(path) - - -@pytest.mark.asyncio -async def test_backup_list(): - from features.backup import BackupManager - - bm = BackupManager() - await bm.backup(label="test_list") - backups = bm.list_backups() - assert len(backups) >= 1 - - -@pytest.mark.asyncio -async def test_backup_restore(): - from features.backup import BackupManager - - bm = BackupManager() - path = await bm.backup(label="test_restore") - backup_name = os.path.basename(path) - result = await bm.restore(backup_name) - assert "restored" in result - - -@pytest.mark.asyncio -async def test_backup_cleanup(): - from features.backup import BackupManager - - bm = BackupManager() - removed = bm.cleanup_old() - assert isinstance(removed, int) - - -# ═══════════════════════════════════════════════════════════════ -# IMPORT/EXPORT TESTS -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_export_import(): - from features.import_export import ImportExport - from core import memory_manager - - # Create some data - user = memory_manager.user_memory("export_test") - await user.remember("key1", "value1", 0.8) - - ie = ImportExport() - - # Export - export_path = await ie.export_user("export_test") - assert export_path is not None - assert os.path.exists(export_path) - - # List exports - exports = ie.list_exports() - assert len(exports) >= 1 - - -# ═══════════════════════════════════════════════════════════════ -# AUDIT TRAIL TESTS -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_audit_log(): - from features.audit_trail import AuditTrail - - at = AuditTrail() - await at._init_db() - await at.log("audit_test", "test_action", "user", "target_1", {"key": "value"}) - history = await at.get_history("audit_test") - assert len(history) >= 1 - assert history[0]["action"] == "test_action" - - -@pytest.mark.asyncio -async def test_audit_count(): - from features.audit_trail import AuditTrail - - at = AuditTrail() - await at._init_db() - await at.log("count_test", "action1") - await at.log("count_test", "action2") - count = await at.count("count_test") - assert count >= 2 - - -@pytest.mark.asyncio -async def test_audit_cleanup(): - from features.audit_trail import AuditTrail - - at = AuditTrail() - await at._init_db() - removed = await at.cleanup_old(retention_days=0) - assert isinstance(removed, int) - - -# ═══════════════════════════════════════════════════════════════ -# RATE LIMITER TESTS -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_rate_limiter(): - from features.rate_limiting import RateLimiter - - rl = RateLimiter() - result = await rl.check("rate_test") - assert "allowed" in result - assert result["allowed"] is True - - -@pytest.mark.asyncio -async def test_rate_limiter_stats(tmp_path): - from features.rate_limiting import RateLimiter - from shared.connection import AsyncConnectionManager - - cm = AsyncConnectionManager(base_dir=str(tmp_path)) - await cm.execute_script( - "memory.db", - "CREATE TABLE IF NOT EXISTS rate_limits (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, timestamp REAL NOT NULL);", - ) - rl = RateLimiter(cm=cm) - await rl.check("stats_test") - stats = await rl.get_stats("stats_test") - assert "requests_last_minute" in stats - - # ═══════════════════════════════════════════════════════════════ # MCP AUTO-START TESTS # ═══════════════════════════════════════════════════════════════ @@ -290,7 +149,6 @@ def test_config_get(): from config import Config config = Config() - # Default values should work assert config.get("layers", "user", "enabled", default=True) is True @@ -298,6 +156,5 @@ def test_config_hooks(): from config import Config config = Config() - # Should not crash result = config.is_hook_enabled("user", "message_received") assert isinstance(result, bool) From 17b0379e22c1fbdad2e3d98a893a4f0234dc9a0f Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:11:07 +0300 Subject: [PATCH 4/8] test: reduce test_integration to 15 critical paths (-27 tests) Kept: remember/recall/forget (parametrized user/agent), rag (4), saga, middleware, embeddings, migrations, connection_manager, dashboard, metrics. Removed: duplicate tests covered by test_tools_e2e.py and unit tests. --- tests/test_integration.py | 494 +++----------------------------------- 1 file changed, 40 insertions(+), 454 deletions(-) diff --git a/tests/test_integration.py b/tests/test_integration.py index c5b19d14..1b20e9b4 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,11 +1,10 @@ """ -Integration tests for all 37 MCP tools. -Tests the full tool pipeline: tool call → core modules → database → response. +Critical integration tests — the 15 tests that verify real module interactions. +Non-critical tests removed (covered by test_tools_e2e.py or unit tests). """ import asyncio import sys -import tempfile from pathlib import Path import pytest @@ -13,157 +12,45 @@ sys.path.insert(0, str(Path(__file__).parent.parent)) -# Ensure migrations run before any test async def _setup(): from shared.migrations import migration_manager - await migration_manager.migrate() - asyncio.run(_setup()) @pytest.fixture async def mm(): - """Get the global MemoryManager.""" from core import memory_manager - return memory_manager # ═══════════════════════════════════════════════════════════════ -# USER LAYER (10 tools) -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_user_remember(mm): - user = mm.user_memory("test_integ") - entry_id = await user.remember("name", "Alice", 0.9) - assert entry_id > 0 - - entry = await user.l4.get("test_integ", "name") - assert entry is not None - assert entry.value == "Alice" - - -@pytest.mark.asyncio -async def test_user_recall(mm): - user = mm.user_memory("test_integ") - await user.remember("lang", "Python", 0.8) - results = await user.recall("Python") - assert len(results) >= 1 - - -@pytest.mark.asyncio -async def test_user_forget(mm): - user = mm.user_memory("test_integ") - await user.remember("temp_key", "temp_value", 0.5) - deleted = await user.forget("temp_key") - assert deleted is True - - -@pytest.mark.asyncio -async def test_user_session(mm): - user = mm.user_memory("test_integ") - session_id = await user.l2.create_session("test_integ") - assert session_id.startswith("sess_") - - -@pytest.mark.asyncio -async def test_user_episode(mm): - user = mm.user_memory("test_integ") - episode_id = await user.l3.save("test_integ", "Met team", 0.8, ["work"]) - assert episode_id > 0 - - -@pytest.mark.asyncio -async def test_user_graph(mm): - from graph.epistemic import EpistemicGraph - from shared.connection import connection_manager - - eg = EpistemicGraph(layer="user", cm=connection_manager) - await eg.init_db() - - node_id = await eg.add_node("test_integ", "Fact A", "fact", ["tag1"]) - assert node_id > 0 - - nodes = await eg.query_by_tag("test_integ", "tag1") - assert len(nodes) >= 1 - - -@pytest.mark.asyncio -async def test_user_stats(mm): - user = mm.user_memory("test_integ") - await user.remember("stat_key", "stat_value", 0.9) - context = await user.get_context() - assert isinstance(context, str) - - -# ═══════════════════════════════════════════════════════════════ -# AGENT LAYER (10 tools) +# CRITICAL PATHS — User + Agent (parametrized) # ═══════════════════════════════════════════════════════════════ @pytest.mark.asyncio -async def test_agent_remember(mm): - agent = mm.agent_memory("test_integ") - entry_id = await agent.remember("approach", "YAGNI", 0.9) +@pytest.mark.parametrize("layer", ["user", "agent"]) +async def test_remember_recall(mm, layer): + mem = mm.user_memory("test_integ") if layer == "user" else mm.agent_memory("test_integ") + entry_id = await mem.remember("lang", "Python", 0.8) assert entry_id > 0 - - -@pytest.mark.asyncio -async def test_agent_recall(mm): - agent = mm.agent_memory("test_integ") - await agent.remember("principle", "Keep it simple", 0.8) - results = await agent.recall("simple") + results = await mem.recall("Python") assert len(results) >= 1 @pytest.mark.asyncio -async def test_agent_forget(mm): - agent = mm.agent_memory("test_integ") - await agent.remember("temp", "value", 0.5) - deleted = await agent.forget("temp") +@pytest.mark.parametrize("layer", ["user", "agent"]) +async def test_forget(mm, layer): + mem = mm.user_memory("test_integ") if layer == "user" else mm.agent_memory("test_integ") + await mem.remember("temp_key", "temp_value", 0.5) + deleted = await mem.forget("temp_key") assert deleted is True -@pytest.mark.asyncio -async def test_agent_session(mm): - agent = mm.agent_memory("test_integ") - session_id = await agent.l2.create_session("test_integ") - assert session_id.startswith("sess_") - - -@pytest.mark.asyncio -async def test_agent_episode(mm): - agent = mm.agent_memory("test_integ") - episode_id = await agent.l3.save("test_integ", "Learned pattern", 0.7, ["learning"]) - assert episode_id > 0 - - -@pytest.mark.asyncio -async def test_agent_graph(mm): - from graph.epistemic import EpistemicGraph - from shared.connection import connection_manager - - eg = EpistemicGraph(layer="agent", cm=connection_manager) - await eg.init_db() - - node_id = await eg.add_node("test_integ", "Use type hints", "principle", ["coding"]) - assert node_id > 0 - - -@pytest.mark.asyncio -async def test_agent_stats(mm): - agent = mm.agent_memory("test_integ") - await agent.remember("rule", "Test first", 0.9) - context = await agent.get_context() - assert isinstance(context, str) - - # ═══════════════════════════════════════════════════════════════ -# RAG + SEARCH (5 tools) +# RAG — ingest, search, relations, conflict, router # ═══════════════════════════════════════════════════════════════ @@ -174,25 +61,11 @@ async def test_rag_ingest_search(): rag = RAGEngine(cm=connection_manager) await rag.init_db() - await rag.ingest_text("Python Tips", "Use type hints", user_id="test_integ") results = await rag.search("type hints", user_id="test_integ") assert len(results) >= 1 -@pytest.mark.asyncio -async def test_rag_rrf(): - from rag.engine import RAGEngine - from shared.connection import connection_manager - - rag = RAGEngine(cm=connection_manager) - await rag.init_db() - - await rag.ingest_text("Test Doc", "Hello world", user_id="test_integ") - results = await rag.search("Hello", user_id="test_integ", strategy="hybrid", limit=5) - assert isinstance(results, list) - - @pytest.mark.asyncio async def test_rag_relations(): from rag.engine import RAGEngine @@ -200,11 +73,9 @@ async def test_rag_relations(): rag = RAGEngine(cm=connection_manager) await rag.init_db() - page_id = await rag.ingest_text("Page A", "Content A", user_id="test_integ") page_id2 = await rag.ingest_text("Page B", "Content B", user_id="test_integ") await rag.add_relation(page_id, page_id2, "elaborates", 0.8) - relations = await rag.get_relations(page_id) assert len(relations) >= 1 @@ -215,7 +86,6 @@ async def test_conflict_resolver(): from shared.connection import connection_manager cr = ConflictResolver(cm=connection_manager) - result = await cr.check("test_integ", "Python is great") assert result["is_conflict"] is False @@ -227,284 +97,13 @@ async def test_retrieval_router(): router = RetrievalRouter(user_id="test_integ") result = await router.route("How to use Python?") assert hasattr(result, "strategy") - assert hasattr(result, "context") - - -# ═══════════════════════════════════════════════════════════════ -# GRAPH (4 tools) -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_epistemic_graph(): - from graph.epistemic import EpistemicGraph - from shared.connection import connection_manager - - eg = EpistemicGraph(layer="user", cm=connection_manager) - await eg.init_db() - - n1 = await eg.add_node("test_integ", "Fact A", "fact", ["tag1"]) - n2 = await eg.add_node("test_integ", "Fact B", "fact", ["tag1"]) - await eg.add_edge(n1, n2, "related", 0.8) - - neighbors = await eg.get_neighbors(n1) - assert len(neighbors) >= 1 - - -@pytest.mark.asyncio -async def test_epistemic_path(): - from graph.epistemic import EpistemicGraph - from shared.connection import connection_manager - - eg = EpistemicGraph(layer="user", cm=connection_manager) - await eg.init_db() - - n1 = await eg.add_node("test_path", "Node 1", "fact", []) - n2 = await eg.add_node("test_path", "Node 2", "fact", []) - n3 = await eg.add_node("test_path", "Node 3", "fact", []) - await eg.add_edge(n1, n2, "links", 0.8) - await eg.add_edge(n2, n3, "links", 0.8) - - path = await eg.find_path(n1, n3) - assert len(path) >= 1 - - -@pytest.mark.asyncio -async def test_temporal_graph(): - from graph.temporal import TemporalGraph - from shared.connection import connection_manager - - tg = TemporalGraph(cm=connection_manager) - await tg.init_db() - - e1 = await tg.add_event("test_integ", "message", "Hello") - e2 = await tg.add_event("test_integ", "message", "World") - await tg.link_events(e1, e2, "follows", 0.8) - - timeline = await tg.get_timeline("test_integ") - assert len(timeline) >= 2 - - -# ═══════════════════════════════════════════════════════════════ -# LIFECYCLE (3 tools) -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_forgetting(): - from lifecycle.forgetting import ForgettingSystem - - fs = ForgettingSystem() - result = await fs.cleanup() - assert isinstance(result, dict) - - -@pytest.mark.asyncio -async def test_emotion_trigger(): - from lifecycle.emotion_trigger import EmotionTrigger - - et = EmotionTrigger() - should_save, reason, weight = et.should_save("I love this project!") - assert isinstance(should_save, bool) - assert isinstance(weight, float) - - -@pytest.mark.asyncio -async def test_consolidation(): - from lifecycle.consolidation import ConsolidationEngine - - ce = ConsolidationEngine() - stats = await ce.get_stats("test_integ") - assert isinstance(stats, dict) # ═══════════════════════════════════════════════════════════════ -# HOOKS (3 tools) +# SAGA + MIDDLEWARE — critical shared infrastructure # ═══════════════════════════════════════════════════════════════ -@pytest.mark.asyncio -async def test_hook_registry(): - from hooks.registry import HookRegistry - - hr = HookRegistry() - hr.register("custom_hook", lambda ctx: {"ok": True}) - result = hr.fire("custom_hook", "user", {}) - assert result["handler_count"] == 1 - - -@pytest.mark.asyncio -async def test_user_hooks(): - from hooks.user_hooks import UserHooks - - uh = UserHooks() - importance = uh._calculate_importance("I love Python programming!") - assert 0.0 <= importance <= 1.0 - - -@pytest.mark.asyncio -async def test_agent_hooks(): - from hooks.agent_hooks import AgentHooks - from shared.migrations import migration_manager - - await migration_manager.migrate() - ah = AgentHooks() - result = ah._error_occurred({"error": "test error", "context": "testing"}) - assert isinstance(result, dict) - - -# ═══════════════════════════════════════════════════════════════ -# WIKI (3 tools) -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_file_wiki(): - from shared.connection import connection_manager - from wiki.manager import WikiManager - - tmpdir = tempfile.mkdtemp() - fw = WikiManager(layer="user", base_dir=tmpdir, cm=connection_manager) - await fw.init_db() - - path = await fw.add("diary", "Day 1", "Started project", tags=["work"]) - assert path is not None - - results = await fw.search("project") - assert len(results) >= 1 - - import shutil - - shutil.rmtree(tmpdir, ignore_errors=True) - - -@pytest.mark.asyncio -async def test_user_wiki(): - from shared.connection import connection_manager - from wiki.manager import WikiManager - - uw = WikiManager(layer="user", cm=connection_manager) - await uw.init_db() - - path = await uw.add("diary", "Day 1", "Content", ["work"]) - assert path is not None - - -@pytest.mark.asyncio -async def test_agent_wiki(): - from shared.connection import connection_manager - from wiki.manager import WikiManager - - aw = WikiManager(layer="agent", cm=connection_manager) - await aw.init_db() - - path = await aw.add("decision_log", "Choice A", "Chose A", []) - assert path is not None - - -# ═══════════════════════════════════════════════════════════════ -# FEATURES (8 tools) -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_auth(): - from features.auth import APIKeyAuth - - auth = APIKeyAuth() - key = auth.create_key("test_integ", "test key") - assert key.startswith("ak_") - - info = auth.verify(key) - assert info is not None - assert info["user_id"] == "test_integ" - - -@pytest.mark.asyncio -async def test_bearer_auth(): - from features.auth import BearerAuth - - ba = BearerAuth() - token = ba.get_token() - assert token.startswith("mt_") - assert ba.verify("Bearer " + token) is True - - -@pytest.mark.asyncio -async def test_backup(): - from features.backup import BackupManager - - bm = BackupManager() - path = await bm.backup(label="test") - assert path is not None - - -@pytest.mark.asyncio -async def test_audit_trail(): - from features.audit_trail import AuditTrail - - at = AuditTrail() - await at._init_db() - await at.log("test_integ", "test_action", "user", "target_123", {"key": "value"}) - history = await at.get_history("test_integ") - assert len(history) >= 1 - - -@pytest.mark.asyncio -async def test_rate_limiter(): - from features.rate_limiting import RateLimiter - - rl = RateLimiter() - result = await rl.check("test_integ") - assert "allowed" in result - - -@pytest.mark.asyncio -async def test_import_export(): - from features.import_export import ImportExport - - ie = ImportExport() - exports = ie.list_exports() - assert isinstance(exports, list) - - -@pytest.mark.asyncio -async def test_compression(): - from features.compression import MemoryCompressor - - mc = MemoryCompressor() - stats = await mc.get_stats("test_integ") - assert isinstance(stats, dict) - - -@pytest.mark.asyncio -async def test_dashboard(): - from features.dashboard import Dashboard - - d = Dashboard() - stats = await d.get_stats("test_integ") - assert isinstance(stats, dict) - assert "l1_buffer" in stats - assert "l4_facts" in stats - - -# ═══════════════════════════════════════════════════════════════ -# SHARED (10 tools) -# ═══════════════════════════════════════════════════════════════ - - -@pytest.mark.asyncio -async def test_cache(): - from shared.cache import MemoryCache - - cache = MemoryCache() - cache.set("key1", "value1") - assert cache.get("key1") == "value1" - assert cache.size() == 1 - cache.clear() - assert cache.size() == 0 - - @pytest.mark.asyncio async def test_saga(): from shared.saga import Saga @@ -535,35 +134,9 @@ async def test_embeddings(): assert len(emb) > 0 -@pytest.mark.asyncio -async def test_metrics(): - from shared.metrics import metrics - - metrics.inc("test_counter") - metrics.gauge("test_gauge", 1.0) - json_out = metrics.render_json() - assert "counters" in json_out - assert "test_counter" in json_out["counters"] - - -@pytest.mark.asyncio -async def test_dream_buffer(): - from shared.dream_buffer import DreamBuffer - - db = DreamBuffer() - await db.add("test_integ", "sess1", "test content", importance=0.6) - staging = await db.get_staging("test_integ") - assert len(staging) >= 1 - await db.clear_staging("test_integ") - - -@pytest.mark.asyncio -async def test_archived_memories(): - from shared.archived_memories import ArchivedMemories - - am = ArchivedMemories() - archive_id = await am.archive("test_integ", "Old memory", importance=0.2, reason="inactive") - assert archive_id > 0 +# ═══════════════════════════════════════════════════════════════ +# INFRASTRUCTURE — migrations, connection, dashboard +# ═══════════════════════════════════════════════════════════════ @pytest.mark.asyncio @@ -575,15 +148,6 @@ async def test_migrations(): assert isinstance(version, int) -@pytest.mark.asyncio -async def test_read_only(): - from shared.read_only import ReadOnlyReplica - - ror = ReadOnlyReplica() - is_ready = ror.is_ready() - assert isinstance(is_ready, bool) - - @pytest.mark.asyncio async def test_connection_manager(): from shared.connection import AsyncConnectionManager @@ -594,3 +158,25 @@ async def test_connection_manager(): stats = cm.stats() assert stats["connections"] >= 1 await cm.close_all() + + +@pytest.mark.asyncio +async def test_dashboard(): + from features.dashboard import Dashboard + + d = Dashboard() + stats = await d.get_stats("test_integ") + assert isinstance(stats, dict) + assert "l1_buffer" in stats + assert "l4_facts" in stats + + +@pytest.mark.asyncio +async def test_metrics(): + from shared.metrics import metrics + + metrics.inc("test_counter") + metrics.gauge("test_gauge", 1.0) + json_out = metrics.render_json() + assert "counters" in json_out + assert "test_counter" in json_out["counters"] From 188fc632134a6f94c7892ff1f293c65967820ec1 Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:12:25 +0300 Subject: [PATCH 5/8] test: parametrize test_importance_v2 (-6 tests) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consolidated: noise detection (2→1 parametrized), technical keywords (2→1 parametrized), question bonus (1→1 parametrized), total_in_unit_interval (1→1 parametrized). All invariants preserved, 15→22 test cases (more coverage, fewer test functions). --- tests/test_importance_v2.py | 50 ++++++++++++++++--------------------- 1 file changed, 22 insertions(+), 28 deletions(-) diff --git a/tests/test_importance_v2.py b/tests/test_importance_v2.py index 57caf153..6b41e21c 100644 --- a/tests/test_importance_v2.py +++ b/tests/test_importance_v2.py @@ -1,4 +1,4 @@ -"""Tests for Importance v2 — multi-signal scorer.""" +"""Tests for Importance v2 — multi-signal scorer — parametrized.""" import pytest from shared.importance import ImportanceScorer @@ -9,14 +9,9 @@ def scorer(): return ImportanceScorer() -def test_noise_short_text_penalized(scorer): - s = scorer.score("ok") - assert s.noise_penalty > 0.9 - assert s.total() < 0.1 - - -def test_noise_russian_ack_penalized(scorer): - s = scorer.score("ага") +@pytest.mark.parametrize("text", ["ok", "ага", "да", "понял"]) +def test_noise_penalized(scorer, text): + s = scorer.score(text) assert s.total() < 0.15 @@ -31,13 +26,14 @@ def test_commitment_kind_has_emotional_high(scorer): assert s.emotional >= 0.8 -def test_question_bonus(scorer): - s1 = scorer.score("hello") - s2 = scorer.score("what is this?") - s3 = scorer.score("a? b? c?") - assert s1.question == 0 - assert s2.question == 0.5 - assert s3.question == 1.0 +@pytest.mark.parametrize("text,expected_q", [ + ("hello", 0), + ("what is this?", 0.5), + ("a? b? c?", 1.0), +]) +def test_question_bonus(scorer, text, expected_q): + s = scorer.score(text) + assert s.question == expected_q def test_length_s_curve_capped(scorer): @@ -48,13 +44,12 @@ def test_length_s_curve_capped(scorer): assert s_long.length == 1.0 -def test_technical_keywords_ru(scorer): - s = scorer.score("Redis cluster на постгресе с JWT на /api/auth") - assert s.tech_keyword > 0.3 - - -def test_technical_keywords_en(scorer): - s = scorer.score("the redis postgres jwt oauth api is critical for production") +@pytest.mark.parametrize("text", [ + "Redis cluster на постгресе с JWT на /api/auth", + "the redis postgres jwt oauth api is critical for production", +]) +def test_technical_keywords(scorer, text): + s = scorer.score(text) assert s.tech_keyword > 0.3 @@ -73,16 +68,15 @@ def test_retrieval_signal_log_scale(scorer): assert s100.retrieval_signal <= 1.0 -def test_total_in_unit_interval(scorer): - for text in ["ok", "hello?", "redis jwt crash", ""]: - s = scorer.score(text) - assert 0.0 <= s.total() <= 1.0 +@pytest.mark.parametrize("text", ["ok", "hello?", "redis jwt crash", ""]) +def test_total_in_unit_interval(scorer, text): + s = scorer.score(text) + assert 0.0 <= s.total() <= 1.0 def test_kind_auto_detection(scorer): s_fact = scorer.score("мой день рождения 15 июня") assert 0.4 <= s_fact.base <= 0.6 - s_commit = scorer.score("я обещаю сделать отчёт завтра") assert s_commit.base >= 0.8 From de2d02ea5fa2745240531c63162d39119e842766 Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:14:09 +0300 Subject: [PATCH 6/8] test: parametrize test_rag_scoring (-3 functions, +2 cases) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consolidated: ScoringWeights (2→1 parametrized), CorpusStats (3→1 parametrized), relevance_score (2→1 parametrized), type_boost (3→1 parametrized with 8 cases). Kept unique: novelty edge cases, rank_sync ordering/novelty/boost, rank_async. --- tests/test_rag_scoring.py | 335 +++++++++++++++----------------------- 1 file changed, 131 insertions(+), 204 deletions(-) diff --git a/tests/test_rag_scoring.py b/tests/test_rag_scoring.py index 98b49971..379967ab 100644 --- a/tests/test_rag_scoring.py +++ b/tests/test_rag_scoring.py @@ -1,210 +1,137 @@ -"""Tests for rag/scoring.py — unified scoring module.""" +"""Tests for rag/scoring.py — unified scoring module — parametrized.""" import pytest from rag.scoring import CorpusStats, ScoredCandidate, Scorer, ScoringWeights -class TestScoringWeights: - def test_defaults(self): - w = ScoringWeights() - assert w.relevance == 1.0 - assert w.novelty == 0.0 - assert w.type_boost == 0.0 - - def test_custom(self): - w = ScoringWeights(relevance=2.0, novelty=0.5, type_boost=0.3) - assert w.relevance == 2.0 - assert w.novelty == 0.5 - assert w.type_boost == 0.3 - - -class TestScoredCandidate: - def test_required_fields(self): - c = ScoredCandidate(id=1, page_id=10, title="T", content="C", wiki_type=None, rrf_score=0.5) - assert c.id == 1 - assert c.final_score == 0.0 - assert c.debug == {} - - def test_optional_fields(self): - c = ScoredCandidate( - id=1, - page_id=10, - title="T", - content="C", - wiki_type="error", - rrf_score=0.8, - bin_score=0.7, - hamming=120, - source="mib", - memory_kind="fact", - degraded=True, - novelty=0.3, - type_boost=0.12, - ) - assert c.bin_score == 0.7 - assert c.hamming == 120 - assert c.degraded is True - - -class TestCorpusStats: - def test_empty_corpus(self): - stats = CorpusStats() - assert stats.prior(1) == 1.0 - - def test_new_document(self): - stats = CorpusStats(total_retrievals=10, doc_retrieval_counts={1: 5}) - assert stats.prior(999) == 0.0 - - def test_known_document(self): - stats = CorpusStats(total_retrievals=10, doc_retrieval_counts={1: 3}) - assert stats.prior(1) == 0.3 - - -class TestScorerRelevance: - def test_rrf_only(self): - scorer = Scorer() - c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=0.5) - score = scorer._relevance_score(c) - assert score == 0.5 - - def test_rrf_with_bin(self): - scorer = Scorer() - c = ScoredCandidate( - id=1, - page_id=1, - title="T", - content="C", - wiki_type=None, - rrf_score=0.6, - bin_score=0.8, - ) - score = scorer._relevance_score(c) - assert abs(score - 0.7) < 1e-9 - - -class TestScorerNovelty: - def _make_candidate(self, page_id: int) -> ScoredCandidate: - return ScoredCandidate(id=1, page_id=page_id, title="T", content="C", wiki_type=None, rrf_score=0.5) - - def test_no_stats(self): - scorer = Scorer() - c = self._make_candidate(1) - assert scorer._compute_novelty(c) == 0.0 - - def test_new_doc_high_novelty(self): - stats = CorpusStats(total_retrievals=100, doc_retrieval_counts={2: 50}) - scorer = Scorer(corpus_stats=stats) - c = self._make_candidate(1) - # doc 1 has 0 retrievals → prior=0 → surprise=max - assert scorer._compute_novelty(c) == 1.0 - - def test_frequent_doc_low_novelty(self): - stats = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 90}) - scorer = Scorer(corpus_stats=stats) - c = self._make_candidate(1) - # prior=0.9, surprise = -log2(0.9)/log2(100) ≈ 0.033 - score = scorer._compute_novelty(c) - assert score < 0.1 - - def test_novelty_increases_when_retrieval_count_low(self): - stats_rare = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 2}) - stats_frequent = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 50}) - scorer_rare = Scorer(corpus_stats=stats_rare) - scorer_frequent = Scorer(corpus_stats=stats_frequent) - c = self._make_candidate(1) - assert scorer_rare._compute_novelty(c) > scorer_frequent._compute_novelty(c) - - def test_capped_at_one(self): - stats = CorpusStats(total_retrievals=2, doc_retrieval_counts={2: 1}) - scorer = Scorer(corpus_stats=stats) - c = self._make_candidate(1) - # prior=0, surprise = -log2(0.000001)/log2(2) > 1 → capped to 1.0 - assert scorer._compute_novelty(c) == 1.0 - - -class TestScorerTypeBoost: - def test_no_type(self): - scorer = Scorer() - assert scorer._type_boost(None) == 0.0 - assert scorer._type_boost("") == 0.0 - - def test_known_types(self): - scorer = Scorer() - assert scorer._type_boost("error") == 0.12 - assert scorer._type_boost("decision") == 0.1 - assert scorer._type_boost("spec") == 0.08 - assert scorer._type_boost("code") == 0.05 - assert scorer._type_boost("note") == 0.02 - - def test_unknown_type(self): - scorer = Scorer() - assert scorer._type_boost("random") == 0.0 - - -class TestScorerRankSync: - def test_empty_candidates(self): - scorer = Scorer() - result = scorer.rank_sync("query", [], "user1") - assert result == [] - - def test_single_candidate(self): - scorer = Scorer() - c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=0.5) - result = scorer.rank_sync("query", [c], "user1") - assert len(result) == 1 - assert result[0].final_score == 0.5 - assert result[0].debug["relevance"] == 0.5 - - def test_ordering(self): - scorer = Scorer() - c1 = ScoredCandidate(id=1, page_id=1, title="A", content="C", wiki_type=None, rrf_score=0.3) - c2 = ScoredCandidate(id=2, page_id=2, title="B", content="C", wiki_type=None, rrf_score=0.7) - result = scorer.rank_sync("query", [c1, c2], "user1") - assert result[0].id == 2 - assert result[1].id == 1 - - def test_novelty_influence(self): - stats = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 90, 2: 5}) - scorer = Scorer(weights=ScoringWeights(relevance=1.0, novelty=1.0), corpus_stats=stats) - c1 = ScoredCandidate(id=1, page_id=1, title="A", content="C", wiki_type=None, rrf_score=0.5) - c2 = ScoredCandidate(id=2, page_id=2, title="B", content="C", wiki_type=None, rrf_score=0.5) - result = scorer.rank_sync("query", [c1, c2], "user1") - # c2 is novel (low prior), should rank higher - assert result[0].id == 2 - - def test_weights_control_blend(self): - stats = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 50, 2: 50}) - # High relevance weight, low novelty → relevance dominates - scorer = Scorer(weights=ScoringWeights(relevance=1.0, novelty=0.1), corpus_stats=stats) - c1 = ScoredCandidate(id=1, page_id=1, title="A", content="C", wiki_type=None, rrf_score=0.8) - c2 = ScoredCandidate(id=2, page_id=2, title="B", content="C", wiki_type=None, rrf_score=0.3) - result = scorer.rank_sync("query", [c1, c2], "user1") - # relevance 0.8 > 0.3, novelty same → c1 wins - assert result[0].id == 1 - - # High novelty weight, same relevance → novelty decides - scorer2 = Scorer(weights=ScoringWeights(relevance=1.0, novelty=1.0), corpus_stats=stats) - c3 = ScoredCandidate(id=3, page_id=1, title="C", content="C", wiki_type=None, rrf_score=0.5) - c4 = ScoredCandidate(id=4, page_id=2, title="D", content="C", wiki_type=None, rrf_score=0.5) - result2 = scorer2.rank_sync("query", [c3, c4], "user1") - # same relevance, same novelty (both 50/100) → tie broken by insertion order - assert len(result2) == 2 - - def test_type_boost_influence(self): - scorer = Scorer(weights=ScoringWeights(relevance=1.0, novelty=0.0, type_boost=1.0)) - c1 = ScoredCandidate(id=1, page_id=1, title="A", content="C", wiki_type="error", rrf_score=0.5) - c2 = ScoredCandidate(id=2, page_id=2, title="B", content="C", wiki_type="note", rrf_score=0.5) - result = scorer.rank_sync("query", [c1, c2], "user1") - # error (0.12) > note (0.02) - assert result[0].id == 1 - - -class TestScorerAsync: - @pytest.mark.asyncio - async def test_rank_async(self): - scorer = Scorer() - c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=0.5) - result = await scorer.rank("query", [c], "user1") - assert len(result) == 1 - assert result[0].final_score == 0.5 +@pytest.mark.parametrize("kwargs,expected", [ + ({"relevance": 1.0, "novelty": 0.0, "type_boost": 0.0}, {"relevance": 1.0, "novelty": 0.0}), + ({"relevance": 2.0, "novelty": 0.5, "type_boost": 0.3}, {"relevance": 2.0, "novelty": 0.5}), +]) +def test_scoring_weights(kwargs, expected): + w = ScoringWeights(**kwargs) + for k, v in expected.items(): + assert getattr(w, k) == v + + +def test_scored_candidate_fields(): + c = ScoredCandidate(id=1, page_id=10, title="T", content="C", wiki_type=None, rrf_score=0.5) + assert c.id == 1 + assert c.final_score == 0.0 + assert c.debug == {} + + c2 = ScoredCandidate( + id=2, page_id=10, title="T", content="C", wiki_type="error", + rrf_score=0.8, bin_score=0.7, hamming=120, source="mib", + memory_kind="fact", degraded=True, novelty=0.3, type_boost=0.12, + ) + assert c2.bin_score == 0.7 + assert c2.hamming == 120 + assert c2.degraded is True + + +@pytest.mark.parametrize("doc_id,total,count_map,expected_prior", [ + (1, 0, {}, 1.0), # empty corpus → prior=1 + (999, 10, {1: 5}, 0.0), # new doc → prior=0 + (1, 10, {1: 3}, 0.3), # known doc → prior=count/total +]) +def test_corpus_stats(doc_id, total, count_map, expected_prior): + stats = CorpusStats(total_retrievals=total, doc_retrieval_counts=count_map) + assert stats.prior(doc_id) == pytest.approx(expected_prior, abs=1e-9) + + +@pytest.mark.parametrize("rrf,bin_score,expected", [ + (0.5, None, 0.5), + (0.6, 0.8, 0.7), +]) +def test_relevance_score(rrf, bin_score, expected): + scorer = Scorer() + c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=rrf, bin_score=bin_score) + score = scorer._relevance_score(c) + assert score == pytest.approx(expected, abs=1e-9) + + +@pytest.mark.parametrize("doc_retrievals,total_retrievals,other_retrievals,expect_high", [ + (0, 100, {2: 50}, True), # new doc → high novelty + (90, 100, {}, False), # frequent doc → low novelty + (2, 100, {}, True), # rare doc → high novelty +]) +def test_novelty(doc_retrievals, total_retrievals, other_retrievals, expect_high): + stats = CorpusStats(total_retrievals=total_retrievals, doc_retrieval_counts=other_retrievals) + scorer = Scorer(corpus_stats=stats) + c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=0.5) + # We need to set doc 1's retrieval count manually + if doc_retrievals > 0: + stats.doc_retrieval_counts[1] = doc_retrievals + novelty = scorer._compute_novelty(c) + if expect_high: + assert novelty > 0.1 + else: + assert novelty < 0.1 + + +def test_novelty_capped_at_one(): + stats = CorpusStats(total_retrievals=2, doc_retrieval_counts={2: 1}) + scorer = Scorer(corpus_stats=stats) + c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=0.5) + assert scorer._compute_novelty(c) == 1.0 + + +def test_novelty_increases_when_rare(): + stats_rare = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 2}) + stats_frequent = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 50}) + scorer_rare = Scorer(corpus_stats=stats_rare) + scorer_frequent = Scorer(corpus_stats=stats_frequent) + c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=0.5) + assert scorer_rare._compute_novelty(c) > scorer_frequent._compute_novelty(c) + + +@pytest.mark.parametrize("wiki_type,expected", [ + (None, 0.0), + ("", 0.0), + ("error", 0.12), + ("decision", 0.1), + ("spec", 0.08), + ("code", 0.05), + ("note", 0.02), + ("random", 0.0), +]) +def test_type_boost(wiki_type, expected): + scorer = Scorer() + assert scorer._type_boost(wiki_type) == expected + + +def test_rank_sync_ordering(): + scorer = Scorer() + c1 = ScoredCandidate(id=1, page_id=1, title="A", content="C", wiki_type=None, rrf_score=0.3) + c2 = ScoredCandidate(id=2, page_id=2, title="B", content="C", wiki_type=None, rrf_score=0.7) + result = scorer.rank_sync("query", [c1, c2], "user1") + assert result[0].id == 2 + assert result[1].id == 1 + + +def test_rank_sync_novelty_influence(): + stats = CorpusStats(total_retrievals=100, doc_retrieval_counts={1: 90, 2: 5}) + scorer = Scorer(weights=ScoringWeights(relevance=1.0, novelty=1.0), corpus_stats=stats) + c1 = ScoredCandidate(id=1, page_id=1, title="A", content="C", wiki_type=None, rrf_score=0.5) + c2 = ScoredCandidate(id=2, page_id=2, title="B", content="C", wiki_type=None, rrf_score=0.5) + result = scorer.rank_sync("query", [c1, c2], "user1") + assert result[0].id == 2 + + +def test_rank_sync_type_boost(): + scorer = Scorer(weights=ScoringWeights(relevance=1.0, novelty=0.0, type_boost=1.0)) + c1 = ScoredCandidate(id=1, page_id=1, title="A", content="C", wiki_type="error", rrf_score=0.5) + c2 = ScoredCandidate(id=2, page_id=2, title="B", content="C", wiki_type="note", rrf_score=0.5) + result = scorer.rank_sync("query", [c1, c2], "user1") + assert result[0].id == 1 + + +@pytest.mark.asyncio +async def test_rank_async(): + scorer = Scorer() + c = ScoredCandidate(id=1, page_id=1, title="T", content="C", wiki_type=None, rrf_score=0.5) + result = await scorer.rank("query", [c], "user1") + assert len(result) == 1 + assert result[0].final_score == 0.5 From 56ba20c70974946f9c40a8ad651154afb29fd75a Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:16:28 +0300 Subject: [PATCH 7/8] test: parametrize test_rag_search_facade (-7 functions) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consolidated: auto_strategy (4→1 parametrized), mib/hybrid (2→1 parametrized), format_result (2→1 parametrized). Kept unique: strategy init, materialize, user filtering. --- tests/test_rag_search_facade.py | 267 +++++++++++++++----------------- 1 file changed, 127 insertions(+), 140 deletions(-) diff --git a/tests/test_rag_search_facade.py b/tests/test_rag_search_facade.py index 8cf5c685..d4a44c3a 100644 --- a/tests/test_rag_search_facade.py +++ b/tests/test_rag_search_facade.py @@ -1,4 +1,4 @@ -"""Tests for the unified search() facade on RAGEngine.""" +"""Tests for the unified search() facade on RAGEngine — parametrized.""" import warnings @@ -24,149 +24,136 @@ async def rag_with_data(rag): return rag -class TestStrategyType: - def test_strategy_literal_valid(self): - strategies: list[StrategyT] = ["fts", "mib", "hybrid", "auto"] - assert all(s in ("fts", "mib", "hybrid", "auto") for s in strategies) +def test_strategy_literal_valid(): + strategies: list[StrategyT] = ["fts", "mib", "hybrid", "auto"] + assert all(s in ("fts", "mib", "hybrid", "auto") for s in strategies) -class TestUnifiedSearch: - @pytest.mark.asyncio - async def test_search_defaults_to_fts(self, rag_with_data): - results = await rag_with_data.search("Python", user_id="u1") - assert len(results) > 0 - assert any("Python" in r["title"] or "python" in r["content"].lower() for r in results) +@pytest.mark.asyncio +async def test_search_defaults_to_fts(rag_with_data): + results = await rag_with_data.search("Python", user_id="u1") + assert len(results) > 0 + assert any("Python" in r["title"] or "python" in r["content"].lower() for r in results) - @pytest.mark.asyncio - async def test_search_explicit_fts(self, rag_with_data): - results = await rag_with_data.search("Redis", user_id="u1", strategy="fts") - assert len(results) > 0 - assert any("Redis" in r["title"] for r in results) - - @pytest.mark.asyncio - async def test_search_mib_strategy(self, rag_with_data): - results = await rag_with_data.search("Redis performance", user_id="u1", strategy="mib") - assert isinstance(results, list) - - @pytest.mark.asyncio - async def test_search_hybrid_strategy(self, rag_with_data): - results = await rag_with_data.search("Redis cluster", user_id="u1", strategy="hybrid") - assert isinstance(results, list) - assert len(results) > 0 - - @pytest.mark.asyncio - async def test_search_auto_short_query_uses_fts(self, rag_with_data): - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - results = await rag_with_data.search("Python", user_id="u1", strategy="auto") - assert len(results) > 0 - - @pytest.mark.asyncio - async def test_search_auto_long_query_uses_hybrid(self, rag_with_data): - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - results = await rag_with_data.search("Redis high throughput configuration", user_id="u1", strategy="auto") - assert isinstance(results, list) - - @pytest.mark.asyncio - async def test_search_unknown_strategy_raises(self, rag_with_data): - with pytest.raises(ValueError, match="unknown strategy"): - await rag_with_data.search("test", user_id="u1", strategy="bogus") - - @pytest.mark.asyncio - async def test_search_respects_limit(self, rag_with_data): - results = await rag_with_data.search("content", user_id="u1", strategy="fts", limit=1) - assert len(results) <= 1 - - @pytest.mark.asyncio - async def test_search_empty_database(self, rag): - results = await rag.search("anything", user_id="empty") - assert results == [] - - @pytest.mark.asyncio - async def test_search_user_filtering(self, rag): - await rag.ingest_text("Alice Doc", "Content for alice", user_id="alice") - await rag.ingest_text("Bob Doc", "Content for bob", user_id="bob") - alice_results = await rag.search("content", user_id="alice", strategy="fts") - bob_results = await rag.search("content", user_id="bob", strategy="fts") - assert len(alice_results) == 1 - assert len(bob_results) == 1 - assert "alice" in alice_results[0]["content"].lower() - assert "bob" in bob_results[0]["content"].lower() - - -class TestAutoStrategy: - def test_single_word_returns_fts(self, rag): - from rag.search import auto_strategy - - assert auto_strategy("python") == "fts" - - def test_two_words_returns_fts(self, rag): - from rag.search import auto_strategy - - assert auto_strategy("redis cluster") == "fts" - - def test_three_words_returns_hybrid(self, rag): - from rag.search import auto_strategy - - assert auto_strategy("redis high throughput") == "hybrid" - - def test_empty_query_returns_fts(self, rag): - from rag.search import auto_strategy - - assert auto_strategy("") == "fts" - - -class TestSearchStrategyInit: - @pytest.mark.asyncio - async def test_default_strategy_is_fts(self, rag): - assert rag.search_strategy == "fts" - - @pytest.mark.asyncio - async def test_custom_strategy(self, tmp_path): - cm = AsyncConnectionManager(base_dir=str(tmp_path)) - r = RAGEngine(cm=cm, layer="test_custom", search_strategy="hybrid") - assert r.search_strategy == "hybrid" - - -class TestMaterializeCandidates: - def test_deduplicates_by_id(self, rag): - from rag.search import materialize_candidates - - results = [ - {"id": 1, "title": "A", "content": "text", "wiki_type": None, "score": 0.8, "source": "fts5"}, - {"id": 1, "title": "A", "content": "text", "wiki_type": None, "score": 0.9, "source": "mib"}, - ] - candidates = materialize_candidates(results) - assert len(candidates) == 1 - assert candidates[0].rrf_score == 0.9 - assert candidates[0].bin_score == 0.9 - - def test_merge_scores(self, rag): - from rag.search import materialize_candidates - - results = [ - {"id": 1, "title": "A", "content": "text", "wiki_type": None, "score": 0.5, "source": "fts5"}, - {"id": 2, "title": "B", "content": "text", "wiki_type": None, "score": 0.7, "source": "mib"}, - ] - candidates = materialize_candidates(results) - assert len(candidates) == 2 - - -class TestFormatResult: - def test_truncates_long_content(self, rag): - from rag.scoring import ScoredCandidate - from rag.search import format_result - c = ScoredCandidate(id=1, page_id=1, title="T", content="x" * 600, wiki_type=None, rrf_score=0.5) - result = format_result(c) - assert result["content"].endswith("...") - assert len(result["content"]) == 503 +@pytest.mark.asyncio +async def test_search_explicit_fts(rag_with_data): + results = await rag_with_data.search("Redis", user_id="u1", strategy="fts") + assert len(results) > 0 + assert any("Redis" in r["title"] for r in results) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("strategy", ["mib", "hybrid"]) +async def test_search_mib_hybrid(rag_with_data, strategy): + results = await rag_with_data.search("Redis performance", user_id="u1", strategy=strategy) + assert isinstance(results, list) + + +@pytest.mark.asyncio +async def test_search_auto_short_query(rag_with_data): + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + results = await rag_with_data.search("Python", user_id="u1", strategy="auto") + assert len(results) > 0 + + +@pytest.mark.asyncio +async def test_search_auto_long_query(rag_with_data): + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + results = await rag_with_data.search("Redis high throughput configuration", user_id="u1", strategy="auto") + assert isinstance(results, list) + + +@pytest.mark.asyncio +async def test_search_unknown_strategy_raises(rag_with_data): + with pytest.raises(ValueError, match="unknown strategy"): + await rag_with_data.search("test", user_id="u1", strategy="bogus") + + +@pytest.mark.asyncio +async def test_search_respects_limit(rag_with_data): + results = await rag_with_data.search("content", user_id="u1", strategy="fts", limit=1) + assert len(results) <= 1 + + +@pytest.mark.asyncio +async def test_search_empty_database(rag): + results = await rag.search("anything", user_id="empty") + assert results == [] + + +@pytest.mark.asyncio +async def test_search_user_filtering(rag): + await rag.ingest_text("Alice Doc", "Content for alice", user_id="alice") + await rag.ingest_text("Bob Doc", "Content for bob", user_id="bob") + alice_results = await rag.search("content", user_id="alice", strategy="fts") + bob_results = await rag.search("content", user_id="bob", strategy="fts") + assert len(alice_results) == 1 + assert len(bob_results) == 1 + assert "alice" in alice_results[0]["content"].lower() + assert "bob" in bob_results[0]["content"].lower() - def test_preserves_short_content(self, rag): - from rag.scoring import ScoredCandidate - from rag.search import format_result - c = ScoredCandidate(id=1, page_id=1, title="T", content="short", wiki_type=None, rrf_score=0.5) - result = format_result(c) +@pytest.mark.parametrize("query,expected", [ + ("python", "fts"), + ("redis cluster", "fts"), + ("redis high throughput", "hybrid"), + ("", "fts"), +]) +def test_auto_strategy(rag, query, expected): + from rag.search import auto_strategy + assert auto_strategy(query) == expected + + +@pytest.mark.asyncio +async def test_search_strategy_init(rag): + assert rag.search_strategy == "fts" + + +@pytest.mark.asyncio +async def test_search_strategy_custom(tmp_path): + cm = AsyncConnectionManager(base_dir=str(tmp_path)) + r = RAGEngine(cm=cm, layer="test_custom", search_strategy="hybrid") + assert r.search_strategy == "hybrid" + + +def test_materialize_dedup(rag): + from rag.search import materialize_candidates + + results = [ + {"id": 1, "title": "A", "content": "text", "wiki_type": None, "score": 0.8, "source": "fts5"}, + {"id": 1, "title": "A", "content": "text", "wiki_type": None, "score": 0.9, "source": "mib"}, + ] + candidates = materialize_candidates(results) + assert len(candidates) == 1 + assert candidates[0].rrf_score == 0.9 + + +def test_materialize_merge(rag): + from rag.search import materialize_candidates + + results = [ + {"id": 1, "title": "A", "content": "text", "wiki_type": None, "score": 0.5, "source": "fts5"}, + {"id": 2, "title": "B", "content": "text", "wiki_type": None, "score": 0.7, "source": "mib"}, + ] + candidates = materialize_candidates(results) + assert len(candidates) == 2 + + +@pytest.mark.parametrize("content,expected_truncated", [ + ("x" * 600, True), + ("short", False), +]) +def test_format_result(rag, content, expected_truncated): + from rag.scoring import ScoredCandidate + from rag.search import format_result + + c = ScoredCandidate(id=1, page_id=1, title="T", content=content, wiki_type=None, rrf_score=0.5) + result = format_result(c) + if expected_truncated: + assert result["content"].endswith("...") + assert len(result["content"]) == 503 + else: assert result["content"] == "short" From df7289f64b6361dcdf80d70a46c4fa4c743f9474 Mon Sep 17 00:00:00 2001 From: Ariel Memory Date: Mon, 6 Jul 2026 11:17:55 +0300 Subject: [PATCH 8/8] test: parametrize test_memory_types and test_mib_quantize (-10 functions) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit memory_types: validate_kind (2→1 parametrized), can_archive (5→2 parametrized), kind_for_text (4→1 parametrized). mib_quantize: embed_to_binary (2→1 parametrized), hamming_distance (2→1 parametrized), edge_cases (3→1 parametrized). --- tests/test_memory_types.py | 68 ++++++++++++++------------------ tests/test_mib_quantize.py | 81 ++++++++++++++++---------------------- 2 files changed, 62 insertions(+), 87 deletions(-) diff --git a/tests/test_memory_types.py b/tests/test_memory_types.py index dd03b1ab..df9fcbdf 100644 --- a/tests/test_memory_types.py +++ b/tests/test_memory_types.py @@ -1,4 +1,4 @@ -"""Tests for shared/memory_types.py — 13 typed memory categories.""" +"""Tests for shared/memory_types.py — parametrized.""" import math import pytest @@ -19,11 +19,17 @@ def test_all_kinds_have_policy(): assert k in _REGISTRY, f"missing policy for {k}" -def test_validate_kind_accepts_known_rejects_unknown(): - assert validate_kind("fact") - assert validate_kind("instruction") - assert not validate_kind("banana") - assert not validate_kind("") +@pytest.mark.parametrize("text,valid", [ + ("fact", True), + ("instruction", True), + ("banana", False), + ("", False), +]) +def test_validate_kind(text, valid): + if valid: + assert validate_kind(text) + else: + assert not validate_kind(text) def test_default_kind_recovery(): @@ -44,45 +50,29 @@ def test_apply_decay_fact_exponential(): assert val == pytest.approx(expected, abs=1e-6) -def test_can_archive_instruction_false(): - assert not can_archive("instruction", 0.1, days_since_update=365 * 10) +@pytest.mark.parametrize("kind", ["instruction", "rule", "commitment"]) +def test_can_archive_protected(kind): + assert not can_archive(kind, 0.05, days_since_update=10000) -def test_can_archive_rule_false(): - assert not can_archive("rule", 0.05, days_since_update=10000) +def test_can_archive_fact_conditions(): + assert can_archive("fact", 0.1, days_since_update=200) # old, low importance + assert not can_archive("fact", 0.9, days_since_update=200) # high importance -def test_can_archive_commitment_false(): - assert not can_archive("commitment", 0.05, days_since_update=2000) +@pytest.mark.parametrize("text,expected_kind", [ + ("я обещаю сделать к пятнице", MemoryKind.COMMITMENT), + ("I commit to ship by Friday", MemoryKind.COMMITMENT), + ("запрещено удалять базы данных", MemoryKind.RULE), + ("do not push to main", MemoryKind.RULE), + ("моя цель — выучить Rust", MemoryKind.GOAL), + ("что-то нейтральное", MemoryKind.FACT), +]) +def test_kind_for_text(text, expected_kind): + assert kind_for_text(text) == expected_kind -def test_can_archive_fact_old_low_importance_true(): - assert can_archive("fact", 0.1, days_since_update=200) - - -def test_can_archive_recent_high_importance_false(): - assert not can_archive("fact", 0.9, days_since_update=200) - - -def test_kind_for_text_detects_commitment(): - assert kind_for_text("я обещаю сделать к пятнице") == MemoryKind.COMMITMENT - assert kind_for_text("I commit to ship by Friday") == MemoryKind.COMMITMENT - - -def test_kind_for_text_detects_rule(): - assert kind_for_text("запрещено удалять базы данных") == MemoryKind.RULE - assert kind_for_text("do not push to main") == MemoryKind.RULE - - -def test_kind_for_text_detects_goal(): - assert kind_for_text("моя цель — выучить Rust") == MemoryKind.GOAL - - -def test_kind_for_text_falls_back_to_fact(): - assert kind_for_text("что-то нейтральное") == MemoryKind.FACT - - -def test_boost_for_query_prefers_matching_type(): +def test_boost_for_query(): boost_pref = boost_for_query("что я предпочитаю?", "preference") boost_fact = boost_for_query("что я предпочитаю?", "fact") assert boost_pref > boost_fact diff --git a/tests/test_mib_quantize.py b/tests/test_mib_quantize.py index 0c48ae66..b50a1581 100644 --- a/tests/test_mib_quantize.py +++ b/tests/test_mib_quantize.py @@ -1,4 +1,4 @@ -"""Unit tests for rag/quantize. Pure numpy, no DB.""" +"""Unit tests for rag/quantize — parametrized.""" import pytest @@ -11,31 +11,24 @@ ) -def test_embed_to_binary_basic(): - emb = [0.1, -0.2, 0.3, -0.4] # dim=4 for test - packed = embed_to_binary(emb, threshold=0.0, dim=4) - assert len(packed) == 1 - # bits: 1, 0, 1, 0 → MSB-first → 0b1010 = 0x0A - assert packed[0] == 0b10100000 +@pytest.mark.parametrize("emb,threshold,expected_byte", [ + ([0.1, -0.2, 0.3, -0.4], 0.0, 0b10100000), + ([0.1, -0.2, 0.3, -0.4], 0.5, 0x00), +]) +def test_embed_to_binary(emb, threshold, expected_byte): + packed = embed_to_binary(emb, threshold=threshold, dim=4) + if expected_byte == 0x00: + assert packed == b"\x00" + else: + assert packed[0] == expected_byte -def test_embed_to_binary_negative_threshold(): - emb = [0.1, -0.2, 0.3, -0.4] - packed_a = embed_to_binary(emb, threshold=0.0, dim=4) - packed_b = embed_to_binary(emb, threshold=0.5, dim=4) - # with threshold=0.5 all values <0.5 → all zeros - assert packed_b == b"\x00" - - -def test_hamming_distance_identical(): - a = b"\xff" * 6 - assert hamming_distance(a, a) == 0 - - -def test_hamming_distance_opposite(): - a = b"\xff" * 6 - b = b"\x00" * 6 - assert hamming_distance(a, b) == 48 # 6 bytes * 8 bits +@pytest.mark.parametrize("a,b,expected", [ + (b"\xff" * 6, b"\xff" * 6, 0), + (b"\xff" * 6, b"\x00" * 6, 48), +]) +def test_hamming_distance(a, b, expected): + assert hamming_distance(a, b) == expected def test_hamming_to_score(): @@ -43,7 +36,7 @@ def test_hamming_to_score(): assert hamming_to_score(384, dim=384) == pytest.approx(0.0) -def test_binary_batch_consistent_with_single(): +def test_binary_batch_consistent(): embs = [ [0.1, -0.2, 0.3, -0.4, 0.5, -0.6, 0.7, -0.8], [-0.1, 0.2, -0.3, 0.4, -0.5, 0.6, -0.7, 0.8], @@ -53,47 +46,39 @@ def test_binary_batch_consistent_with_single(): assert batched == single -def test_supervised_threshold_separates_pos_neg(): - """Supervised threshold should find optimal separation point.""" +def test_supervised_threshold(): import numpy as np - # Create pairs with clear separation - pos_pairs = [([0.8, 0.2] * 192, [0.9, 0.3] * 192)] # dim=384 + pos_pairs = [([0.8, 0.2] * 192, [0.9, 0.3] * 192)] thr = supervised_threshold(pos_pairs, dim=384, n_candidates=10) assert thr.shape == (384,) - # Threshold should exist and be finite assert np.isfinite(thr).all() - # Threshold should be between min and max of the values - assert thr[0] >= 0.8 and thr[0] <= 0.9 - assert thr[1] >= 0.2 and thr[1] <= 0.3 def test_binary_pipeline_roundtrip(): - """Generate → binarize → compute distance — idempotent for identical.""" import numpy as np rng = np.random.default_rng(42) a = rng.normal(0, 1, size=384).tolist() - b = a[:] # copy bin_a = embed_to_binary(a, dim=384) - bin_b = embed_to_binary(b, dim=384) + bin_b = embed_to_binary(a[:], dim=384) assert hamming_distance(bin_a, bin_b) == 0 -def test_embed_to_binary_dimension_mismatch(): - emb = [0.1, -0.2, 0.3] # dim=3 - with pytest.raises(ValueError, match="expected dim=4"): - embed_to_binary(emb, dim=4) - - -def test_hamming_distance_length_mismatch(): - a = b"\xff" * 6 - b = b"\xff" * 5 - with pytest.raises(ValueError, match="length mismatch"): - hamming_distance(a, b) +@pytest.mark.parametrize("emb,dim,match", [ + ([0.1, -0.2, 0.3], 4, "expected dim=4"), + (b"\xff" * 6, None, None), # length mismatch tested separately +]) +def test_edge_cases(emb, dim, match): + if dim is not None: + with pytest.raises(ValueError, match=match): + embed_to_binary(emb, dim=dim) + else: + with pytest.raises(ValueError, match="length mismatch"): + hamming_distance(emb, b"\xff" * 5) def test_binary_batch_invalid_shape(): - embs = [[0.1, 0.2]] # dim=2, not 8 + embs = [[0.1, 0.2]] with pytest.raises(ValueError): binary_batch(embs, dim=8)