cleveragents-core/features/steps/materializers_coverage_boost_steps.py

"""Step definitions for materializers_coverage_boost.feature.

These steps target uncovered lines in materializers.py, specifically:
- Lines 466-469: on_session_end force-flush of remaining buffered elements
- Line 169: _render_element_plain fallback for unknown element types
- Line 269: _render_element_color fallback for unknown element types
- Line 372: _element_to_dict fallback for unknown element types
- Line 440: _BaseBufferStrategy._render_element default
- Line 445: on_element_closed early return for unknown handle_id
- Line 551: _AccumulateStrategy._serialise default
- Line 576: _AccumulateStrategy._serialise_dict default
"""

from __future__ import annotations

import re

from behave import given, then, when
from behave.runner import Context

from cleveragents.cli.output._color_renderers import (
    render_element_color as _render_element_color,
)
from cleveragents.cli.output._renderers import (
    render_element_plain as _render_element_plain,
)
from cleveragents.cli.output.handles import (
    ElementClosed,
    Panel,
    PanelEntry,
    SessionEnd,
)
from cleveragents.cli.output.materializers import (
    ColorMaterializer,
    PlainMaterializer,
    RichMaterializer,
    TableMaterializer,
    _AccumulateStrategy,
    _BaseBufferStrategy,
    _element_to_dict,
)
from cleveragents.cli.output.session import OutputSession, StructuredOutput

# ===========================================================================
# Scenario 1: Force-flush remaining buffers on session end
# Targets lines 466-469
# ===========================================================================


@given("a PlainMaterializer buffer strategy")
def step_given_plain_materializer(context: Context) -> None:
    context.mat_strategy = PlainMaterializer()


@given("an OutputSession wired to that buffer strategy")
def step_given_session_with_plain_strategy(context: Context) -> None:
    context.mat_session = OutputSession(
        format="plain", command="test-flush", strategy=context.mat_strategy
    )


@when("I create three panels in declaration order")
def step_create_three_panels(context: Context) -> None:
    context.panel_0 = context.mat_session.panel("Panel-Zero")
    context.panel_0.set_entry("key0", "val0")
    context.panel_1 = context.mat_session.panel("Panel-One")
    context.panel_1.set_entry("key1", "val1")
    context.panel_2 = context.mat_session.panel("Panel-Two")
    context.panel_2.set_entry("key2", "val2")


@when("I close only the third and second panels")
def step_close_third_and_second(context: Context) -> None:
    # Close out of declaration order: index 2 first, then index 1
    # This means index 0 is still open, so _flush_ready won't drain
    # anything (it waits for _next_render_index=0 which isn't closed).
    # Items 1 and 2 accumulate in self._buffers.
    context.panel_2.close()
    context.panel_1.close()


@when("I end the session without closing the first panel")
def step_end_session_without_first(context: Context) -> None:
    # session.close() force-closes open handles, but the first panel
    # (index 0) gets closed last by the session. By the time
    # on_session_end fires, the buffers for index 1 and 2 should
    # still be present (since _flush_ready would have flushed 0,
    # then potentially 1 and 2 in sequence).
    # Actually, let's be precise: session.close() calls handle.close()
    # for open handles. Panel 0 close triggers on_element_closed with
    # idx=0, which flushes 0, then 1, then 2 via _flush_ready.
    # To truly leave items in buffers at session_end time, we need to
    # avoid the session force-closing the handle. Instead, we directly
    # invoke on_session_end while items are still buffered.
    #
    # Strategy: Manually invoke on_session_end before session.close()
    # to exercise the force-flush path while buffers are non-empty.
    event = SessionEnd(
        event_type="session_end",
        handle_id="",
        element_kind="session",
        exit_code=0,
    )
    # At this point: panel_0 is open (not closed), so _flush_ready hasn't
    # flushed anything. Panels 1 and 2 are closed but stuck in _buffers
    # because _next_render_index is 0 and index 0 isn't in _closed_indices.
    # But wait - when panel_1 and panel_2 closed, on_element_closed stored
    # them in _buffers. _flush_ready starts at _next_render_index=0, and
    # since 0 is NOT in _closed_indices, it doesn't flush anything.
    # So _buffers has {1: rendered_1, 2: rendered_2}.
    # Calling on_session_end now will iterate sorted(self._buffers.keys())
    # = [1, 2] and write their content → covers lines 466-469.
    context.mat_strategy.on_session_end(event)


@then("the strategy output includes content from all three panels")
def step_verify_all_panels_in_output(context: Context) -> None:
    output = context.mat_strategy.get_output()
    assert "Panel-One" in output, f"Panel-One not found in output:\n{output}"
    assert "Panel-Two" in output, f"Panel-Two not found in output:\n{output}"
    # Panel-Zero might not be in the output since it was never closed
    # (no rendered content for it). That's expected - we're testing that
    # the force-flush path writes the buffered panels 1 and 2.


