cleveragents-core/robot/helper_plugin_architecture.py

"""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 <command>
"""

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 <command>")
        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())