temp/features/steps/stream_router_new_branches_steps.py

"""
Step definitions for additional stream_router uncovered branches.
"""

from __future__ import annotations

from typing import Any

import rx
from behave import given, then, when
from rx import operators as ops

from cleveragents.reactive.stream_router import (
    StreamConfig,
    StreamMessage,
    StreamRoutingError,
    StreamType,
)


@given("a LangGraph bridge stub is registered on the stream router")
def step_langgraph_bridge_stub(context) -> None:
    class BridgeStub:
        def __init__(self) -> None:
            self.called = False

        def _operator_graph_execute(self, params):
            self.called = True
            suffix = params.get("value", "mapped")
            return ops.map(lambda msg: f"{getattr(msg, 'content', msg)}-{suffix}")

    context.bridge = BridgeStub()
    context.stream_router._langgraph_bridge = context.bridge


@when("I build a graph_execute operator through the bridge")
def step_build_graph_execute(context) -> None:
    operator = context.stream_router._create_operator(
        {"type": "graph_execute", "params": {"value": "ok"}}
    )
    captured: list[Any] = []
    rx.just(StreamMessage(content="ping")).pipe(operator).subscribe(captured.append)
    context.graph_result = captured[0] if captured else None


@then("the bridge factory should be invoked and map data")
def step_verify_bridge_called(context) -> None:
    assert context.bridge.called is True
    assert getattr(context.graph_result, "content", context.graph_result).endswith(
        "-ok"
    )


@given("a dummy agent is registered on the stream router")
def step_register_dummy_agent(context) -> None:
    class DummyAgent:
        def process(self, content, metadata):
            return f"processed-{content}"

    context.dummy_agent = DummyAgent()
    context.stream_router.register_agent("dummy", context.dummy_agent)


@when("I map a message using that agent")
def step_map_with_agent(context) -> None:
    operator = context.stream_router._create_operator(
        {"type": "map", "params": {"agent": "dummy"}}
    )
    captured: list[StreamMessage] = []
    rx.just(StreamMessage(content="payload", metadata={"foo": "bar"})).pipe(
        operator
    ).subscribe(captured.append)
    context.agent_result = captured[0] if captured else None


@then("the agent-mapped content should be returned with metadata")
def step_verify_agent_map(context) -> None:
    assert isinstance(context.agent_result, StreamMessage)
    assert context.agent_result.content == "processed-payload"
    assert (
        context.agent_result.metadata.get("processed_by")
        == context.dummy_agent.__class__.__name__
    )
    assert context.agent_result.metadata.get("foo") == "bar"


@when("I build a filter operator with equals condition")
def step_build_filter_equals(context) -> None:
    operator = context.stream_router._create_operator(
        {"type": "filter", "params": {"condition": {"equals": 2}}}
    )
    captured: list[int] = []
    rx.from_([1, 2, 3]).pipe(operator).subscribe(captured.append)
    context.filter_result = captured


@then("the filter should emit only matching values")
def step_verify_filter(context) -> None:
    assert context.filter_result == [2]


@when("I build a transform operator with an unregistered function string")
def step_build_transform_fn_rejected(context) -> None:
    context.error = None
    try:
        context.stream_router._create_operator(
            {"type": "transform", "params": {"fn": "lambda x: x * 3"}}
        )
    except Exception as exc:  # pylint: disable=broad-except
        context.error = exc


@then("I should get a stream routing error about unregistered transform")
def step_verify_transform_rejected(context) -> None:
    from cleveragents.reactive.stream_router import StreamRoutingError

    assert isinstance(context.error, StreamRoutingError)
    assert "Unknown transform" in str(context.error)
    assert "register_transform" in str(context.error).lower()


@when("I run a switch operator with a case operator pipeline")
def step_switch_case_operators(context) -> None:
    case_ops = [
        {
            "type": "map",
            "params": {"transform": {"type": "replace", "target": "a", "value": 2}},
        }
    ]
    switch_op = context.stream_router._create_operator(
        {
            "type": "switch",
            "params": {
                "cases": [
                    {"condition": {"equals": {"a": 1}}, "operators": case_ops},
                ]
            },
        }
    )
    message = {"a": 1}
    captured: list[Any] = []
    switch_op(rx.just(message)).subscribe(captured.append)
    context.switch_case_result = captured[0] if captured else None