# ===========================================================================
# Scenario 2: Session end with non-empty buffers (color strategy)
# Also targets lines 466-469
# ===========================================================================


@given("a ColorMaterializer buffer strategy")
def step_given_color_materializer(context: Context) -> None:
    context.color_mat_strategy = ColorMaterializer()


@given("an OutputSession wired to that color buffer strategy")
def step_given_session_with_color_strategy(context: Context) -> None:
    context.color_mat_session = OutputSession(
        format="color", command="test-color-flush", strategy=context.color_mat_strategy
    )


@when("I create two status elements in order")
def step_create_two_statuses(context: Context) -> None:
    context.status_0 = context.color_mat_session.status("First status", level="info")
    context.status_1 = context.color_mat_session.status("Second status", level="ok")


@when("I close only the second status element")
def step_close_second_status(context: Context) -> None:
    context.status_1.close()


@when("the session is closed via context exit")
def step_session_closed_via_exit(context: Context) -> None:
    # Manually fire session_end while buffer has items.
    # Status 0 is open, so _next_render_index=0 hasn't advanced.
    # Status 1 is closed but in _buffers at index 1.
    event = SessionEnd(
        event_type="session_end",
        handle_id="",
        element_kind="session",
        exit_code=0,
    )
    context.color_mat_strategy.on_session_end(event)


@then("the color strategy output includes both status messages")
def step_color_output_has_both(context: Context) -> None:
    output = context.color_mat_strategy.get_output()
    # Second status was buffered and force-flushed
    assert "Second status" in output, f"Second status not in output:\n{output}"


# ===========================================================================
# Scenario 3: _render_element_plain fallback (line 169)
# ===========================================================================


@given("an unknown element type object")
def step_given_unknown_element(context: Context) -> None:
    # Create an object that is not Panel, Table, StatusMessage, or ProgressIndicator
    context.unknown_element = {"custom": "data", "id": 42}


@when("I render it with the plain element dispatcher")
def step_render_plain_unknown(context: Context) -> None:
    context.plain_render_result = _render_element_plain(context.unknown_element)


@then("the result is the string representation of the object")
def step_verify_plain_str_fallback(context: Context) -> None:
    expected = str(context.unknown_element)
    assert context.plain_render_result == expected, (
        f"Expected {expected!r}, got {context.plain_render_result!r}"
    )


# ===========================================================================
# Scenario 4: _render_element_color fallback (line 269)
# ===========================================================================


@when("I render it with the color element dispatcher")
def step_render_color_unknown(context: Context) -> None:
    context.color_render_result = _render_element_color(context.unknown_element)


@then("the color result is the string representation of the object")
def step_verify_color_str_fallback(context: Context) -> None:
    expected = str(context.unknown_element)
    assert context.color_render_result == expected, (
        f"Expected {expected!r}, got {context.color_render_result!r}"
    )


# ===========================================================================
# Scenario 5: _element_to_dict fallback (line 372)
# ===========================================================================


@when("I convert it to a dict using element_to_dict")
def step_convert_to_dict_unknown(context: Context) -> None:
    context.element_dict_result = _element_to_dict(context.unknown_element)


@then("the result dict has type equal to unknown")
def step_verify_unknown_type_dict(context: Context) -> None:
    assert context.element_dict_result == {"type": "unknown"}, (
        f"Expected {{'type': 'unknown'}}, got {context.element_dict_result!r}"
    )


# ===========================================================================
# Scenario 6: _BaseBufferStrategy._render_element default (line 440)
# ===========================================================================


@given("a raw BaseBufferStrategy instance")
def step_given_base_buffer_strategy(context: Context) -> None:
    context.base_strategy = _BaseBufferStrategy()


@given("a simple Panel element for rendering")
def step_given_simple_panel(context: Context) -> None:
    context.simple_panel = Panel(
        title="Base Panel",
        entries=[PanelEntry(key="host", value="localhost")],
    )


@when("I call the base strategy render_element with the panel")
def step_call_base_render(context: Context) -> None:
    context.base_render_result = context.base_strategy._render_element(
        context.simple_panel
    )


@then("the rendered output contains the panel title")
def step_verify_base_render(context: Context) -> None:
    assert "Base Panel" in context.base_render_result, (
        f"Expected 'Base Panel' in {context.base_render_result!r}"
    )


# ===========================================================================
# Scenario 7: _AccumulateStrategy._serialise default (line 551)
# ===========================================================================


@given("a raw AccumulateStrategy instance")
def step_given_accumulate_strategy(context: Context) -> None:
    context.accum_strategy = _AccumulateStrategy()


@given("a mock StructuredOutput snapshot")
def step_given_mock_snapshot(context: Context) -> None:
    context.mock_snapshot = StructuredOutput(
        command="test",
        session_id="ses-000001",
        elements=[],
        exit_code=0,
        metadata={},
    )


