209 lines
7.5 KiB
Python
209 lines
7.5 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 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.types import 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
|
|
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:
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exception assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("a PluginError should be raised")
|
|
def step_plugin_error_raised(context: Context) -> None:
|
|
"""Assert that a PluginError was raised."""
|
|
assert context.caught_exception is not None, (
|
|
"Expected a PluginError to be raised, but no exception was caught"
|
|
)
|
|
assert isinstance(context.caught_exception, PluginError), (
|
|
f"Expected PluginError, but got {type(context.caught_exception).__name__}"
|
|
)
|
|
|
|
|
|
@then("a PluginNotFoundError should be raised")
|
|
def step_plugin_not_found_error_raised(context: Context) -> None:
|
|
"""Assert that a PluginNotFoundError was raised."""
|
|
assert context.caught_exception is not None, (
|
|
"Expected a PluginNotFoundError to be raised, but no exception was caught"
|
|
)
|
|
assert isinstance(context.caught_exception, PluginNotFoundError), (
|
|
f"Expected PluginNotFoundError, but got {type(context.caught_exception).__name__}"
|
|
)
|
|
|
|
|
|
@then('the plugin error message should contain "{expected_text}"')
|
|
def step_plugin_error_message_contains(context: Context, expected_text: str) -> None:
|
|
"""Assert that the plugin error message contains expected text."""
|
|
assert context.caught_exception is not None, (
|
|
"Expected an exception to be raised, but none was caught"
|
|
)
|
|
error_message = str(context.caught_exception)
|
|
assert expected_text in error_message, (
|
|
f"Expected error message to contain '{expected_text}', "
|
|
f"but got: {error_message}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Deactivation guard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I attempt to deactivate the plugin "{name}"')
|
|
def step_attempt_deactivate_plugin(context: Context, name: str) -> None:
|
|
"""Attempt to deactivate a plugin."""
|
|
context.caught_exception = None
|
|
try:
|
|
context.manager.deactivate_plugin(name)
|
|
except PluginError as exc:
|
|
context.caught_exception = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PluginState enum verification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the PluginState enum is available")
|
|
def step_plugin_state_enum_available(context: Context) -> None:
|
|
"""Verify that the PluginState enum is available."""
|
|
context.plugin_state_enum = PluginState
|
|
|
|
|
|
@then('it should have values "{values}"')
|
|
def step_enum_has_values(context: Context, values: str) -> None:
|
|
"""Assert that the PluginState enum has the expected values."""
|
|
expected_values = set(values.split('", "'))
|
|
# Clean up the first and last values (they have extra quotes)
|
|
expected_values = {v.strip('"') for v in expected_values}
|
|
|
|
actual_values = {state.value for state in context.plugin_state_enum}
|
|
|
|
assert actual_values == expected_values, (
|
|
f"Expected PluginState enum to have values {expected_values}, "
|
|
f"but got {actual_values}"
|
|
)
|