HAL9000
552dcb15b0
CI / lint (pull_request) Successful in 43s
CI / helm (pull_request) Successful in 45s
CI / build (pull_request) Successful in 53s
CI / quality (pull_request) Successful in 56s
CI / push-validation (pull_request) Successful in 55s
CI / typecheck (pull_request) Successful in 1m18s
CI / security (pull_request) Successful in 1m21s
CI / unit_tests (pull_request) Successful in 6m29s
CI / integration_tests (pull_request) Successful in 9m16s
CI / docker (pull_request) Successful in 1m41s
CI / coverage (pull_request) Successful in 12m27s
CI / status-check (pull_request) Successful in 3s
AmbiguousStep error caused all 8 features to error: plugin_executing_state_steps.py
re-defined 6 steps already present in plugin_architecture_steps.py:
- @then("a PluginError should be raised")
- @then("a PluginNotFoundError should be raised")
- @then('the plugin error message should contain "{text}"')
- @when('I attempt to deactivate the plugin "{name}"')
- @given("the PluginState enum is available")
- @then('it should have values ...')
Remove all duplicates, keeping only the 6 new steps unique to this feature.
Also fix the last scenario where `And I attempt to execute method on unknown plugin`
followed a `Given` step (so behave treated it as `given` context, not matching
the `@when` definition). Changed `And` to `When`.
ISSUES CLOSED: #5691
123 lines
4.1 KiB
Python
123 lines
4.1 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 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:
|
|
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:
|
|
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:
|
|
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:
|
|
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:
|
|
descriptor = context.manager.get_plugin(name)
|
|
descriptor.state = PluginState(state)
|