HAL9000
b0edcb53a4
- Added execute_plugin() method to PluginManager that properly sets PluginState.EXECUTING before calling a plugin method - Transitions state back to ACTIVATED on success, or ERRORED on failure - Completes the lifecycle state machine defined in PluginState enum - Added BDD feature file features/plugin_executing_state.feature with scenarios covering state transitions - Added step definitions in features/steps/plugin_executing_state_steps.py ISSUES CLOSED: #5691
137 lines
4.7 KiB
Python
137 lines
4.7 KiB
Python
"""Behave step definitions for PluginState.EXECUTING lifecycle state.
|
|
|
|
Covers the execute_plugin() method on PluginManager, which transitions
|
|
plugins through ACTIVATED -> EXECUTING -> ACTIVATED (success) or
|
|
ACTIVATED -> EXECUTING -> ERRORED (failure).
|
|
|
|
Based on issue #5691.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
|
|
from cleveragents.infrastructure.plugins.exceptions import (
|
|
PluginError,
|
|
PluginNotFoundError,
|
|
)
|
|
from cleveragents.infrastructure.plugins.manager import PluginManager
|
|
from cleveragents.infrastructure.plugins.types import (
|
|
PluginDescriptor,
|
|
PluginState,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Execute plugin — happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I execute method "search" on plugin "{name}" with valid args')
|
|
def step_execute_plugin_search(context: Context, name: str) -> None:
|
|
"""Execute the search method on an InMemoryTextBackend plugin."""
|
|
context.caught_exception = None
|
|
context.execution_result = None
|
|
try:
|
|
context.execution_result = context.manager.execute_plugin(
|
|
name,
|
|
"search",
|
|
"test query",
|
|
scope=frozenset(),
|
|
max_results=5,
|
|
)
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@then("the execution result should be an empty list")
|
|
def step_execution_result_empty_list(context: Context) -> None:
|
|
assert context.execution_result == [], (
|
|
f"Expected empty list, got {context.execution_result!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Execute plugin — error path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I attempt to execute a failing method on plugin "{name}"')
|
|
def step_attempt_execute_failing(context: Context, name: str) -> None:
|
|
"""Execute a method that raises an exception on the plugin."""
|
|
context.caught_exception = None
|
|
# Patch the instance's search method to raise a RuntimeError
|
|
instance = context.manager.get_plugin_instance(name)
|
|
if instance is not None:
|
|
with patch.object(instance, "search", side_effect=RuntimeError("boom")):
|
|
try:
|
|
context.manager.execute_plugin(
|
|
name,
|
|
"search",
|
|
"test query",
|
|
scope=frozenset(),
|
|
max_results=5,
|
|
)
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
else:
|
|
# Plugin not activated — try to execute anyway to get the error
|
|
try:
|
|
context.manager.execute_plugin(name, "search", "test", scope=frozenset())
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Execute plugin — guard: not activated
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I attempt to execute method on non-activated plugin "{name}"')
|
|
def step_attempt_execute_non_activated(context: Context, name: str) -> None:
|
|
"""Attempt to execute a method on a plugin that is not in ACTIVATED state."""
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.execute_plugin(
|
|
name,
|
|
"search",
|
|
"test query",
|
|
scope=frozenset(),
|
|
max_results=5,
|
|
)
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
@when('I attempt to execute method on unknown plugin "{name}"')
|
|
def step_attempt_execute_unknown(context: Context, name: str) -> None:
|
|
"""Attempt to execute a method on a plugin that is not registered."""
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.execute_plugin(
|
|
name,
|
|
"search",
|
|
"test query",
|
|
scope=frozenset(),
|
|
max_results=5,
|
|
)
|
|
except (PluginError, PluginNotFoundError) as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thread safety: EXECUTING state prevents deactivation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I manually set plugin "{name}" state to "{state}"')
|
|
def step_manually_set_state(context: Context, name: str, state: str) -> None:
|
|
"""Manually set a plugin's state for testing purposes."""
|
|
descriptor = context.manager.get_plugin(name)
|
|
descriptor.state = PluginState(state)
|