"""Robot Framework helper for Plugin Architecture integration tests. Provides a CLI-style interface for Robot to invoke plugin loading, validation, lifecycle, and DI resolution. Exit code 0 = success, 1 = failure. Usage: python robot/helper_plugin_architecture.py """ from __future__ import annotations import sys import threading from pathlib import Path # Ensure the src directory is on the import path. _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) from cleveragents.domain.models.acms.backends import ( # noqa: E402 TextBackend, TextResult, ) from cleveragents.infrastructure.plugins.exceptions import ( # noqa: E402 PluginError, PluginLoadError, PluginNotFoundError, ProtocolMismatchError, ) from cleveragents.infrastructure.plugins.loader import PluginLoader # noqa: E402 from cleveragents.infrastructure.plugins.manager import PluginManager # noqa: E402 from cleveragents.infrastructure.plugins.types import ( # noqa: E402 ExtensionPoint, PluginDescriptor, PluginState, ) class _FakeTextBackend: """Minimal class satisfying the TextBackend protocol.""" def search( self, query: str, *, scope: frozenset[str], max_results: int = 20, ) -> list[TextResult]: return [] class _NonProtocolClass: """A class that does not implement any backend protocol.""" pass def main() -> int: """Entry point called by Robot Framework ``Run Process``.""" if len(sys.argv) < 2: print("Usage: helper_plugin_architecture.py ") return 1 command: str = sys.argv[1] if command == "load-class": try: loader = PluginLoader() cls = loader.load_class( "cleveragents.domain.models.acms.stubs", "InMemoryTextBackend", ) assert cls is not None assert cls.__name__ == "InMemoryTextBackend" print("plugin-load-class-ok") return 0 except Exception as exc: print(f"plugin-load-class-fail: {exc}") return 1 if command == "prefix-security": try: loader = PluginLoader() raised = False try: loader.load_class("pathlib", "Path") except PluginLoadError: raised = True assert raised, "Expected PluginLoadError for disallowed prefix" print("plugin-prefix-security-ok") return 0 except Exception as exc: print(f"plugin-prefix-security-fail: {exc}") return 1 if command == "protocol-pass": try: result = PluginLoader.validate_protocol(_FakeTextBackend, TextBackend) assert result is True print("plugin-protocol-pass-ok") return 0 except Exception as exc: print(f"plugin-protocol-pass-fail: {exc}") return 1 if command == "protocol-fail": try: raised = False try: PluginLoader.validate_protocol(_NonProtocolClass, TextBackend) except ProtocolMismatchError: raised = True assert raised, "Expected ProtocolMismatchError" print("plugin-protocol-fail-ok") return 0 except Exception as exc: print(f"plugin-protocol-fail-fail: {exc}") return 1 if command == "lifecycle": try: manager = PluginManager() desc = PluginDescriptor( name="lifecycle-test", module_path="cleveragents.domain.models.acms.stubs", class_name="InMemoryTextBackend", ) manager.register_plugin(desc) assert manager.get_plugin("lifecycle-test").state == PluginState.DISCOVERED manager.activate_plugin("lifecycle-test") assert manager.get_plugin("lifecycle-test").state == PluginState.ACTIVATED assert manager.get_plugin_class("lifecycle-test") is not None assert manager.get_plugin_instance("lifecycle-test") is not None manager.deactivate_plugin("lifecycle-test") assert manager.get_plugin("lifecycle-test").state == PluginState.DEACTIVATED assert manager.get_plugin_class("lifecycle-test") is None assert manager.get_plugin_instance("lifecycle-test") is None print("plugin-lifecycle-ok") return 0 except Exception as exc: print(f"plugin-lifecycle-fail: {exc}") return 1 if command == "config-registration": try: manager = PluginManager() config = { "custom_module": "cleveragents.domain.models.acms.stubs", "custom_class": "InMemoryTextBackend", "name": "config-test", } result = manager.register_from_config(config) assert result is not None assert result.name == "config-test" # Empty config should return None empty = manager.register_from_config({}) assert empty is None print("plugin-config-registration-ok") return 0 except Exception as exc: print(f"plugin-config-registration-fail: {exc}") return 1 if command == "error-handling": try: manager = PluginManager() # PluginNotFoundError raised_not_found = False try: manager.get_plugin("nonexistent") except PluginNotFoundError: raised_not_found = True assert raised_not_found # PluginLoadError on bad module desc = PluginDescriptor( name="bad-module", module_path="cleveragents.nonexistent_xyz", class_name="BadClass", ) manager.register_plugin(desc) raised_load = False try: manager.activate_plugin("bad-module") except PluginLoadError: raised_load = True assert raised_load assert manager.get_plugin("bad-module").state == PluginState.ERRORED # Duplicate registration desc2 = PluginDescriptor( name="dup-test", module_path="cleveragents.domain.models.acms.stubs", class_name="InMemoryTextBackend", ) manager.register_plugin(desc2) raised_dup = False try: manager.register_plugin(desc2) except PluginError: raised_dup = True assert raised_dup print("plugin-error-handling-ok") return 0 except Exception as exc: print(f"plugin-error-handling-fail: {exc}") return 1 if command == "di-resolution": try: from cleveragents.application.container import ( get_container, reset_container, ) reset_container() container = get_container() pm1 = container.plugin_manager() assert isinstance(pm1, PluginManager) pm2 = container.plugin_manager() assert pm1 is pm2, "PluginManager should be singleton" print("plugin-di-resolution-ok") return 0 except Exception as exc: print(f"plugin-di-resolution-fail: {exc}") return 1 if command == "thread-safety": try: manager = PluginManager() errors: list[str] = [] def register_plugin(idx: int) -> None: try: desc = PluginDescriptor( name=f"thread-{idx}", module_path="cleveragents.domain.models.acms.stubs", class_name="InMemoryTextBackend", ) manager.register_plugin(desc) except Exception as exc: errors.append(str(exc)) threads = [ threading.Thread(target=register_plugin, args=(i,)) for i in range(20) ] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Thread errors: {errors}" assert len(manager.list_plugins()) == 20 print("plugin-thread-safety-ok") return 0 except Exception as exc: print(f"plugin-thread-safety-fail: {exc}") return 1 if command == "extension-points": try: manager = PluginManager() ep = ExtensionPoint( name="test-ep", protocol_type=TextBackend, description="Test extension point", ) manager.register_extension_point(ep) eps = manager.list_extension_points() assert len(eps) == 1 assert eps[0].name == "test-ep" print("plugin-extension-points-ok") return 0 except Exception as exc: print(f"plugin-extension-points-fail: {exc}") return 1 if command == "entry-point-security": # Security regression test for issue #7476. # Verifies that load_from_entry_points does NOT call ep.load() for # entry points whose module prefix is not in the allowlist. # Uses importlib.metadata.EntryPoint mocking via a real PluginLoader # instance with a controlled entry_points() override. try: import importlib.metadata import unittest.mock as mock loader = PluginLoader() # Build a fake entry point that references a disallowed module. fake_ep = mock.MagicMock() fake_ep.name = "malicious" fake_ep.value = "maliciouspkg.attack:Exploit" fake_ep.group = "cleveragents.plugins" # Patch importlib.metadata.entry_points to return our fake entry point. fake_eps = mock.MagicMock() fake_eps.select.return_value = [fake_ep] with mock.patch.object( importlib.metadata, "entry_points", return_value=fake_eps, ): descriptors = loader.load_from_entry_points("cleveragents.plugins") # The disallowed entry point must be skipped — no descriptors returned. assert descriptors == [], ( f"Expected empty descriptor list, got: {descriptors}" ) # ep.load() must NEVER have been called for the disallowed entry point. fake_ep.load.assert_not_called() print("plugin-entry-point-security-ok") return 0 except Exception as exc: print(f"plugin-entry-point-security-fail: {exc}") return 1 print(f"Unknown command: {command}") return 1 if __name__ == "__main__": sys.exit(main())