# pyright: reportRedeclaration=false
"""Step definitions for LSP transport header injection security (issue #7112).

Mocked ``StdioTransport`` using a ``MagicMock`` subprocess and
a ``BytesIO``-based stdout so that non-ASCII header bytes can be
injected and the strict-ASCII enforcement guard exercised without
requiring a real language server.
"""

from __future__ import annotations

import select
from io import BytesIO
from typing import Any
from unittest.mock import MagicMock, patch

from behave import given, then, when

from cleveragents.lsp.errors import LspError
from cleveragents.lsp.transport import StdioTransport


def patched_select(
    readable: list[Any],
    *_args: Any,
    **_kwargs: Any,
) -> tuple[list[Any], list[Any], list[Any]]:
    """Patch ``select.select`` to return a ready file descriptor immediately.

    Returns a proper 3-tuple matching the ``select.select()`` API contract:
    ``([readable[0]], [], [])`` rather than just ``[readable[0]]`` which
    would cause ``ValueError`` during tuple unpacking in ``_read_one_message()``.
    """
    if readable:
        return ([readable[0]], [], [])
    return ([], [], [])


# ── Given steps ──────────────────────────────────────────────────────────


@given("a Transport mock with BytesIO stream containing {raw_headers}")
def step_transport_with_streams(context: Any, raw_headers: str) -> None:
    """Set up a transport whose subprocess.stdout is a BytesIO initialised
    with the specified raw header+body bytes."""
    data: bytes = eval(raw_headers)
    transport = object.__new__(StdioTransport)  # Bypass __init__ validation
    transport._process = MagicMock()
    transport._process.poll.return_value = None  # Pretend process is alive
    context.stream = BytesIO(data)
    transport._process.stdout = context.stream
    context.transport = transport
    context.mock_process = transport._process


# ── When steps ───────────────────────────────────────────────────────────


@when("_read_one_message() is invoked")
def step_invoke_read_message(context: Any) -> None:
    """Call ``_read_one_message`` on the mocked transport.

    Because :meth:`select.select` does not work natively on ``BytesIO``, we
    patch it so that a ready file descriptor is returned immediately,
    unblocking ``stdout.readline()`` and driving the header-reading loop with
    the data already sitting in the ``BytesIO`` buffer.
    """

    with patch.object(select, "select", side_effect=patched_select):
        context.raised_error = None
        try:
            context.read_result = context.transport._read_one_message(timeout=1.0)
        except Exception as exc:
            context.raised_error = exc


# ── Then steps ───────────────────────────────────────────────────────────


@then("it should raise an LspError exception")
def step_raises_lsp_error(context: Any) -> None:
    assert context.raised_error is not None, "Expected an error to be raised"
    assert isinstance(context.raised_error, (LspError, UnicodeDecodeError)), (
        f"Expected LspError or UnicodeDecodeError, got {type(context.raised_error).__name__}"
    )


@then("it should return a parsed JSON dict")
def step_returns_parsed_json_dict(context: Any) -> None:
    assert context.raised_error is None
    assert isinstance(context.read_result, dict), (
        f"Expected parsed dict, got {type(context.read_result).__name__}"
    )


@then('the error message must contain "non-ASCII"')
def step_error_contains_non_ascii(context: Any) -> None:
    assert context.raised_error is not None
    error_text = str(context.raised_error).lower()
    assert "non-ascii" in error_text or "unicode" in error_text


@then('the result must contain "jsonrpc" == "2.0"')
def step_result_has_jsonrpc(context: Any) -> None:
    assert isinstance(context.read_result, dict), "Result should be a dict"
    assert context.read_result.get("jsonrpc") == "2.0"


@then('the result must contain "id" == 1')
def step_result_has_id(context: Any) -> None:
    assert isinstance(context.read_result, dict), "Result should be a dict"
    assert context.read_result.get("id") == 1


@then('the result must contain "result" == "success"')
def step_result_has_result(context: Any) -> None:
    assert isinstance(context.read_result, dict), "Result should be a dict"
    assert context.read_result.get("result") == "success"


@then('the result must contain "Content-Length"')
def step_result_has_content_length(context: Any) -> None:
    assert isinstance(context.read_result, dict), "Result should be a dict"
    cl = context.read_result.get("Content-Length")
    assert cl is None or isinstance(cl, int)


@then("the transport should still be alive")
def step_transport_alive(context: Any) -> None:
    assert context.mock_process.poll.return_value is None