@then("the switch case should apply its operators")
def step_verify_switch_case(context) -> None:
    assert isinstance(context.switch_case_result, dict)
    assert context.switch_case_result.get("a") == 2


@given('I have an existing target stream named "case_target"')
def step_setup_case_target(context) -> None:
    context.stream_router.create_stream(
        StreamConfig(name="case_target", type=StreamType.COLD)
    )


@when("I run a switch operator that routes to that case target")
def step_switch_routes_case_target(context) -> None:
    switch_op = context.stream_router._create_operator(
        {
            "type": "switch",
            "params": {
                "cases": [
                    {"condition": {"equals": "route"}, "target": "case_target"},
                ]
            },
        }
    )
    message = StreamMessage(content="route")
    emitted: list[Any] = []
    context.stream_router.streams["case_target"].subscribe(
        lambda v: emitted.append(getattr(v, "content", v))
    )
    switch_op(rx.just(message)).subscribe(lambda _v: None)
    context.stream_router.streams["case_target"].on_next(message)
    context.case_target_emitted = emitted[0] if emitted else None


@then("the case target stream should receive the message")
def step_verify_case_target(context) -> None:
    assert context.case_target_emitted == "route"


@when("I build an operator with an unknown type")
def step_build_unknown_operator(context) -> None:
    context.error = None
    try:
        context.stream_router._create_operator({"type": "unknown", "params": {}})
    except Exception as exc:  # pylint: disable=broad-except
        context.error = exc


@then("I should get a stream routing error about unknown operator")
def step_verify_unknown_operator_error(context) -> None:
    assert isinstance(context.error, StreamRoutingError)
    assert "Unknown operator type" in str(context.error)


@when("I apply an extract_field transform to different message types")
def step_apply_extract_transform(context) -> None:
    context.extract_value = context.stream_router._apply_transform(
        {"foo": "bar"}, {"type": "extract_field", "field": "foo"}
    )
    context.extract_fallback = context.stream_router._apply_transform(
        "raw", {"type": "extract_field", "field": "foo"}
    )


@then("the transform should return the field value or original message")
def step_verify_extract_transform(context) -> None:
    assert context.extract_value == "bar"
    assert context.extract_fallback == "raw"


@when("I evaluate equals and field conditions")
def step_evaluate_conditions(context) -> None:
    context.equals_result = context.stream_router._evaluate_condition(
        StreamMessage(content="yes"), {"equals": "yes"}
    )
    context.field_result = context.stream_router._evaluate_condition(
        {"present": 1}, {"field": "present"}
    )


@then("both condition checks should return true")
def step_verify_conditions(context) -> None:
    assert context.equals_result is True
    assert context.field_result is True


@when("I apply the append accumulator to a non-list accumulator")
def step_append_non_list(context) -> None:
    context.append_nonlist_result = context.stream_router._apply_accumulator(
        0, "item", {"op": "append"}
    )


@then("the accumulator should be returned unchanged")
def step_verify_append_non_list(context) -> None:
    assert context.append_nonlist_result == 0


@given("I prepare stream config with existing and missing subscriptions")
def step_prepare_subscriptions(context) -> None:
    from rx.subject import Subject

    context.stream_router.streams["existing"] = Subject()
    context.stream_router.streams["subscriber"] = Subject()
    context.stream_router.stream_configs["existing"] = StreamConfig(name="existing")
    context.stream_router.stream_configs["subscriber"] = StreamConfig(name="subscriber")
    context.subscription_config = StreamConfig(
        name="subscriber", subscriptions=["existing", "missing"]
    )


@when("I setup subscriptions for that config")
def step_setup_subscriptions(context) -> None:
    context.stream_router._setup_subscriptions(context.subscription_config)


@then("only existing sources should be subscribed")
def step_verify_subscriptions(context) -> None:
    assert len(context.stream_router.subscriptions) == 1