@when("I call the base strategy serialise method")
def step_call_base_serialise(context: Context) -> None:
    context.serialise_result = context.accum_strategy._serialise(context.mock_snapshot)


@then("the result is an empty string")
def step_verify_empty_serialise(context: Context) -> None:
    assert context.serialise_result == "", (
        f"Expected empty string, got {context.serialise_result!r}"
    )


# ===========================================================================
# Scenario 8: _AccumulateStrategy._serialise_dict default (line 576)
# ===========================================================================


@when("I call the base strategy serialise_dict with a sample dict")
def step_call_base_serialise_dict(context: Context) -> None:
    context.serialise_dict_result = context.accum_strategy._serialise_dict(
        {"key": "value"}
    )


@then("the serialise_dict result is an empty string")
def step_verify_empty_serialise_dict(context: Context) -> None:
    assert context.serialise_dict_result == "", (
        f"Expected empty string, got {context.serialise_dict_result!r}"
    )


# ===========================================================================
# Scenario 9: on_element_closed with unknown handle_id (line 445)
# ===========================================================================


@when("I dispatch an ElementClosed event with an unknown handle id")
def step_dispatch_unknown_handle_closed(context: Context) -> None:
    event = ElementClosed(
        event_type="closed",
        handle_id="hdl-nonexistent-999",
        element_kind="panel",
        final_state=Panel(title="Ghost"),
    )
    context.mat_strategy.on_element_closed(event)


@then("the strategy output is empty")
def step_verify_empty_output(context: Context) -> None:
    output = context.mat_strategy.get_output()
    assert output == "", f"Expected empty output, got {output!r}"


# ===========================================================================
# Scenario 10: TableMaterializer renders non-table element via plain
# Targets line 327 (_render_element_table -> _render_element_plain)
# ===========================================================================


@given("a TableMaterializer buffer strategy")
def step_given_table_materializer(context: Context) -> None:
    context.table_mat_strategy = TableMaterializer()


@given("an OutputSession wired to that table buffer strategy")
def step_given_session_with_table_strategy(context: Context) -> None:
    context.table_mat_session = OutputSession(
        format="table",
        command="test-table-fallback",
        strategy=context.table_mat_strategy,
    )


@when('I create and close a status element with message "{msg}" and level "{level}"')
def step_create_close_status(context: Context, msg: str, level: str) -> None:
    handle = context.table_mat_session.status(msg, level=level)
    handle.close()


@then('the table strategy output contains "{text}"')
def step_table_strategy_contains(context: Context, text: str) -> None:
    output = context.table_mat_strategy.get_output()
    assert text in output, f"Expected {text!r} in:\n{output}"


# ===========================================================================
# Scenario 11: Session end with all elements already flushed
# ===========================================================================


@when('I create and close a panel with title "{title}" and entry "{key}" "{val}"')
def step_create_close_panel(context: Context, title: str, key: str, val: str) -> None:
    handle = context.mat_session.panel(title)
    handle.set_entry(key, val)
    handle.close()


@when("the session is ended cleanly")
def step_end_session_cleanly(context: Context) -> None:
    context.mat_session.close()


@then('the strategy output contains "{text}" exactly once')
def step_output_contains_once(context: Context, text: str) -> None:
    output = context.mat_strategy.get_output()
    count = output.count(text)
    assert count == 1, (
        f"Expected '{text}' exactly once, found {count} times in:\n{output}"
    )


# ===========================================================================
# Scenario 12: RichMaterializer renders with ANSI codes
# ===========================================================================

ANSI_RE = re.compile(r"\033\[")


@given("a RichMaterializer buffer strategy")
def step_given_rich_materializer(context: Context) -> None:
    context.rich_mat_strategy = RichMaterializer()


@given("an OutputSession wired to that rich buffer strategy")
def step_given_session_with_rich_strategy(context: Context) -> None:
    context.rich_mat_session = OutputSession(
        format="rich",
        command="test-rich",
        strategy=context.rich_mat_strategy,
    )


@when('I create and close a rich panel with title "{title}" and entry "{key}" "{val}"')
def step_create_close_rich_panel(
    context: Context, title: str, key: str, val: str
) -> None:
    handle = context.rich_mat_session.panel(title)
    handle.set_entry(key, val)
    handle.close()


@then("the rich strategy output contains ANSI escape codes")
def step_rich_has_ansi(context: Context) -> None:
    output = context.rich_mat_strategy.get_output()
    assert ANSI_RE.search(output), f"No ANSI codes found in output:\n{output}"


@then('the rich strategy output contains "{text}"')
def step_rich_output_contains(context: Context, text: str) -> None:
    output = context.rich_mat_strategy.get_output()
    # Strip ANSI codes for text comparison
    clean = re.sub(r"\033\[[0-9;]*m", "", output)
    assert text in output or text in clean, f"Expected {text!r} in:\n{output}"