my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	#!/usr/bin/env python3
#2	"""
#3	Comprehensive test suite for mnemosyne-stats.py
#4	Runs 30 iterations across edge cases, integration, and stress tests.
#5	"""
#6
#7	import subprocess
#8	import json
#9	import os
#10	import sys
#11	import shutil
#12	import tempfile
#13	import sqlite3
#14	from pathlib import Path
#15	from datetime import datetime
#16
#17
#18	def _safe_count(db, table):
#19	"""Mirror scripts/mnemosyne-stats.py cnt(): missing table -> 0."""
#20	try:
#21	return db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
#22	except sqlite3.OperationalError:
#23	return 0
#24
#25
#26	SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "mnemosyne-stats.py"
#27	DB_PATH = Path.home() / ".hermes" / "mnemosyne" / "data" / "mnemosyne.db"
#28	SNAP_DIR = Path.home() / ".hermes" / "mnemosyne" / "stats"
#29	WIKI_PATH = Path.home() / "wiki"
#30
#31	passed = 0
#32	failed = 0
#33	errors = []
#34
#35	def run(args="", check=True):
#36	"""Run the script and return (exit_code, stdout, stderr)."""
#37	cmd = f"cd {SCRIPT.parent} && python3 {SCRIPT} {args}"
#38	result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
#39	return result.returncode, result.stdout, result.stderr
#40
#41	def _test(name, fn):
#42	"""Run a test function and track results."""
#43	global passed, failed
#44	try:
#45	fn()
#46	passed += 1
#47	print(f" ✓ {name}")
#48	except AssertionError as e:
#49	failed += 1
#50	errors.append((name, str(e)))
#51	print(f" ✗ {name}: {e}")
#52	except Exception as e:
#53	failed += 1
#54	errors.append((name, f"EXCEPTION: {e}"))
#55	print(f" ✗ {name}: EXCEPTION: {e}")
#56
#57	# ═══════════════════════════════════════════════════════════
#58	# GROUP 1: Normal Operation (10 tests)
#59	# ═══════════════════════════════════════════════════════════
#60
#61	def test_full_dashboard():
#62	code, out, err = run()
#63	assert code == 0, f"Exit code {code}: {err}"
#64	assert "MNEMOSYNE HEALTH DASHBOARD" in out, "Missing dashboard header"
#65	assert "WORKING MEMORY:" in out, "Missing working memory section"
#66	assert "QUALITY INDICATORS" in out, "Missing quality indicators"
#67	assert "RECOMMENDATIONS" in out, "Missing recommendations"
#68
#69	def test_compact_mode():
#70	code, out, err = run("--compact")
#71	assert code == 0, f"Exit code {code}: {err}"
#72	assert "MNEMOSYNE HEALTH DASHBOARD" in out
#73	# Compact should NOT have source breakdown
#74	assert "By Source:" not in out, "Compact mode should not show source breakdown"
#75	# Compact should NOT have top recalled
#76	assert "TOP RECALLED:" not in out, "Compact mode should not show top recalled"
#77
#78	def test_json_mode():
#79	code, out, err = run("--json")
#80	assert code == 0, f"Exit code {code}: {err}"
#81	data = json.loads(out)
#82	assert "working_memory" in data, "Missing working_memory in JSON"
#83	assert "quality_score" in data, "Missing quality_score in JSON"
#84	assert isinstance(data["working_memory"]["total"], int), "wm_total not int"
#85
#86
#87	def test_json_mode_uses_mnemosyne_data_dir(tmp_path):
#88	"""Stats should read mnemosyne.db from MNEMOSYNE_DATA_DIR when configured."""
#89	home = tmp_path / "home"
#90	data_dir = tmp_path / "custom-data"
#91	env = os.environ.copy()
#92	env["HOME"] = str(home)
#93	env["MNEMOSYNE_DATA_DIR"] = str(data_dir)
#94
#95	store = subprocess.run(
#96	[sys.executable, "-m", "mnemosyne.cli", "store", "stats data dir probe"],
#97	cwd=str(SCRIPT.parent.parent),
#98	env=env,
#99	capture_output=True,
#100	text=True,
#101	timeout=30,
#102	)
#103	assert store.returncode == 0, store.stderr
#104	assert (data_dir / "mnemosyne.db").exists()
#105	assert not (home / ".hermes" / "mnemosyne" / "data" / "mnemosyne.db").exists()
#106
#107	stats = subprocess.run(
#108	[sys.executable, str(SCRIPT), "--json"],
#109	cwd=str(SCRIPT.parent),
#110	env=env,
#111	capture_output=True,
#112	text=True,
#113	timeout=30,
#114	)
#115	assert stats.returncode == 0, stats.stderr
#116	payload = json.loads(stats.stdout)
#117	assert "error" not in payload
#118	assert payload["working_memory"]["total"] == 1
#119
#120
#121	def test_json_mode_empty_mnemosyne_data_dir_falls_back_to_default(tmp_path):
#122	home = tmp_path / "home"
#123	default_db = home / ".hermes" / "mnemosyne" / "data" / "mnemosyne.db"
#124	env = os.environ.copy()
#125	env["HOME"] = str(home)
#126	env["MNEMOSYNE_DATA_DIR"] = ""
#127
#128	store = subprocess.run(
#129	[sys.executable, "-m", "mnemosyne.cli", "store", "stats empty data dir probe"],
#130	cwd=str(SCRIPT.parent.parent),
#131	env=env,
#132	capture_output=True,
#133	text=True,
#134	timeout=30,
#135	)
#136	assert store.returncode == 0, store.stderr
#137	assert default_db.exists()
#138
#139	stats = subprocess.run(
#140	[sys.executable, str(SCRIPT), "--json"],
#141	cwd=str(SCRIPT.parent),
#142	env=env,
#143	capture_output=True,
#144	text=True,
#145	timeout=30,
#146	)
#147	assert stats.returncode == 0, stats.stderr
#148	payload = json.loads(stats.stdout)
#149	assert "error" not in payload
#150	assert payload["working_memory"]["total"] == 1
#151	assert not (SCRIPT.parent / "mnemosyne.db").exists()
#152
#153	def test_save_snapshot():
#154	code, out, err = run("--save-snapshot")
#155	assert code == 0, f"Exit code {code}: {err}"
#156	assert "Snapshot saved:" in out, "Missing snapshot confirmation"
#157
#158	def test_trends():
#159	code, out, err = run("--trends")
#160	assert code == 0, f"Exit code {code}: {err}"
#161	# Should either show trends or "No trend data yet"
#162	assert "TRENDS" in out or "No trend data" in out, "Unexpected trends output"
#163
#164	def test_auto_snapshot():
#165	"""Full dashboard should auto-save snapshot."""
#166	code, out, err = run()
#167	assert code == 0, f"Exit code {code}: {err}"
#168	# Check snapshot was saved
#169	snaps = sorted(SNAP_DIR.glob("snap_*.json"))
#170	assert len(snaps) >= 1, "No snapshots found after full run"
#171
#172	def test_health_score_in_output():
#173	code, out, err = run()
#174	assert code == 0
#175	assert "Health:" in out, "Missing health score"
#176	assert "/7" in out, "Health score not in X/7 format"
#177
#178	def test_db_size_in_output():
#179	code, out, err = run()
#180	assert code == 0
#181	assert "DB:" in out, "Missing DB size"
#182
#183	def test_quality_indicators_section():
#184	code, out, err = run()
#185	assert code == 0
#186	# Should have at least some indicators
#187	lines = [l for l in out.split('\n') if '✓' in l or '✗' in l]
#188	assert len(lines) >= 5, f"Expected at least 5 quality indicators, got {len(lines)}"
#189
#190	def test_recommendations_section():
#191	code, out, err = run()
#192	assert code == 0
#193	# Should have recommendations (even if "All systems healthy")
#194	assert "RECOMMENDATIONS" in out
#195
#196	# ═══════════════════════════════════════════════════════════
#197	# GROUP 2: Edge Cases (10 tests)
#198	# ═══════════════════════════════════════════════════════════
#199
#200	def test_invalid_flag():
#201	"""Unknown flag should still show dashboard."""
#202	code, out, err = run("--bogus-flag")
#203	assert code == 0, f"Exit code {code} on invalid flag"
#204	assert "MNEMOSYNE HEALTH DASHBOARD" in out
#205
#206	def test_multiple_flags():
#207	"""Multiple flags should not crash."""
#208	code, out, err = run("--compact --json")
#209	assert code == 0, f"Exit code {code} on multiple flags"
#210	# JSON should take precedence
#211	data = json.loads(out)
#212	assert "working_memory" in data
#213
#214	def test_empty_db_path():
#215	"""Script should handle missing DB gracefully."""
#216	# Temporarily rename DB
#217	tmp = DB_PATH.with_suffix(".db.bak")
#218	if DB_PATH.exists():
#219	DB_PATH.rename(tmp)
#220	try:
#221	code, out, err = run("--json")
#222	# Should either error gracefully or return empty data
#223	if code == 0:
#224	data = json.loads(out)
#225	# If it returns data, it should handle missing DB
#226	finally:
#227	tmp.rename(DB_PATH)
#228
#229	def test_corrupted_json_snapshot():
#230	"""Script should handle corrupted snapshot files."""
#231	snap_file = SNAP_DIR / "snap_corrupted.json"
#232	snap_file.write_text("NOT VALID JSON{{{")
#233	try:
#234	code, out, err = run("--trends")
#235	# Should handle gracefully, not crash
#236	assert code == 0, f"Crashed on corrupted snapshot: {err}"
#237	finally:
#238	snap_file.unlink(missing_ok=True)
#239
#240	def test_empty_snapshot_dir():
#241	"""Script should handle empty snapshot directory."""
#242	tmp_dir = SNAP_DIR / "tmp_empty_test"
#243	tmp_dir.mkdir(exist_ok=True)
#244	try:
#245	# The script uses SNAP_DIR, not a configurable path
#246	# So we can't easily test this without mocking
#247	# But we can verify the script doesn't crash with current state
#248	code, out, err = run("--trends")
#249	assert code == 0
#250	finally:
#251	tmp_dir.rmdir()
#252
#253	def test_special_characters_in_memory():
#254	"""SQLite special characters should not crash output."""
#255	code, out, err = run()
#256	assert code == 0
#257	# If any memory has special chars, they should be handled
#258	assert "Traceback" not in err
#259
#260	def test_large_output():
#261	"""Full dashboard should not produce excessively large output."""
#262	code, out, err = run()
#263	assert code == 0
#264	assert len(out) < 100000, f"Output too large: {len(out)} chars"
#265
#266	def test_json_valid_structure():
#267	"""JSON output should have consistent structure."""
#268	code, out, err = run("--json")
#269	assert code == 0
#270	data = json.loads(out)
#271	required_keys = ["db_size_mb", "working_memory", "episodic", "triples",
#272	"consolidation", "dreamer", "embeddings", "wiki", "quality_score"]
#273	for key in required_keys:
#274	assert key in data, f"Missing key: {key}"
#275
#276	def test_snapshot_json_valid():
#277	"""Saved snapshots should be valid JSON."""
#278	code, out, err = run("--save-snapshot")
#279	assert code == 0
#280	snaps = sorted(SNAP_DIR.glob("snap_*.json"))
#281	assert len(snaps) >= 1
#282	with open(snaps[-1]) as f:
#283	data = json.load(f)
#284	assert "timestamp" in data, "Snapshot missing timestamp"
#285	assert "quality_score" in data, "Snapshot missing quality_score"
#286
#287	def test_concurrent_access():
#288	"""Two rapid runs should not corrupt snapshot files."""
#289	code1, _, _ = run("--save-snapshot")
#290	code2, _, _ = run("--save-snapshot")
#291	assert code1 == 0 and code2 == 0
#292	# All snapshots should be valid JSON
#293	for snap in SNAP_DIR.glob("snap_*.json"):
#294	with open(snap) as f:
#295	json.load(f) # Should not raise
#296
#297	# ═══════════════════════════════════════════════════════════
#298	# GROUP 3: Integration Tests (5 tests)
#299	# ═══════════════════════════════════════════════════════════
#300
#301	def test_wiki_count_matches():
#302	"""Wiki page count should match actual files."""
#303	code, out, err = run("--json")
#304	assert code == 0
#305	data = json.loads(out)
#306	actual = len(list(WIKI_PATH.rglob("*.md")))
#307	reported = data["wiki"]["total"]
#308	assert reported == actual, f"Wiki count mismatch: reported={reported}, actual={actual}"
#309
#310	def test_db_count_matches():
#311	"""Working memory count should match actual DB."""
#312	code, out, err = run("--json")
#313	assert code == 0
#314	data = json.loads(out)
#315	db = sqlite3.connect(str(DB_PATH))
#316	actual = _safe_count(db, "working_memory")
#317	db.close()
#318	reported = data["working_memory"]["total"]
#319	assert reported == actual, f"WM count mismatch: reported={reported}, actual={actual}"
#320
#321	def test_episodic_count_matches():
#322	"""Episodic count should match actual DB."""
#323	code, out, err = run("--json")
#324	assert code == 0
#325	data = json.loads(out)
#326	db = sqlite3.connect(str(DB_PATH))
#327	actual = _safe_count(db, "episodic_memory")
#328	db.close()
#329	reported = data["episodic"]["total"]
#330	assert reported == actual, f"Episodic mismatch: reported={reported}, actual={actual}"
#331
#332	def test_triples_count_matches():
#333	"""Triple count should match actual DB."""
#334	code, out, err = run("--json")
#335	assert code == 0
#336	data = json.loads(out)
#337	db = sqlite3.connect(str(DB_PATH))
#338	actual = _safe_count(db, "triples")
#339	db.close()
#340	reported = data["triples"]["total"]
#341	assert reported == actual, f"Triples mismatch: reported={reported}, actual={actual}"
#342
#343	def test_consolidation_count_matches():
#344	"""Consolidation count should match actual DB."""
#345	code, out, err = run("--json")
#346	assert code == 0
#347	data = json.loads(out)
#348	db = sqlite3.connect(str(DB_PATH))
#349	actual = _safe_count(db, "consolidation_log")
#350	db.close()
#351	reported = data["consolidation"]["events"]
#352	assert reported == actual, f"Consolidation mismatch: reported={reported}, actual={actual}"
#353
#354	# ═══════════════════════════════════════════════════════════
#355	# GROUP 4: Stress / Boundary Tests (5 tests)
#356	# ═══════════════════════════════════════════════════════════
#357
#358	def test_rapid_fire():
#359	"""10 rapid runs should all succeed."""
#360	for i in range(10):
#361	code, _, err = run("--compact")
#362	assert code == 0, f"Run {i+1} failed: {err}"
#363
#364	def test_json_pipe_to_python():
#365	"""JSON output should be pipeable to python."""
#366	code, out, err = run("--json")
#367	assert code == 0
#368	# Parse it back
#369	data = json.loads(out)
#370	assert isinstance(data, dict)
#371
#372	def test_output_encoding():
#373	"""Output should be valid UTF-8."""
#374	code, out, err = run()
#375	assert code == 0
#376	# Should not have encoding errors
#377	assert "UnicodeEncodeError" not in err
#378
#379	def test_performance():
#380	"""Dashboard should complete in under 5 seconds."""
#381	import time
#382	start = time.time()
#383	code, _, err = run()
#384	elapsed = time.time() - start
#385	assert code == 0
#386	assert elapsed < 5, f"Dashboard took {elapsed:.1f}s (>5s limit)"
#387
#388	def test_snapshot_growth():
#389	"""Snapshots should not accumulate infinitely (check count)."""
#390	snaps_before = len(list(SNAP_DIR.glob("snap_*.json")))
#391	run("--save-snapshot")
#392	snaps_after = len(list(SNAP_DIR.glob("snap_*.json")))
#393	assert snaps_after == snaps_before + 1, f"Expected +1 snapshot, got {snaps_after - snaps_before}"
#394
#395	# ═══════════════════════════════════════════════════════════
#396	# GROUP 5: Data Integrity Tests (5 tests)
#397	# ═══════════════════════════════════════════════════════════
#398
#399	def test_importance_distribution_sums():
#400	"""Importance distribution should sum to total."""
#401	code, out, err = run("--json")
#402	assert code == 0
#403	data = json.loads(out)
#404	dist_sum = sum(data["working_memory"]["importance_dist"].values())
#405	total = data["working_memory"]["total"]
#406	assert dist_sum == total, f"Distribution sum {dist_sum} != total {total}"
#407
#408	def test_recall_distribution_sums():
#409	"""Recall distribution should sum to total."""
#410	code, out, err = run("--json")
#411	assert code == 0
#412	data = json.loads(out)
#413	dist_sum = sum(data["working_memory"]["recall_dist"].values())
#414	total = data["working_memory"]["total"]
#415	assert dist_sum == total, f"Recall sum {dist_sum} != total {total}"
#416
#417	def test_noise_pct_calculation():
#418	"""Noise percentage should be correctly calculated."""
#419	code, out, err = run("--json")
#420	assert code == 0
#421	data = json.loads(out)
#422	wm = data["working_memory"]
#423	# Verify noise_pct matches manual calculation
#424	db = sqlite3.connect(str(DB_PATH))
#425	noise_count = db.execute(
#426	"SELECT COUNT(*) FROM working_memory WHERE importance<0.3 AND recall_count=0"
#427	).fetchone()[0]
#428	total = db.execute("SELECT COUNT(*) FROM working_memory").fetchone()[0]
#429	db.close()
#430	expected_pct = round(noise_count / total * 100, 1) if total > 0 else 0
#431	assert wm["noise_pct"] == expected_pct, f"Noise mismatch: {wm['noise_pct']} != {expected_pct}"
#432
#433	def test_global_count_accuracy():
#434	"""Global count should match DB."""
#435	code, out, err = run("--json")
#436	assert code == 0
#437	data = json.loads(out)
#438	db = sqlite3.connect(str(DB_PATH))
#439	actual = db.execute("SELECT COUNT(*) FROM working_memory WHERE scope='global'").fetchone()[0]
#440	db.close()
#441	assert data["working_memory"]["global_count"] == actual
#442
#443	def test_quality_score_bounds():
#444	"""Quality score should be between 0 and 7."""
#445	code, out, err = run("--json")
#446	assert code == 0
#447	data = json.loads(out)
#448	score = data["quality_score"]
#449	assert 0 <= score <= 7, f"Quality score out of bounds: {score}"
#450
#451	# ═══════════════════════════════════════════════════════════
#452	# RUN ALL TESTS
#453	# ═══════════════════════════════════════════════════════════
#454
#455	if __name__ == "__main__":
#456	print("=" * 60)
#457	print(" MNEMOSYNE-STATS.PY TEST SUITE")
#458	print(f" {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
#459	print("=" * 60)
#460
#461	print("\n GROUP 1: Normal Operation (10 tests)")
#462	print(" " + "─" * 40)
#463	_test("Full dashboard output", test_full_dashboard)
#464	_test("Compact mode", test_compact_mode)
#465	_test("JSON mode", test_json_mode)
#466	_test("Save snapshot", test_save_snapshot)
#467	_test("Trends display", test_trends)
#468	_test("Auto-snapshot on full run", test_auto_snapshot)
#469	_test("Health score format", test_health_score_in_output)
#470	_test("DB size in output", test_db_size_in_output)
#471	_test("Quality indicators section", test_quality_indicators_section)
#472	_test("Recommendations section", test_recommendations_section)
#473
#474	print("\n GROUP 2: Edge Cases (10 tests)")
#475	print(" " + "─" * 40)
#476	_test("Invalid flag handling", test_invalid_flag)
#477	_test("Multiple flags", test_multiple_flags)
#478	_test("Missing DB path", test_empty_db_path)
#479	_test("Corrupted snapshot file", test_corrupted_json_snapshot)
#480	_test("Empty snapshot dir", test_empty_snapshot_dir)
#481	_test("Special characters in memories", test_special_characters_in_memory)
#482	_test("Output size reasonable", test_large_output)
#483	_test("JSON structure valid", test_json_valid_structure)
#484	_test("Snapshot JSON valid", test_snapshot_json_valid)
#485	_test("Concurrent access safety", test_concurrent_access)
#486
#487	print("\n GROUP 3: Integration (5 tests)")
#488	print(" " + "─" * 40)
#489	_test("Wiki count matches files", test_wiki_count_matches)
#490	_test("WM count matches DB", test_db_count_matches)
#491	_test("Episodic count matches DB", test_episodic_count_matches)
#492	_test("Triples count matches DB", test_triples_count_matches)
#493	_test("Consolidation count matches DB", test_consolidation_count_matches)
#494
#495	print("\n GROUP 4: Stress / Boundary (5 tests)")
#496	print(" " + "─" * 40)
#497	_test("Rapid fire (10 runs)", test_rapid_fire)
#498	_test("JSON pipe to python", test_json_pipe_to_python)
#499	_test("Output encoding (UTF-8)", test_output_encoding)
#500	_test("Performance (<5s)", test_performance)
#501	_test("Snapshot growth tracking", test_snapshot_growth)
#502
#503	print("\n GROUP 5: Data Integrity (5 tests)")
#504	print(" " + "─" * 40)
#505	_test("Importance dist sums to total", test_importance_distribution_sums)
#506	_test("Recall dist sums to total", test_recall_distribution_sums)
#507	_test("Noise % calculation correct", test_noise_pct_calculation)
#508	_test("Global count accuracy", test_global_count_accuracy)
#509	_test("Quality score in bounds", test_quality_score_bounds)
#510
#511	# Summary
#512	print(f"\n{'=' * 60}")
#513	print(f" RESULTS: {passed} passed, {failed} failed, {passed+failed} total")
#514	print(f"{'=' * 60}")
#515
#516	if errors:
#517	print("\n FAILURES:")
#518	for name, err in errors:
#519	print(f" ✗ {name}: {err}")
#520
#521	sys.exit(0 if failed == 0 else 1)
#522

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public