forked from HAL9000/cleveragents-core
CoreRasurae
f2f7aa5dc9
Implement multiple Stage A/B/E/SEC milestones for the v3 lifecycle system: - Stage A5.3+A5.4: Add LifecycleActionModel and LifecyclePlanModel SQLAlchemy models with to_domain()/from_domain() conversion methods - Stage A5.6: Implement ActionRepository with full CRUD, namespace/state queries, referential integrity checks, and retry decorator - Stage E1: Add subplan domain models (ExecutionMode, SubplanMergeStrategy, SubplanConfig, SubplanStatus, SubplanAttempt, SubplanFailureHandler) with computed properties on Plan (is_subplan, is_root_plan, depth, has_subplans) - Stage A6: Add AutomationLevel enum (MANUAL, REVIEW_BEFORE_APPLY, FULL_AUTOMATION), settings integration, PlanLifecycleService auto-progression, pause/resume, and CLI commands (--automation-level, set-automation-level) - Stage SEC1: Remove eval()/exec() from stream_router.py, replace with named operation and transform registries; code blocks and unregistered transforms now raise StreamRoutingError - Add langchain-anthropic dependency - Update BDD tests for security changes and relax ADR directory requirement
342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""
|
|
Step definitions for uncovered stream_router paths.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
import rx
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.reactive.stream_router import (
|
|
ReactiveStreamRouter,
|
|
StreamConfig,
|
|
StreamMessage,
|
|
StreamRoutingError,
|
|
StreamType,
|
|
)
|
|
|
|
|
|
@given("the CleverAgents reactive system is available")
|
|
def step_system_available(context: Context) -> None:
|
|
context.stream_router = ReactiveStreamRouter()
|
|
|
|
|
|
@given("I have a base stream message with context metadata")
|
|
def step_base_stream_message(context: Context) -> None:
|
|
context.original_message = StreamMessage(
|
|
content={"value": 1},
|
|
metadata={"context": {"trace": 1}, "other": {"a": 1}},
|
|
source_stream="base",
|
|
timestamp=1.0,
|
|
)
|
|
|
|
|
|
@when("I copy the message without metadata override")
|
|
def step_copy_message(context: Context) -> None:
|
|
context.copied_message = context.original_message.copy_with()
|
|
|
|
|
|
@then("the metadata should be deep-copied except context")
|
|
def step_verify_copy(context: Context) -> None:
|
|
assert context.copied_message.metadata["other"] == {"a": 1}
|
|
assert (
|
|
context.copied_message.metadata["other"]
|
|
is not context.original_message.metadata["other"]
|
|
)
|
|
assert (
|
|
context.copied_message.metadata["context"]
|
|
is context.original_message.metadata["context"]
|
|
)
|
|
|
|
|
|
@when("I build a map operator without required params")
|
|
def step_map_without_params(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router._create_operator({"type": "map", "params": {}})
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about missing map params")
|
|
def step_verify_map_error(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "Map operator requires" in str(context.error)
|
|
|
|
|
|
@when("I build a buffer operator without count")
|
|
def step_buffer_without_count(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router._create_operator({"type": "buffer", "params": {}})
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about buffer count")
|
|
def step_verify_buffer_error(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "Buffer operator requires" in str(context.error)
|
|
|
|
|
|
@when("I build a window operator without count")
|
|
def step_window_without_count(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router._create_operator({"type": "window", "params": {}})
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about window count")
|
|
def step_verify_window_error(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "Window operator requires" in str(context.error)
|
|
|
|
|
|
@when("I evaluate a field condition against a message without that field")
|
|
def step_eval_missing_field(context: Context) -> None:
|
|
context.condition_result = context.stream_router._evaluate_condition(
|
|
{}, {"field": "missing"}
|
|
)
|
|
|
|
|
|
@then("the condition result should be false")
|
|
def step_verify_condition_false(context: Context) -> None:
|
|
assert context.condition_result is False
|
|
|
|
|
|
@when("I apply the append accumulator with list initialization")
|
|
def step_apply_append_accumulator(context: Context) -> None:
|
|
context.acc_result = context.stream_router._apply_accumulator(
|
|
None, "item", {"op": "append", "init": "list"}
|
|
)
|
|
|
|
|
|
@then("the accumulator should return a list with the item")
|
|
def step_verify_append_accumulator(context: Context) -> None:
|
|
assert context.acc_result == ["item"]
|
|
|
|
|
|
@when("I send a message to an unknown stream")
|
|
def step_send_unknown_stream(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router.send_message("unknown_stream", "data")
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about missing stream")
|
|
def step_verify_missing_stream(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "not found" in str(context.error)
|
|
|
|
|
|
@when("I split a stream from an unknown source")
|
|
def step_split_unknown_source(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router.split_stream("unknown_source", [])
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about missing source")
|
|
def step_verify_missing_source(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "not found" in str(context.error)
|
|
|
|
|
|
@when('I create a stream from the string name "string_stream"')
|
|
def step_create_stream_from_string(context: Context) -> None:
|
|
context.created_stream = context.stream_router.create_stream("string_stream")
|
|
|
|
|
|
@then("the stream should exist in the router")
|
|
def step_stream_exists(context: Context) -> None:
|
|
assert "string_stream" in context.stream_router.streams
|
|
assert context.stream_router.stream_configs["string_stream"].name == "string_stream"
|
|
|
|
|
|
@given('I have already created a stream named "dup_stream"')
|
|
def step_existing_stream(context: Context) -> None:
|
|
context.stream_router.create_stream("dup_stream")
|
|
|
|
|
|
@when('I try to create the stream "dup_stream" again')
|
|
def step_create_duplicate_stream(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router.create_stream("dup_stream")
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about duplicate stream")
|
|
def step_verify_duplicate_stream_error(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "already exists" in str(context.error)
|
|
|
|
|
|
@when('I map a message using function param "noop_function"')
|
|
def step_map_with_function_param(context: Context) -> None:
|
|
context.stream_router._builtin_noop_function = lambda x: x
|
|
operator = context.stream_router._create_operator(
|
|
{"type": "map", "params": {"function": "noop_function"}}
|
|
)
|
|
captured: list[Any] = []
|
|
rx.just(5).pipe(operator).subscribe(lambda v: captured.append(v))
|
|
context.map_result = captured[0] if captured else None
|
|
|
|
|
|
@then("the mapped result should equal the original message")
|
|
def step_verify_map_function_result(context: Context) -> None:
|
|
assert context.map_result == 5
|
|
|
|
|
|
@when("I build a transform operator with an invalid function string")
|
|
def step_transform_invalid_fn(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router._create_operator(
|
|
{"type": "transform", "params": {"fn": "lambda x: x +"}}
|
|
)
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about unregistered transform fn")
|
|
def step_verify_transform_invalid_fn(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "Unknown transform" in str(context.error)
|
|
|
|
|
|
@when("I build a filter operator without condition")
|
|
def step_filter_without_condition(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.stream_router._create_operator({"type": "filter", "params": {}})
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.error = exc
|
|
|
|
|
|
@then("I should get a stream routing error about missing filter condition")
|
|
def step_verify_filter_condition_error(context: Context) -> None:
|
|
assert isinstance(context.error, StreamRoutingError)
|
|
assert "Filter operator requires" in str(context.error)
|
|
|
|
|
|
@when("I run a switch operator with default operators against a message")
|
|
def step_switch_default_operators(context: Context) -> None:
|
|
default_ops = [
|
|
{
|
|
"type": "map",
|
|
"params": {"transform": {"type": "replace", "target": "x", "value": 10}},
|
|
}
|
|
]
|
|
switch_op = context.stream_router._create_operator(
|
|
{"type": "switch", "params": {"cases": [], "default_operators": default_ops}}
|
|
)
|
|
message = {"x": 1}
|
|
captured = []
|
|
switch_op(rx.just(message)).subscribe(lambda v: captured.append(v))
|
|
if captured and isinstance(captured[0], dict):
|
|
context.switch_result = captured[0].get("x")
|
|
else:
|
|
context.switch_result = None
|
|
|
|
|
|
@then("the switch operator should output the transformed value")
|
|
def step_verify_switch_default_ops(context: Context) -> None:
|
|
assert context.switch_result == 10
|
|
|
|
|
|
@when("I run a switch operator that targets a default stream subject")
|
|
def step_switch_default_stream(context: Context) -> None:
|
|
context.stream_router.create_stream(
|
|
StreamConfig(name="default_target", type=StreamType.COLD)
|
|
)
|
|
switch_op = context.stream_router._create_operator(
|
|
{"type": "switch", "params": {"cases": [], "default": "default_target"}}
|
|
)
|
|
message = StreamMessage(content="payload")
|
|
emitted = []
|
|
context.stream_router.streams["default_target"].subscribe(
|
|
lambda v: emitted.append(v.content if hasattr(v, "content") else v)
|
|
)
|
|
switch_op(rx.just(message)).subscribe(lambda _v: None) # drive operator
|
|
# manually emit to default stream to exercise branch
|
|
context.stream_router.streams["default_target"].on_next(message)
|
|
context.default_stream_value = emitted[0] if emitted else None
|
|
|
|
|
|
@then("the default stream should emit its initial value")
|
|
def step_verify_switch_default_stream(context: Context) -> None:
|
|
assert context.default_stream_value == "payload"
|
|
|
|
|
|
@when("I apply a replace transform to a dictionary message")
|
|
def step_apply_replace_transform(context: Context) -> None:
|
|
context.replace_result = context.stream_router._apply_transform(
|
|
{"foo": 1}, {"type": "replace", "target": "foo", "value": 2}
|
|
)
|
|
|
|
|
|
@then("the dictionary field should be updated")
|
|
def step_verify_replace_transform(context: Context) -> None:
|
|
assert context.replace_result["foo"] == 2
|
|
|
|
|
|
@when("I apply the sum accumulator to numeric messages")
|
|
def step_apply_sum_accumulator(context: Context) -> None:
|
|
acc = context.stream_router._apply_accumulator(None, 1, {"op": "sum"})
|
|
context.sum_result = context.stream_router._apply_accumulator(acc, 2, {"op": "sum"})
|
|
|
|
|
|
@then("the accumulator should produce the summed value")
|
|
def step_verify_sum_accumulator(context: Context) -> None:
|
|
assert context.sum_result == 3
|
|
|
|
|
|
@given('I have two source streams named "merge_a" and "merge_b"')
|
|
def step_create_merge_sources(context: Context) -> None:
|
|
context.stream_router.create_stream("merge_a")
|
|
context.stream_router.create_stream("merge_b")
|
|
|
|
|
|
@when('I merge the streams into a new target "merge_output"')
|
|
def step_merge_streams(context: Context) -> None:
|
|
context.stream_router.merge_streams(["merge_a", "merge_b"], "merge_output")
|
|
|
|
|
|
@then("the merged target stream should exist")
|
|
def step_verify_merge_target(context: Context) -> None:
|
|
assert "merge_output" in context.stream_router.streams
|
|
|
|
|
|
@given("I have a stream router with disposable streams and subscriptions")
|
|
def step_router_with_disposables(context: Context) -> None:
|
|
from rx.subject import Subject
|
|
|
|
disposable_stream = Subject()
|
|
disposable_stream.dispose = lambda: setattr(context, "disposed_called", True) # type: ignore[attr-defined]
|
|
context.stream_router.streams["disposable"] = disposable_stream
|
|
context.stream_router.observables["disposable"] = disposable_stream
|
|
context.stream_router.stream_configs["disposable"] = StreamConfig(name="disposable")
|
|
context.stream_router.subscriptions.append(
|
|
lambda: setattr(context, "subscription_disposed", True)
|
|
)
|
|
|
|
|
|
@when("I dispose the stream router")
|
|
def step_dispose_router(context: Context) -> None:
|
|
context.stream_router.dispose()
|
|
|
|
|
|
@then("the stream router internals should be cleared")
|
|
def step_verify_dispose(context: Context) -> None:
|
|
assert context.stream_router.streams == {}
|
|
assert context.stream_router.subscriptions == []
|