HAL9000
717a05f604
Parse entry point targets before import so allowlist enforcement happens prior to execution and add a Behave regression scenario covering the disallowed-prefix path. ISSUES CLOSED: #7476
873 lines
29 KiB
Python
873 lines
29 KiB
Python
"""Behave step definitions for the Plugin Architecture Framework.
|
|
|
|
Covers PluginState, ExtensionPoint, PluginDescriptor, PluginLoader,
|
|
PluginManager, protocol validation, entry-point discovery, config-driven
|
|
registration, thread safety, and DI container integration.
|
|
|
|
Based on issue #585.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import threading
|
|
from typing import Any
|
|
from unittest.mock import ANY, MagicMock, patch
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.domain.models.acms.backends import TextBackend, TextResult
|
|
from cleveragents.infrastructure.plugins.exceptions import (
|
|
PluginError,
|
|
PluginLoadError,
|
|
PluginNotFoundError,
|
|
ProtocolMismatchError,
|
|
)
|
|
from cleveragents.infrastructure.plugins.loader import PluginLoader
|
|
from cleveragents.infrastructure.plugins.manager import PluginManager
|
|
from cleveragents.infrastructure.plugins.types import (
|
|
ExtensionPoint,
|
|
PluginDescriptor,
|
|
PluginState,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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."""
|
|
|
|
def unrelated_method(self) -> str:
|
|
return "nope"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PluginState enum
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the PluginState enum is available")
|
|
def step_plugin_state_available(context: Context) -> None:
|
|
context.plugin_state = PluginState
|
|
|
|
|
|
@then(
|
|
'it should have values "discovered", "activated", "executing", "deactivated", "errored"'
|
|
)
|
|
def step_plugin_state_values(context: Context) -> None:
|
|
expected = {"discovered", "activated", "executing", "deactivated", "errored"}
|
|
actual = {s.value for s in PluginState}
|
|
assert actual == expected, f"Expected {expected}, got {actual}"
|
|
|
|
|
|
@then("each PluginState value should be a string")
|
|
def step_plugin_state_is_str(context: Context) -> None:
|
|
for state in PluginState:
|
|
assert isinstance(state.value, str), f"{state} value is not a string"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ExtensionPoint model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('I create an ExtensionPoint with name "{name}" and a protocol type')
|
|
def step_create_extension_point(context: Context, name: str) -> None:
|
|
context.ext_point = ExtensionPoint(
|
|
name=name,
|
|
protocol_type=TextBackend,
|
|
description="Test extension point",
|
|
registry_key="test_key",
|
|
)
|
|
|
|
|
|
@then('the ExtensionPoint name should be "{name}"')
|
|
def step_ext_point_name(context: Context, name: str) -> None:
|
|
assert context.ext_point.name == name
|
|
|
|
|
|
@then("the ExtensionPoint should have a protocol_type")
|
|
def step_ext_point_protocol(context: Context) -> None:
|
|
assert context.ext_point.protocol_type is not None
|
|
|
|
|
|
@then("attempting to mutate the ExtensionPoint name should raise an error")
|
|
def step_ext_point_frozen(context: Context) -> None:
|
|
raised = False
|
|
try:
|
|
context.ext_point.name = "mutated" # type: ignore[misc]
|
|
except (ValidationError, TypeError):
|
|
raised = True
|
|
assert raised, "Expected frozen model to reject mutation"
|
|
|
|
|
|
@when("I attempt to create an ExtensionPoint with an empty name")
|
|
def step_create_ext_point_empty_name(context: Context) -> None:
|
|
context.raised_validation_error = False
|
|
try:
|
|
ExtensionPoint(name="", protocol_type=TextBackend)
|
|
except ValidationError:
|
|
context.raised_validation_error = True
|
|
|
|
|
|
@then("a plugin validation error should be raised")
|
|
def step_plugin_validation_error_raised(context: Context) -> None:
|
|
assert context.raised_validation_error, "Expected ValidationError"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PluginDescriptor model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('I create a PluginDescriptor with name "{name}"')
|
|
def step_create_descriptor(context: Context, name: str) -> None:
|
|
context.descriptor = PluginDescriptor(name=name)
|
|
|
|
|
|
@given("I create a PluginDescriptor with full metadata")
|
|
def step_create_descriptor_full(context: Context) -> None:
|
|
context.descriptor = PluginDescriptor(
|
|
name="full-plugin",
|
|
version="1.2.3",
|
|
author="TestAuthor",
|
|
description="A fully described plugin",
|
|
module_path="cleveragents.test",
|
|
class_name="TestClass",
|
|
extension_points=["ep1"],
|
|
dependencies=["dep1"],
|
|
)
|
|
|
|
|
|
@then('the descriptor state should be "{state}"')
|
|
def step_descriptor_state(context: Context, state: str) -> None:
|
|
actual = (
|
|
context.descriptor.state.value
|
|
if hasattr(context.descriptor.state, "value")
|
|
else context.descriptor.state
|
|
)
|
|
assert actual == state, f"Expected state '{state}', got '{actual}'"
|
|
|
|
|
|
@then('the descriptor version should be "{version}"')
|
|
def step_descriptor_version(context: Context, version: str) -> None:
|
|
assert context.descriptor.version == version
|
|
|
|
|
|
@then("the descriptor dependencies should be empty")
|
|
def step_descriptor_deps_empty(context: Context) -> None:
|
|
assert context.descriptor.dependencies == []
|
|
|
|
|
|
@then('the descriptor should have name "{name}"')
|
|
def step_descriptor_has_name(context: Context, name: str) -> None:
|
|
assert context.descriptor.name == name
|
|
|
|
|
|
@then('the descriptor should have version "{version}"')
|
|
def step_descriptor_has_version(context: Context, version: str) -> None:
|
|
assert context.descriptor.version == version
|
|
|
|
|
|
@then('the descriptor should have author "{author}"')
|
|
def step_descriptor_has_author(context: Context, author: str) -> None:
|
|
assert context.descriptor.author == author
|
|
|
|
|
|
@then('the descriptor should have module_path "{path}"')
|
|
def step_descriptor_has_module_path(context: Context, path: str) -> None:
|
|
assert context.descriptor.module_path == path
|
|
|
|
|
|
@then('the descriptor should have class_name "{cls}"')
|
|
def step_descriptor_has_class_name(context: Context, cls: str) -> None:
|
|
assert context.descriptor.class_name == cls
|
|
|
|
|
|
@when('I set the descriptor state to "{state}"')
|
|
def step_set_descriptor_state(context: Context, state: str) -> None:
|
|
context.descriptor.state = PluginState(state)
|
|
|
|
|
|
@when("I attempt to create a PluginDescriptor with an empty name")
|
|
def step_create_descriptor_empty(context: Context) -> None:
|
|
context.raised_validation_error = False
|
|
try:
|
|
PluginDescriptor(name="")
|
|
except ValidationError:
|
|
context.raised_validation_error = True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plugin exceptions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("PluginLoadError should be a subclass of PluginError")
|
|
def step_load_error_subclass(context: Context) -> None:
|
|
assert issubclass(PluginLoadError, PluginError)
|
|
|
|
|
|
@then("PluginNotFoundError should be a subclass of PluginError")
|
|
def step_not_found_subclass(context: Context) -> None:
|
|
assert issubclass(PluginNotFoundError, PluginError)
|
|
|
|
|
|
@then("ProtocolMismatchError should be a subclass of PluginError")
|
|
def step_mismatch_subclass(context: Context) -> None:
|
|
assert issubclass(ProtocolMismatchError, PluginError)
|
|
|
|
|
|
@when('I raise a PluginLoadError with message "{msg}"')
|
|
def step_raise_load_error(context: Context, msg: str) -> None:
|
|
try:
|
|
raise PluginLoadError(msg)
|
|
except PluginLoadError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when('I raise a PluginNotFoundError with message "{msg}"')
|
|
def step_raise_not_found_error(context: Context, msg: str) -> None:
|
|
try:
|
|
raise PluginNotFoundError(msg)
|
|
except PluginNotFoundError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when('I raise a ProtocolMismatchError with message "{msg}"')
|
|
def step_raise_mismatch_error(context: Context, msg: str) -> None:
|
|
try:
|
|
raise ProtocolMismatchError(msg)
|
|
except ProtocolMismatchError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then('the exception message should contain "{text}"')
|
|
def step_exception_contains(context: Context, text: str) -> None:
|
|
assert text in str(context.caught_exception), (
|
|
f"Expected '{text}' in '{context.caught_exception}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PluginLoader — load_class
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a PluginLoader with default prefixes")
|
|
def step_loader_default(context: Context) -> None:
|
|
context.loader = PluginLoader()
|
|
|
|
|
|
@given('a PluginLoader with allowed prefixes "{prefixes}"')
|
|
def step_loader_custom_prefixes(context: Context, prefixes: str) -> None:
|
|
prefix_tuple = tuple(p.strip() for p in prefixes.split(","))
|
|
context.loader = PluginLoader(allowed_prefixes=prefix_tuple)
|
|
|
|
|
|
@given("a PluginLoader with empty prefix allowlist")
|
|
def step_loader_empty_prefixes(context: Context) -> None:
|
|
context.loader = PluginLoader(allowed_prefixes=())
|
|
|
|
|
|
@when('I load class "{cls}" from module "{mod}"')
|
|
def step_load_class(context: Context, cls: str, mod: str) -> None:
|
|
context.loaded_class = context.loader.load_class(mod, cls)
|
|
|
|
|
|
@when('I attempt to load class "{cls}" from module "{mod}"')
|
|
def step_attempt_load_class(context: Context, cls: str, mod: str) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.loader.load_class(mod, cls)
|
|
except PluginLoadError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then("the loaded class should not be None")
|
|
def step_loaded_class_not_none(context: Context) -> None:
|
|
assert context.loaded_class is not None
|
|
|
|
|
|
@then('the loaded class name should be "{name}"')
|
|
def step_loaded_class_name(context: Context, name: str) -> None:
|
|
assert context.loaded_class.__name__ == name
|
|
|
|
|
|
@then("a PluginLoadError should be raised")
|
|
def step_plugin_load_error_raised(context: Context) -> None:
|
|
assert isinstance(context.caught_exception, PluginLoadError), (
|
|
f"Expected PluginLoadError, got {type(context.caught_exception)}"
|
|
)
|
|
|
|
|
|
@then('the plugin error message should contain "{text}"')
|
|
def step_plugin_error_message_contains(context: Context, text: str) -> None:
|
|
assert text in str(context.caught_exception), (
|
|
f"Expected '{text}' in '{context.caught_exception}'"
|
|
)
|
|
|
|
|
|
@then('the loader allowed_prefixes should contain "{prefix}"')
|
|
def step_loader_prefixes_contain(context: Context, prefix: str) -> None:
|
|
assert prefix in context.loader.allowed_prefixes
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Protocol validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a class that implements TextBackend protocol")
|
|
def step_conforming_class(context: Context) -> None:
|
|
context.test_class = _FakeTextBackend
|
|
|
|
|
|
@given("a class that does not implement any protocol")
|
|
def step_non_conforming_class(context: Context) -> None:
|
|
context.test_class = _NonProtocolClass
|
|
|
|
|
|
@when("I validate it against the TextBackend protocol")
|
|
def step_validate_protocol(context: Context) -> None:
|
|
context.validation_result = PluginLoader.validate_protocol(
|
|
context.test_class,
|
|
TextBackend,
|
|
)
|
|
|
|
|
|
@when("I attempt to validate it against the TextBackend protocol")
|
|
def step_attempt_validate_protocol(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
PluginLoader.validate_protocol(context.test_class, TextBackend)
|
|
except ProtocolMismatchError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then("the validation should return True")
|
|
def step_validation_true(context: Context) -> None:
|
|
assert context.validation_result is True
|
|
|
|
|
|
@then("a ProtocolMismatchError should be raised")
|
|
def step_protocol_mismatch_raised(context: Context) -> None:
|
|
assert isinstance(context.caught_exception, ProtocolMismatchError)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry-point discovery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I discover plugins from entry point group "{group}"')
|
|
def step_discover_entry_points(context: Context, group: str) -> None:
|
|
context.discovered = context.loader.load_from_entry_points(group=group)
|
|
|
|
|
|
@then("the discovered plugin list should be empty")
|
|
def step_discovered_empty(context: Context) -> None:
|
|
assert len(context.discovered) == 0
|
|
|
|
|
|
@given('a mocked entry point group "{group}" with entry "{entry_spec}"')
|
|
def step_mock_entry_point(context: Context, group: str, entry_spec: str) -> None:
|
|
# Parse "name=module:ClassName"
|
|
name, value = entry_spec.split("=", 1)
|
|
mock_ep = MagicMock()
|
|
mock_ep.name = name
|
|
mock_ep.value = value
|
|
module_path, class_name = value.rsplit(":", 1)
|
|
|
|
# Make ep.load() return the actual class
|
|
import importlib
|
|
|
|
mod = importlib.import_module(module_path)
|
|
mock_ep.load.return_value = getattr(mod, class_name)
|
|
|
|
# Store for use in discovery step
|
|
context.mock_group = group
|
|
context.mock_eps = [mock_ep]
|
|
|
|
|
|
@when("I discover plugins from the mocked entry point group")
|
|
def step_discover_mocked_eps(context: Context) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.select.return_value = context.mock_eps
|
|
with patch("importlib.metadata.entry_points", return_value=mock_result):
|
|
context.discovered = context.loader.load_from_entry_points(
|
|
group=context.mock_group,
|
|
)
|
|
|
|
|
|
@then("the discovered plugin list should have {count:d} entry")
|
|
def step_discovered_count(context: Context, count: int) -> None:
|
|
assert len(context.discovered) == count, (
|
|
f"Expected {count}, got {len(context.discovered)}"
|
|
)
|
|
|
|
|
|
@then('the first descriptor name should be "{name}"')
|
|
def step_first_descriptor_name(context: Context, name: str) -> None:
|
|
assert context.discovered[0].name == name
|
|
|
|
|
|
@given('a mocked entry point group "{group}" with raw entry "{entry_spec}"')
|
|
def step_mock_raw_entry(context: Context, group: str, entry_spec: str) -> None:
|
|
name, value = entry_spec.split("=", 1)
|
|
mock_ep = MagicMock()
|
|
mock_ep.name = name
|
|
mock_ep.value = value
|
|
mock_ep.load = MagicMock(name="load")
|
|
|
|
context.mock_group = group
|
|
context.mock_eps = [mock_ep]
|
|
context.mock_ep_last = mock_ep
|
|
context.loader._logger = MagicMock()
|
|
|
|
|
|
@then("the mocked entry point load should not be called")
|
|
def step_entry_point_not_loaded(context: Context) -> None:
|
|
context.mock_ep_last.load.assert_not_called()
|
|
|
|
|
|
@then('a security warning should be emitted for disallowed entry point "{name}"')
|
|
def step_security_warning_emitted(context: Context, name: str) -> None:
|
|
logger_mock = context.loader._logger
|
|
logger_mock.warning.assert_any_call(
|
|
"plugin.entry_point_disallowed_prefix",
|
|
name=name,
|
|
value=context.mock_ep_last.value,
|
|
group=context.mock_group,
|
|
error=ANY,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PluginManager — lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh PluginManager instance")
|
|
def step_fresh_manager(context: Context) -> None:
|
|
context.manager = PluginManager()
|
|
context.descriptor = None
|
|
context.caught_exception = None
|
|
|
|
|
|
@given('a PluginDescriptor for "{name}" with module "{mod}" and class "{cls}"')
|
|
def step_descriptor_for_manager(
|
|
context: Context, name: str, mod: str, cls: str
|
|
) -> None:
|
|
context.descriptor = PluginDescriptor(
|
|
name=name,
|
|
module_path=mod,
|
|
class_name=cls,
|
|
)
|
|
|
|
|
|
@given('a PluginDescriptor for "{name}" with empty module_path')
|
|
def step_descriptor_empty_path(context: Context, name: str) -> None:
|
|
context.descriptor = PluginDescriptor(
|
|
name=name,
|
|
module_path="",
|
|
class_name="",
|
|
)
|
|
|
|
|
|
@when("I register the plugin descriptor")
|
|
def step_register_descriptor(context: Context) -> None:
|
|
context.manager.register_plugin(context.descriptor)
|
|
|
|
|
|
@when("I attempt to register the same descriptor again")
|
|
def step_attempt_register_again(context: Context) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.register_plugin(context.descriptor)
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then('get_plugin should return the descriptor for "{name}"')
|
|
def step_get_plugin(context: Context, name: str) -> None:
|
|
result = context.manager.get_plugin(name)
|
|
assert result.name == name
|
|
|
|
|
|
@then('the plugin state should be "{state}"')
|
|
def step_plugin_state(context: Context, state: str) -> None:
|
|
desc = context.descriptor
|
|
actual = desc.state.value if hasattr(desc.state, "value") else desc.state
|
|
assert actual == state, f"Expected '{state}', got '{actual}'"
|
|
|
|
|
|
@then("a PluginError should be raised")
|
|
def step_plugin_error_raised(context: Context) -> None:
|
|
assert isinstance(context.caught_exception, PluginError), (
|
|
f"Expected PluginError, got {type(context.caught_exception)}"
|
|
)
|
|
|
|
|
|
@given('I register plugins named "{names_str}"')
|
|
def step_register_multiple(context: Context, names_str: str) -> None:
|
|
for name in [n.strip().strip('"') for n in names_str.split(",")]:
|
|
desc = PluginDescriptor(
|
|
name=name,
|
|
module_path="cleveragents.domain.models.acms.stubs",
|
|
class_name="InMemoryTextBackend",
|
|
)
|
|
context.manager.register_plugin(desc)
|
|
|
|
|
|
@then("list_plugins should return {count:d} descriptors")
|
|
def step_list_plugins_count(context: Context, count: int) -> None:
|
|
plugins = context.manager.list_plugins()
|
|
assert len(plugins) == count, f"Expected {count}, got {len(plugins)}"
|
|
|
|
|
|
@then("list_plugins should return {count:d} descriptor")
|
|
def step_list_plugins_count_singular(context: Context, count: int) -> None:
|
|
plugins = context.manager.list_plugins()
|
|
assert len(plugins) == count, f"Expected {count}, got {len(plugins)}"
|
|
|
|
|
|
@when('I attempt to get plugin "{name}"')
|
|
def step_attempt_get_plugin(context: Context, name: str) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.get_plugin(name)
|
|
except PluginNotFoundError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then("a PluginNotFoundError should be raised")
|
|
def step_not_found_raised(context: Context) -> None:
|
|
assert isinstance(context.caught_exception, PluginNotFoundError)
|
|
|
|
|
|
@when('I activate the plugin "{name}"')
|
|
def step_activate_plugin(context: Context, name: str) -> None:
|
|
context.manager.activate_plugin(name)
|
|
context.descriptor = context.manager.get_plugin(name)
|
|
|
|
|
|
@when('I attempt to activate the plugin "{name}" again')
|
|
def step_attempt_activate_again(context: Context, name: str) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.activate_plugin(name)
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when('I attempt to activate the plugin "{name}"')
|
|
def step_attempt_activate(context: Context, name: str) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.activate_plugin(name)
|
|
except (PluginLoadError, PluginError) as exc:
|
|
context.caught_exception = exc
|
|
context.descriptor = context.manager.get_plugin(name)
|
|
|
|
|
|
@when('I attempt to activate the plugin "{name}" ignoring error')
|
|
def step_attempt_activate_ignore(context: Context, name: str) -> None:
|
|
with contextlib.suppress(PluginLoadError, PluginError):
|
|
context.manager.activate_plugin(name)
|
|
context.descriptor = context.manager.get_plugin(name)
|
|
|
|
|
|
@then('the plugin "{name}" state should be "{state}"')
|
|
def step_named_plugin_state(context: Context, name: str, state: str) -> None:
|
|
desc = context.manager.get_plugin(name)
|
|
actual = desc.state.value if hasattr(desc.state, "value") else desc.state
|
|
assert actual == state, f"Expected '{state}', got '{actual}'"
|
|
|
|
|
|
@then('get_plugin_class for "{name}" should not be None')
|
|
def step_plugin_class_not_none(context: Context, name: str) -> None:
|
|
assert context.manager.get_plugin_class(name) is not None
|
|
|
|
|
|
@then('get_plugin_instance for "{name}" should not be None')
|
|
def step_plugin_instance_not_none(context: Context, name: str) -> None:
|
|
assert context.manager.get_plugin_instance(name) is not None
|
|
|
|
|
|
@then('get_plugin_class for "{name}" should be None')
|
|
def step_plugin_class_none(context: Context, name: str) -> None:
|
|
assert context.manager.get_plugin_class(name) is None
|
|
|
|
|
|
@then('get_plugin_instance for "{name}" should be None')
|
|
def step_plugin_instance_none(context: Context, name: str) -> None:
|
|
assert context.manager.get_plugin_instance(name) is None
|
|
|
|
|
|
@when('I deactivate the plugin "{name}"')
|
|
def step_deactivate_plugin(context: Context, name: str) -> None:
|
|
context.manager.deactivate_plugin(name)
|
|
context.descriptor = context.manager.get_plugin(name)
|
|
|
|
|
|
@when('I attempt to deactivate the plugin "{name}"')
|
|
def step_attempt_deactivate(context: Context, name: str) -> None:
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.deactivate_plugin(name)
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config-driven registration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I register a plugin from config with custom_module "{mod}" and custom_class "{cls}"'
|
|
)
|
|
def step_register_from_config(context: Context, mod: str, cls: str) -> None:
|
|
config: dict[str, Any] = {
|
|
"custom_module": mod,
|
|
"custom_class": cls,
|
|
}
|
|
context.config_result = context.manager.register_from_config(config)
|
|
|
|
|
|
@when("I register a plugin from config with no custom_module or custom_class")
|
|
def step_register_from_config_empty(context: Context) -> None:
|
|
config: dict[str, Any] = {"unrelated": "value"}
|
|
context.config_result = context.manager.register_from_config(config)
|
|
|
|
|
|
@when("I register the same plugin from config again")
|
|
def step_register_from_config_again(context: Context) -> None:
|
|
config: dict[str, Any] = {
|
|
"custom_module": "cleveragents.domain.models.acms.stubs",
|
|
"custom_class": "InMemoryTextBackend",
|
|
}
|
|
context.config_result = context.manager.register_from_config(config)
|
|
|
|
|
|
@when("I register plugins from a config list of {count:d} entries")
|
|
def step_register_batch_config(context: Context, count: int) -> None:
|
|
configs: list[dict[str, Any]] = [
|
|
{
|
|
"custom_module": "cleveragents.domain.models.acms.stubs",
|
|
"custom_class": "InMemoryTextBackend",
|
|
"name": f"batch-plugin-{i}",
|
|
}
|
|
for i in range(count)
|
|
]
|
|
context.batch_result = context.manager.register_all_from_config(configs)
|
|
|
|
|
|
@then("the registered plugin descriptor should not be None")
|
|
def step_config_result_not_none(context: Context) -> None:
|
|
assert context.config_result is not None
|
|
|
|
|
|
@then("the registered plugin descriptor should be None")
|
|
def step_config_result_none(context: Context) -> None:
|
|
assert context.config_result is None
|
|
|
|
|
|
@then("the plugin should be in the manager registry")
|
|
def step_plugin_in_registry(context: Context) -> None:
|
|
assert len(context.manager.list_plugins()) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Extension point management
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I register an extension point named "{name}"')
|
|
def step_register_ext_point(context: Context, name: str) -> None:
|
|
ep = ExtensionPoint(
|
|
name=name,
|
|
protocol_type=TextBackend,
|
|
description="Test EP",
|
|
)
|
|
context.manager.register_extension_point(ep)
|
|
|
|
|
|
@then("list_extension_points should return {count:d} entry")
|
|
def step_list_ext_points_count(context: Context, count: int) -> None:
|
|
eps = context.manager.list_extension_points()
|
|
assert len(eps) == count, f"Expected {count}, got {len(eps)}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Clear
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I clear the plugin manager")
|
|
def step_clear_manager(context: Context) -> None:
|
|
context.manager.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thread safety
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("{n:d} threads register plugins concurrently")
|
|
def step_concurrent_register(context: Context, n: int) -> None:
|
|
errors: list[Exception] = []
|
|
|
|
def _register(idx: int) -> None:
|
|
try:
|
|
desc = PluginDescriptor(
|
|
name=f"thread-plugin-{idx}",
|
|
module_path="cleveragents.domain.models.acms.stubs",
|
|
class_name="InMemoryTextBackend",
|
|
)
|
|
context.manager.register_plugin(desc)
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=_register, args=(i,)) for i in range(n)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert not errors, f"Errors during concurrent registration: {errors}"
|
|
|
|
|
|
@given("a fresh PluginManager with {n:d} registered plugins")
|
|
def step_manager_with_plugins(context: Context, n: int) -> None:
|
|
context.manager = PluginManager()
|
|
for i in range(n):
|
|
desc = PluginDescriptor(
|
|
name=f"conc-plugin-{i}",
|
|
module_path="cleveragents.domain.models.acms.stubs",
|
|
class_name="InMemoryTextBackend",
|
|
)
|
|
context.manager.register_plugin(desc)
|
|
|
|
|
|
@when("{n:d} threads activate plugins concurrently")
|
|
def step_concurrent_activate(context: Context, n: int) -> None:
|
|
errors: list[Exception] = []
|
|
|
|
def _activate(idx: int) -> None:
|
|
try:
|
|
context.manager.activate_plugin(f"conc-plugin-{idx}")
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
threads = [threading.Thread(target=_activate, args=(i,)) for i in range(n)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert not errors, f"Errors during concurrent activation: {errors}"
|
|
|
|
|
|
@then('all {n:d} plugins should be in "{state}" state')
|
|
def step_all_plugins_state(context: Context, n: int, state: str) -> None:
|
|
for i in range(n):
|
|
desc = context.manager.get_plugin(f"conc-plugin-{i}")
|
|
actual = desc.state.value
|
|
assert actual == state, (
|
|
f"Plugin conc-plugin-{i}: expected '{state}', got '{actual}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DI container integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I resolve PluginManager from the DI container")
|
|
def step_resolve_di(context: Context) -> None:
|
|
from cleveragents.application.container import get_container, reset_container
|
|
|
|
reset_container()
|
|
container = get_container()
|
|
context.di_manager = container.plugin_manager()
|
|
|
|
|
|
@then("the resolved PluginManager should not be None")
|
|
def step_di_manager_not_none(context: Context) -> None:
|
|
assert context.di_manager is not None
|
|
assert isinstance(context.di_manager, PluginManager)
|
|
|
|
|
|
@then("resolving PluginManager again should return the same instance")
|
|
def step_di_singleton(context: Context) -> None:
|
|
from cleveragents.application.container import get_container
|
|
|
|
container = get_container()
|
|
second = container.plugin_manager()
|
|
assert context.di_manager is second, "PluginManager should be a singleton"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Discovery integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I call discover with group "{group}"')
|
|
def step_call_discover(context: Context, group: str) -> None:
|
|
context.discovered = context.manager.discover(group=group)
|
|
|
|
|
|
@then("the discovered list should be empty")
|
|
def step_discovered_list_empty(context: Context) -> None:
|
|
assert len(context.discovered) == 0
|
|
|
|
|
|
@given("a mocked entry point group for discover")
|
|
def step_mock_ep_for_discover(context: Context) -> None:
|
|
mock_ep = MagicMock()
|
|
mock_ep.name = "discover-test-plugin"
|
|
mock_ep.value = "cleveragents.domain.models.acms.stubs:InMemoryTextBackend"
|
|
|
|
import importlib
|
|
|
|
mod = importlib.import_module("cleveragents.domain.models.acms.stubs")
|
|
mock_ep.load.return_value = mod.InMemoryTextBackend
|
|
context.mock_discover_eps = [mock_ep]
|
|
|
|
|
|
@when("I call discover through the manager")
|
|
def step_call_manager_discover(context: Context) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.select.return_value = context.mock_discover_eps
|
|
with patch("importlib.metadata.entry_points", return_value=mock_result):
|
|
context.discovered = context.manager.discover(
|
|
group="cleveragents.plugins",
|
|
)
|
|
|
|
|
|
@then("newly discovered plugins should be in the registry")
|
|
def step_discovered_in_registry(context: Context) -> None:
|
|
assert len(context.discovered) > 0
|
|
for desc in context.discovered:
|
|
found = context.manager.get_plugin(desc.name)
|
|
assert found is not None
|