my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Tests for Mnemosyne Plugin Architecture
#3
#4	Validates:
#5	1. PluginManager registration, loading, unloading, listing
#6	2. MnemosynePlugin base class and abstract methods
#7	3. Built-in plugins: LoggingPlugin, MetricsPlugin, FilterPlugin
#8	4. Plugin discovery from filesystem
#9	5. Plugin notification lifecycle
#10	6. Global manager convenience functions
#11	"""
#12
#13	import os
#14	import sys
#15	import pytest
#16	import tempfile
#17	from pathlib import Path
#18	from typing import Any, Dict
#19
#20	sys.path.insert(0, str(Path(__file__).parent.parent))
#21
#22	from mnemosyne.core.plugins import (
#23	MnemosynePlugin,
#24	PluginManager,
#25	LoggingPlugin,
#26	MetricsPlugin,
#27	FilterPlugin,
#28	get_manager,
#29	reset_manager,
#30	DEFAULT_PLUGIN_DIR,
#31	)
#32
#33
#34	# ============================================================================
#35	# Fixtures
#36	# ============================================================================
#37
#38	@pytest.fixture(autouse=True)
#39	def reset_global_manager():
#40	"""Reset the global plugin manager before each test."""
#41	reset_manager()
#42	yield
#43	reset_manager()
#44
#45
#46	@pytest.fixture
#47	def manager():
#48	"""Fresh PluginManager instance."""
#49	return PluginManager()
#50
#51
#52	@pytest.fixture
#53	def sample_memory():
#54	"""A sample memory dict for event testing."""
#55	return {
#56	"id": "mem-123",
#57	"content": "User prefers dark mode in all applications",
#58	"source": "conversation",
#59	"importance": 0.9,
#60	}
#61
#62
#63	@pytest.fixture
#64	def sample_summary():
#65	"""A sample consolidation summary dict."""
#66	return {
#67	"summary": "User prefers dark mode",
#68	"source_wm_ids": ["wm1", "wm2"],
#69	"importance": 0.8,
#70	}
#71
#72
#73	# ============================================================================
#74	# Abstract Base Class
#75	# ============================================================================
#76
#77	class TestMnemosynePlugin:
#78	"""Tests for the abstract base class."""
#79
#80	def test_cannot_instantiate_base(self):
#81	"""MnemosynePlugin is abstract and cannot be instantiated."""
#82	with pytest.raises(TypeError):
#83	MnemosynePlugin()
#84
#85	def test_subclass_must_implement_hooks(self):
#86	"""Subclasses must implement all four lifecycle hooks."""
#87	class PartialPlugin(MnemosynePlugin):
#88	name = "partial"
#89
#90	def on_remember(self, memory):
#91	pass
#92
#93	def on_recall(self, memory):
#94	pass
#95
#96	def on_consolidate(self, summary):
#97	pass
#98
#99	# missing on_invalidate
#100
#101	with pytest.raises(TypeError):
#102	PartialPlugin()
#103
#104	def test_valid_subclass_can_instantiate(self):
#105	"""A fully implemented subclass can be instantiated."""
#106	class ValidPlugin(MnemosynePlugin):
#107	name = "valid"
#108
#109	def on_remember(self, memory):
#110	pass
#111
#112	def on_recall(self, memory):
#113	pass
#114
#115	def on_consolidate(self, summary):
#116	pass
#117
#118	def on_invalidate(self, memory_id):
#119	pass
#120
#121	plugin = ValidPlugin()
#122	assert plugin.name == "valid"
#123	assert plugin.enabled is True
#124	assert plugin._initialized is False
#125
#126	def test_initialize_sets_flag(self):
#127	"""initialize() sets _initialized to True."""
#128	class ValidPlugin(MnemosynePlugin):
#129	name = "valid"
#130
#131	def on_remember(self, memory):
#132	pass
#133
#134	def on_recall(self, memory):
#135	pass
#136
#137	def on_consolidate(self, summary):
#138	pass
#139
#140	def on_invalidate(self, memory_id):
#141	pass
#142
#143	plugin = ValidPlugin()
#144	plugin.initialize()
#145	assert plugin._initialized is True
#146
#147	def test_shutdown_clears_flag(self):
#148	"""shutdown() sets _initialized to False."""
#149	class ValidPlugin(MnemosynePlugin):
#150	name = "valid"
#151
#152	def on_remember(self, memory):
#153	pass
#154
#155	def on_recall(self, memory):
#156	pass
#157
#158	def on_consolidate(self, summary):
#159	pass
#160
#161	def on_invalidate(self, memory_id):
#162	pass
#163
#164	plugin = ValidPlugin()
#165	plugin.initialize()
#166	plugin.shutdown()
#167	assert plugin._initialized is False
#168
#169	def test_to_dict(self):
#170	"""to_dict() returns correct metadata."""
#171	class ValidPlugin(MnemosynePlugin):
#172	name = "valid"
#173	version = "2.0.0"
#174
#175	def on_remember(self, memory):
#176	pass
#177
#178	def on_recall(self, memory):
#179	pass
#180
#181	def on_consolidate(self, summary):
#182	pass
#183
#184	def on_invalidate(self, memory_id):
#185	pass
#186
#187	plugin = ValidPlugin(config={"foo": "bar"})
#188	plugin.initialize()
#189	d = plugin.to_dict()
#190	assert d["name"] == "valid"
#191	assert d["version"] == "2.0.0"
#192	assert d["enabled"] is True
#193	assert d["initialized"] is True
#194	assert d["config"] == {"foo": "bar"}
#195
#196
#197	# ============================================================================
#198	# PluginManager Registration
#199	# ============================================================================
#200
#201	class TestPluginManagerRegistration:
#202	"""Tests for register_plugin and related checks."""
#203
#204	def test_register_valid_plugin(self, manager):
#205	"""register_plugin accepts a valid subclass."""
#206	class TestPlugin(MnemosynePlugin):
#207	name = "test"
#208
#209	def on_remember(self, memory):
#210	pass
#211
#212	def on_recall(self, memory):
#213	pass
#214
#215	def on_consolidate(self, summary):
#216	pass
#217
#218	def on_invalidate(self, memory_id):
#219	pass
#220
#221	manager.register_plugin("test", TestPlugin)
#222	assert manager.is_registered("test")
#223
#224	def test_register_non_subclass_raises(self, manager):
#225	"""register_plugin raises TypeError for non-subclasses."""
#226	with pytest.raises(TypeError):
#227	manager.register_plugin("bad", str)
#228
#229	def test_register_duplicate_raises(self, manager):
#230	"""register_plugin raises ValueError for duplicate names."""
#231	class TestPlugin(MnemosynePlugin):
#232	name = "test"
#233
#234	def on_remember(self, memory):
#235	pass
#236
#237	def on_recall(self, memory):
#238	pass
#239
#240	def on_consolidate(self, summary):
#241	pass
#242
#243	def on_invalidate(self, memory_id):
#244	pass
#245
#246	manager.register_plugin("test", TestPlugin)
#247	with pytest.raises(ValueError, match="already registered"):
#248	manager.register_plugin("test", TestPlugin)
#249
#250	def test_builtins_registered_by_default(self, manager):
#251	"""Built-in plugins are registered on construction."""
#252	assert manager.is_registered("logging")
#253	assert manager.is_registered("metrics")
#254	assert manager.is_registered("filter")
#255
#256	def test_is_registered_false_for_unknown(self, manager):
#257	"""is_registered returns False for unknown plugins."""
#258	assert not manager.is_registered("nonexistent")
#259
#260	def test_is_loaded_false_before_load(self, manager):
#261	"""is_loaded returns False before load_plugin is called."""
#262	assert not manager.is_loaded("logging")
#263
#264
#265	# ============================================================================
#266	# PluginManager Loading / Unloading
#267	# ============================================================================
#268
#269	class TestPluginManagerLoadUnload:
#270	"""Tests for load_plugin, unload_plugin, get_plugin."""
#271
#272	def test_load_plugin(self, manager):
#273	"""load_plugin instantiates and initializes."""
#274	instance = manager.load_plugin("logging")
#275	assert isinstance(instance, LoggingPlugin)
#276	assert manager.is_loaded("logging")
#277	assert instance._initialized is True
#278
#279	def test_load_plugin_with_config(self, manager):
#280	"""load_plugin passes config to the plugin."""
#281	instance = manager.load_plugin("metrics", config={"max_timing_samples": 42})
#282	assert instance.config["max_timing_samples"] == 42
#283
#284	def test_load_unregistered_raises(self, manager):
#285	"""load_plugin raises ValueError for unregistered names."""
#286	with pytest.raises(ValueError, match="not registered"):
#287	manager.load_plugin("unknown")
#288
#289	def test_load_already_loaded_raises(self, manager):
#290	"""load_plugin raises RuntimeError if already loaded."""
#291	manager.load_plugin("filter")
#292	with pytest.raises(RuntimeError, match="already loaded"):
#293	manager.load_plugin("filter")
#294
#295	def test_unload_plugin(self, manager):
#296	"""unload_plugin calls shutdown and removes instance."""
#297	manager.load_plugin("logging")
#298	manager.unload_plugin("logging")
#299	assert not manager.is_loaded("logging")
#300
#301	def test_unload_not_loaded_raises(self, manager):
#302	"""unload_plugin raises ValueError if not loaded."""
#303	with pytest.raises(ValueError, match="not loaded"):
#304	manager.unload_plugin("logging")
#305
#306	def test_get_plugin_returns_instance(self, manager):
#307	"""get_plugin returns the loaded instance."""
#308	loaded = manager.load_plugin("metrics")
#309	assert manager.get_plugin("metrics") is loaded
#310
#311	def test_get_plugin_returns_none(self, manager):
#312	"""get_plugin returns None for unloaded plugins."""
#313	assert manager.get_plugin("metrics") is None
#314
#315	def test_load_all(self, manager):
#316	"""load_all loads every registered plugin."""
#317	loaded = manager.load_all()
#318	assert len(loaded) == 3
#319	assert all(manager.is_loaded(p["name"]) for p in manager.list_plugins())
#320
#321	def test_unload_all(self, manager):
#322	"""unload_all removes every loaded plugin."""
#323	manager.load_all()
#324	manager.unload_all()
#325	assert all(not p["loaded"] for p in manager.list_plugins())
#326
#327	def test_context_manager(self):
#328	"""PluginManager works as a context manager."""
#329	with PluginManager() as mgr:
#330	mgr.load_plugin("logging")
#331	assert mgr.is_loaded("logging")
#332	assert not mgr.is_loaded("logging")
#333
#334
#335	# ============================================================================
#336	# PluginManager Listing
#337	# ============================================================================
#338
#339	class TestPluginManagerList:
#340	"""Tests for list_plugins."""
#341
#342	def test_list_plugins_structure(self, manager):
#343	"""list_plugins returns correct dict structure."""
#344	plugins = manager.list_plugins()
#345	assert len(plugins) == 3
#346	for p in plugins:
#347	assert "name" in p
#348	assert "class" in p
#349	assert "loaded" in p
#350	assert "instance" in p
#351
#352	def test_list_plugins_loaded_state(self, manager):
#353	"""list_plugins reflects loaded vs unloaded state."""
#354	manager.load_plugin("logging")
#355	plugins = manager.list_plugins()
#356	logging_entry = next(p for p in plugins if p["name"] == "logging")
#357	assert logging_entry["loaded"] is True
#358	assert isinstance(logging_entry["instance"], LoggingPlugin)
#359
#360	metrics_entry = next(p for p in plugins if p["name"] == "metrics")
#361	assert metrics_entry["loaded"] is False
#362	assert metrics_entry["instance"] is None
#363
#364
#365	# ============================================================================
#366	# Built-in LoggingPlugin
#367	# ============================================================================
#368
#369	class TestLoggingPlugin:
#370	"""Tests for LoggingPlugin behavior."""
#371
#372	def test_on_remember_logs(self, manager, sample_memory):
#373	"""on_remember creates a log entry."""
#374	plugin = manager.load_plugin("logging")
#375	plugin.clear_log()
#376	plugin.on_remember(sample_memory)
#377	log = plugin.get_log()
#378	assert len(log) == 1
#379	assert log[0]["event"] == "remember"
#380	assert log[0]["memory_id"] == "mem-123"
#381
#382	def test_on_recall_logs(self, manager, sample_memory):
#383	"""on_recall creates a log entry."""
#384	plugin = manager.load_plugin("logging")
#385	plugin.clear_log()
#386	plugin.on_recall(sample_memory)
#387	log = plugin.get_log()
#388	assert len(log) == 1
#389	assert log[0]["event"] == "recall"
#390
#391	def test_on_consolidate_logs(self, manager, sample_summary):
#392	"""on_consolidate creates a log entry."""
#393	plugin = manager.load_plugin("logging")
#394	plugin.clear_log()
#395	plugin.on_consolidate(sample_summary)
#396	log = plugin.get_log()
#397	assert len(log) == 1
#398	assert log[0]["event"] == "consolidate"
#399	assert log[0]["source_count"] == 2
#400
#401	def test_on_invalidate_logs(self, manager):
#402	"""on_invalidate creates a log entry."""
#403	plugin = manager.load_plugin("logging")
#404	plugin.clear_log()
#405	plugin.on_invalidate("mem-123")
#406	log = plugin.get_log()
#407	assert len(log) == 1
#408	assert log[0]["event"] == "invalidate"
#409	assert log[0]["memory_id"] == "mem-123"
#410
#411	def test_log_max_entries(self, manager):
#412	"""Log respects max_entries limit."""
#413	plugin = manager.load_plugin("logging", config={"max_entries": 3})
#414	plugin.clear_log()
#415	for i in range(5):
#416	plugin.on_remember({"id": f"m{i}", "content": f"memory {i}"})
#417	assert len(plugin.get_log()) == 3
#418	assert plugin.get_log()[0]["memory_id"] == "m2"
#419
#420	def test_clear_log(self, manager, sample_memory):
#421	"""clear_log empties the log."""
#422	plugin = manager.load_plugin("logging")
#423	plugin.on_remember(sample_memory)
#424	plugin.clear_log()
#425	assert len(plugin.get_log()) == 0
#426
#427	def test_preview_truncation(self, manager):
#428	"""Long content is truncated in previews."""
#429	plugin = manager.load_plugin("logging")
#430	preview = plugin._preview("x" * 200)
#431	assert preview.endswith("...")
#432	assert len(preview) <= 83
#433
#434
#435	# ============================================================================
#436	# Built-in MetricsPlugin
#437	# ============================================================================
#438
#439	class TestMetricsPlugin:
#440	"""Tests for MetricsPlugin behavior."""
#441
#442	def test_counters_increment(self, manager, sample_memory, sample_summary):
#443	"""Counters increment on each event."""
#444	plugin = manager.load_plugin("metrics")
#445	plugin.on_remember(sample_memory)
#446	plugin.on_recall(sample_memory)
#447	plugin.on_consolidate(sample_summary)
#448	plugin.on_invalidate("m1")
#449	counters = plugin.get_counters()
#450	assert counters["remember"] == 1
#451	assert counters["recall"] == 1
#452	assert counters["consolidate"] == 1
#453	assert counters["invalidate"] == 1
#454
#455	def test_record_timing(self, manager):
#456	"""record_timing stores duration samples."""
#457	plugin = manager.load_plugin("metrics")
#458	plugin.record_timing("remember", 12.5)
#459	plugin.record_timing("remember", 7.5)
#460	assert plugin.get_timings("remember") == [12.5, 7.5]
#461
#462	def test_average_timing(self, manager):
#463	"""get_average_timing computes the mean."""
#464	plugin = manager.load_plugin("metrics")
#465	plugin.record_timing("recall", 10.0)
#466	plugin.record_timing("recall", 20.0)
#467	assert plugin.get_average_timing("recall") == 15.0
#468
#469	def test_average_timing_none(self, manager):
#470	"""get_average_timing returns None with no samples."""
#471	plugin = manager.load_plugin("metrics")
#472	assert plugin.get_average_timing("recall") is None
#473
#474	def test_timing_max_samples(self, manager):
#475	"""Timing list respects max_timing_samples."""
#476	plugin = manager.load_plugin("metrics", config={"max_timing_samples": 2})
#477	for i in range(5):
#478	plugin.record_timing("remember", float(i))
#479	assert len(plugin.get_timings("remember")) == 2
#480	assert plugin.get_timings("remember") == [3.0, 4.0]
#481
#482	def test_reset(self, manager, sample_memory):
#483	"""reset clears counters and timings."""
#484	plugin = manager.load_plugin("metrics")
#485	plugin.on_remember(sample_memory)
#486	plugin.record_timing("remember", 5.0)
#487	plugin.reset()
#488	assert plugin.get_counters()["remember"] == 0
#489	assert plugin.get_timings("remember") == []
#490
#491	def test_get_summary(self, manager, sample_memory):
#492	"""get_summary returns counters and averages."""
#493	plugin = manager.load_plugin("metrics")
#494	plugin.on_remember(sample_memory)
#495	plugin.record_timing("remember", 10.0)
#496	summary = plugin.get_summary()
#497	assert summary["counters"]["remember"] == 1
#498	assert summary["averages"]["remember"] == 10.0
#499
#500
#501	# ============================================================================
#502	# Built-in FilterPlugin
#503	# ============================================================================
#504
#505	class TestFilterPlugin:
#506	"""Tests for FilterPlugin behavior."""
#507
#508	def test_add_remove_rule(self, manager):
#509	"""Rules can be added and removed."""
#510	plugin = manager.load_plugin("filter")
#511	rule = lambda m: True
#512	plugin.add_rule(rule)
#513	assert rule in plugin._rules
#514	plugin.remove_rule(rule)
#515	assert rule not in plugin._rules
#516
#517	def test_clear_rules(self, manager):
#518	"""clear_rules removes all rules."""
#519	plugin = manager.load_plugin("filter")
#520	plugin.add_rule(lambda m: True)
#521	plugin.add_rule(lambda m: False)
#522	plugin.clear_rules()
#523	assert len(plugin._rules) == 0
#524
#525	def test_passes_with_no_rules(self, manager, sample_memory):
#526	"""Without rules, everything passes."""
#527	plugin = manager.load_plugin("filter")
#528	assert plugin._passes(sample_memory) is True
#529
#530	def test_passes_with_allow_rule(self, manager, sample_memory):
#531	"""A rule returning True allows the memory."""
#532	plugin = manager.load_plugin("filter")
#533	plugin.add_rule(lambda m: m.get("importance", 0) > 0.5)
#534	assert plugin._passes(sample_memory) is True
#535
#536	def test_passes_with_block_rule(self, manager, sample_memory):
#537	"""A rule returning False blocks the memory."""
#538	plugin = manager.load_plugin("filter")
#539	plugin.add_rule(lambda m: m.get("importance", 0) > 0.95)
#540	assert plugin._passes(sample_memory) is False
#541
#542	def test_blocked_tracked(self, manager, sample_memory):
#543	"""Blocked memories are tracked."""
#544	plugin = manager.load_plugin("filter")
#545	plugin.add_rule(lambda m: False)
#546	plugin.on_remember(sample_memory)
#547	blocked = plugin.get_blocked()
#548	assert len(blocked) == 1
#549	assert blocked[0]["item"]["id"] == "mem-123"
#550
#551	def test_is_blocked(self, manager, sample_memory):
#552	"""is_blocked checks by memory ID."""
#553	plugin = manager.load_plugin("filter")
#554	plugin.add_rule(lambda m: False)
#555	plugin.on_remember(sample_memory)
#556	assert plugin.is_blocked("mem-123") is True
#557	assert plugin.is_blocked("other") is False
#558
#559	def test_exception_in_rule_blocks(self, manager, sample_memory):
#560	"""An exception in a rule treats the memory as blocked."""
#561	plugin = manager.load_plugin("filter")
#562	plugin.add_rule(lambda m: 1 / 0)
#563	assert plugin._passes(sample_memory) is False
#564
#565	def test_max_blocked(self, manager):
#566	"""Blocked list respects max_blocked limit."""
#567	plugin = manager.load_plugin("filter", config={"max_blocked": 2})
#568	plugin.add_rule(lambda m: False)
#569	for i in range(5):
#570	plugin.on_remember({"id": f"m{i}", "content": f"c{i}"})
#571	assert len(plugin.get_blocked()) == 2
#572
#573
#574	# ============================================================================
#575	# Plugin Notifications
#576	# ============================================================================
#577
#578	class TestPluginNotifications:
#579	"""Tests for notify_* methods on PluginManager."""
#580
#581	def test_notify_remember(self, manager, sample_memory):
#582	"""notify_remember reaches all loaded plugins."""
#583	logging_plugin = manager.load_plugin("logging")
#584	logging_plugin.clear_log()
#585	manager.notify_remember(sample_memory)
#586	assert len(logging_plugin.get_log()) == 1
#587
#588	def test_notify_recall(self, manager, sample_memory):
#589	"""notify_recall reaches all loaded plugins."""
#590	metrics_plugin = manager.load_plugin("metrics")
#591	manager.notify_recall(sample_memory)
#592	assert metrics_plugin.get_counters()["recall"] == 1
#593
#594	def test_notify_consolidate(self, manager, sample_summary):
#595	"""notify_consolidate reaches all loaded plugins."""
#596	logging_plugin = manager.load_plugin("logging")
#597	logging_plugin.clear_log()
#598	manager.notify_consolidate(sample_summary)
#599	assert len(logging_plugin.get_log()) == 1
#600
#601	def test_notify_invalidate(self, manager):
#602	"""notify_invalidate reaches all loaded plugins."""
#603	metrics_plugin = manager.load_plugin("metrics")
#604	manager.notify_invalidate("mid-1")
#605	assert metrics_plugin.get_counters()["invalidate"] == 1
#606
#607	def test_disabled_plugin_skipped(self, manager, sample_memory):
#608	"""Disabled plugins are skipped during notification."""
#609	plugin = manager.load_plugin("logging")
#610	plugin.clear_log()
#611	plugin.enabled = False
#612	manager.notify_remember(sample_memory)
#613	assert len(plugin.get_log()) == 0
#614
#615	def test_plugin_error_does_not_propagate(self, manager, sample_memory):
#616	"""Exceptions in plugins are caught and logged, not propagated."""
#617	class BadPlugin(MnemosynePlugin):
#618	name = "bad"
#619
#620	def on_remember(self, memory):
#621	raise RuntimeError("boom")
#622
#623	def on_recall(self, memory):
#624	pass
#625
#626	def on_consolidate(self, summary):
#627	pass
#628
#629	def on_invalidate(self, memory_id):
#630	pass
#631
#632	manager.register_plugin("bad", BadPlugin)
#633	manager.load_plugin("bad")
#634	# Should not raise
#635	manager.notify_remember(sample_memory)
#636
#637
#638	# ============================================================================
#639	# Plugin Discovery
#640	# ============================================================================
#641
#642	class TestPluginDiscovery:
#643	"""Tests for discover_plugins."""
#644
#645	def test_discover_empty_dir(self, manager):
#646	"""Discover on non-existent directory returns empty list."""
#647	with tempfile.TemporaryDirectory() as tmpdir:
#648	mgr = PluginManager(plugin_dir=Path(tmpdir) / "empty")
#649	discovered = mgr.discover_plugins()
#650	assert discovered == []
#651
#652	def test_discover_valid_plugin(self, manager):
#653	"""Discover registers a valid plugin from a .py file."""
#654	with tempfile.TemporaryDirectory() as tmpdir:
#655	plugin_file = Path(tmpdir) / "my_plugin.py"
#656	plugin_file.write_text(
#657	"from mnemosyne.core.plugins import MnemosynePlugin\n"
#658	"class MyPlugin(MnemosynePlugin):\n"
#659	" name = 'myplugin'\n"
#660	" def on_remember(self, memory): pass\n"
#661	" def on_recall(self, memory): pass\n"
#662	" def on_consolidate(self, summary): pass\n"
#663	" def on_invalidate(self, memory_id): pass\n"
#664	)
#665	mgr = PluginManager(plugin_dir=Path(tmpdir))
#666	discovered = mgr.discover_plugins()
#667	assert "myplugin" in discovered
#668	assert mgr.is_registered("myplugin")
#669
#670	def test_discover_ignores_underscore_files(self, manager):
#671	"""Files starting with _ are ignored during discovery."""
#672	with tempfile.TemporaryDirectory() as tmpdir:
#673	plugin_file = Path(tmpdir) / "_private.py"
#674	plugin_file.write_text(
#675	"from mnemosyne.core.plugins import MnemosynePlugin\n"
#676	"class PrivatePlugin(MnemosynePlugin):\n"
#677	" name = 'private'\n"
#678	" def on_remember(self, memory): pass\n"
#679	" def on_recall(self, memory): pass\n"
#680	" def on_consolidate(self, summary): pass\n"
#681	" def on_invalidate(self, memory_id): pass\n"
#682	)
#683	mgr = PluginManager(plugin_dir=Path(tmpdir))
#684	discovered = mgr.discover_plugins()
#685	assert "private" not in discovered
#686
#687	def test_discover_does_not_duplicate(self, manager):
#688	"""Already-registered plugins are not duplicated."""
#689	with tempfile.TemporaryDirectory() as tmpdir:
#690	plugin_file = Path(tmpdir) / "logging.py"
#691	plugin_file.write_text(
#692	"from mnemosyne.core.plugins import MnemosynePlugin\n"
#693	"class LoggingPlugin(MnemosynePlugin):\n"
#694	" name = 'logging'\n"
#695	" def on_remember(self, memory): pass\n"
#696	" def on_recall(self, memory): pass\n"
#697	" def on_consolidate(self, summary): pass\n"
#698	" def on_invalidate(self, memory_id): pass\n"
#699	)
#700	mgr = PluginManager(plugin_dir=Path(tmpdir))
#701	discovered = mgr.discover_plugins()
#702	assert "logging" not in discovered # already registered as builtin
#703
#704	def test_discover_bad_file_graceful(self, manager):
#705	"""Bad Python files are skipped gracefully."""
#706	with tempfile.TemporaryDirectory() as tmpdir:
#707	plugin_file = Path(tmpdir) / "broken.py"
#708	plugin_file.write_text("this is not valid python!!!")
#709	mgr = PluginManager(plugin_dir=Path(tmpdir))
#710	discovered = mgr.discover_plugins()
#711	assert discovered == []
#712
#713
#714	# ============================================================================
#715	# Global Manager
#716	# ============================================================================
#717
#718	class TestGlobalManager:
#719	"""Tests for get_manager and reset_manager."""
#720
#721	def test_get_manager_singleton(self):
#722	"""get_manager returns the same instance."""
#723	mgr1 = get_manager()
#724	mgr2 = get_manager()
#725	assert mgr1 is mgr2
#726
#727	def test_reset_manager_creates_new(self):
#728	"""reset_manager creates a new instance on next get_manager."""
#729	mgr1 = get_manager()
#730	reset_manager()
#731	mgr2 = get_manager()
#732	assert mgr1 is not mgr2
#733
#734	def test_reset_manager_unloads(self):
#735	"""reset_manager unloads existing plugins."""
#736	mgr = get_manager()
#737	mgr.load_plugin("logging")
#738	reset_manager()
#739	mgr2 = get_manager()
#740	assert not mgr2.is_loaded("logging")
#741
#742
#743	# ============================================================================
#744	# Integration with Mnemosyne memory class (optional hook points)
#745	# ============================================================================
#746
#747	class TestPluginIntegration:
#748	"""Tests showing how plugins integrate with memory operations."""
#749
#750	def test_plugin_manager_can_be_attached_to_memory(self, manager):
#751	"""A PluginManager can be stored and used alongside Mnemosyne."""
#752	# This demonstrates the pattern: Mnemosyne can hold a PluginManager
#753	# and call notify_* at key lifecycle points.
#754	from mnemosyne.core.memory import Mnemosyne as Memory
#755	with tempfile.TemporaryDirectory() as tmpdir:
#756	db_path = Path(tmpdir) / "test.db"
#757	mem = Memory(session_id="s1", db_path=db_path)
#758	# Attach plugin manager
#759	mem.plugins = manager
#760	manager.load_plugin("logging")
#761	manager.load_plugin("metrics")
#762	assert hasattr(mem, "plugins")
#763	assert manager.is_loaded("logging")
#764	assert manager.is_loaded("metrics")
#765

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public