"""Regression tests for Client.Python low-severity code-review findings.

Covers Client.Python-006 (concurrent-close idempotency),
Client.Python-007 (shared cancelling stream helper),
Client.Python-008 (non-finite float / bytes value mapping), and
Client.Python-011 (`success == 0` proto3-default ambiguity).
"""

from __future__ import annotations

import asyncio
import math
from typing import Any

import pytest

from mxgateway import ClientOptions, GalaxyRepositoryClient, GatewayClient
from mxgateway.errors import ensure_mxaccess_success, MxAccessError
from mxgateway.generated import mxaccess_gateway_pb2 as pb
from mxgateway.values import to_mx_value


# --- Client.Python-006: concurrent close() is idempotent -------------------


class CountingChannel:
    """A fake gRPC channel that records and stalls on close()."""

    def __init__(self) -> None:
        self.close_calls = 0
        self._gate = asyncio.Event()

    async def close(self) -> None:
        self.close_calls += 1
        # Yield control so a second concurrent close() can interleave at the
        # exact point a check-then-set guard would have left the window open.
        await self._gate.wait()


@pytest.mark.asyncio
async def test_gateway_client_concurrent_close_closes_channel_once() -> None:
    channel = CountingChannel()
    client = GatewayClient(
        options=ClientOptions(endpoint="fake", plaintext=True),
        stub=object(),
        channel=channel,  # type: ignore[arg-type]
    )

    first = asyncio.create_task(client.close())
    second = asyncio.create_task(client.close())
    await asyncio.sleep(0)  # let both coroutines pass the guard if racy

    channel._gate.set()
    await asyncio.gather(first, second)

    assert channel.close_calls == 1


@pytest.mark.asyncio
async def test_galaxy_client_concurrent_close_closes_channel_once() -> None:
    channel = CountingChannel()
    client = GalaxyRepositoryClient(
        options=ClientOptions(endpoint="fake", plaintext=True),
        stub=object(),
        channel=channel,  # type: ignore[arg-type]
    )

    first = asyncio.create_task(client.close())
    second = asyncio.create_task(client.close())
    await asyncio.sleep(0)

    channel._gate.set()
    await asyncio.gather(first, second)

    assert channel.close_calls == 1


@pytest.mark.asyncio
async def test_session_concurrent_close_sends_one_close_session_rpc() -> None:
    gate = asyncio.Event()
    rpc_calls = 0

    class StallingClient:
        async def close_session_raw(self, request: Any) -> pb.CloseSessionReply:
            nonlocal rpc_calls
            rpc_calls += 1
            await gate.wait()
            return pb.CloseSessionReply(
                session_id=request.session_id,
                final_state=pb.SESSION_STATE_CLOSED,
                protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
            )

    from mxgateway.session import Session

    session = Session(client=StallingClient(), session_id="session-1")  # type: ignore[arg-type]

    first = asyncio.create_task(session.close())
    second = asyncio.create_task(session.close())
    await asyncio.sleep(0)

    gate.set()
    await asyncio.gather(first, second)

    assert rpc_calls == 1


# --- Client.Python-007: shared cancelling stream helper --------------------


@pytest.mark.asyncio
async def test_gateway_stream_iterator_cancels_call_on_task_cancellation() -> None:
    """A cancelled gateway stream iterator must explicitly cancel the call."""

    class CancellableStream:
        def __init__(self) -> None:
            self.cancelled = False

        def __aiter__(self) -> "CancellableStream":
            return self

        async def __anext__(self) -> pb.MxEvent:
            await asyncio.Event().wait()  # blocks until cancelled
            raise AssertionError("unreachable")

        def cancel(self) -> None:
            self.cancelled = True

    from mxgateway.client import _canceling_iterator

    stream = CancellableStream()
    iterator = _canceling_iterator(stream, "stream events")

    task = asyncio.create_task(anext(iterator))
    await asyncio.sleep(0)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    # aclose() unwinds the generator's finally block.
    await iterator.aclose()

    assert stream.cancelled


# --- Client.Python-008: non-finite float and bytes value mapping -----------


def test_to_mx_value_rejects_nan() -> None:
    with pytest.raises(ValueError, match="finite"):
        to_mx_value(float("nan"))


def test_to_mx_value_rejects_positive_infinity() -> None:
    with pytest.raises(ValueError, match="finite"):
        to_mx_value(float("inf"))


def test_to_mx_value_rejects_negative_infinity() -> None:
    with pytest.raises(ValueError, match="finite"):
        to_mx_value(float("-inf"))


def test_to_mx_value_accepts_finite_float() -> None:
    assert to_mx_value(3.5).double_value == 3.5


def test_to_mx_value_rejects_non_finite_float_in_sequence() -> None:
    with pytest.raises(ValueError, match="finite"):
        to_mx_value([1.0, math.inf])


# --- Client.Python-011: success == 0 proto3-default ambiguity --------------


def test_ensure_mxaccess_success_ignores_unpopulated_status_entry() -> None:
    """A status entry left at proto3 defaults is not a real MXAccess failure.

    The gateway emits such a placeholder for a null MXSTATUS_PROXY COM entry
    (``MxStatusProxyConverter.ConvertMany``): ``success`` stays 0 but the
    entry carries no failure category. It must not raise ``MxAccessError``.
    """

    reply = pb.MxCommandReply(
        session_id="session-1",
        kind=pb.MX_COMMAND_KIND_SUBSCRIBE_BULK,
        protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
        statuses=[
            pb.MxStatusProxy(),  # all-default: success == 0, category UNSPECIFIED
            pb.MxStatusProxy(  # the gateway's null-entry placeholder
                category=pb.MX_STATUS_CATEGORY_UNKNOWN,
                detected_by=pb.MX_STATUS_SOURCE_UNKNOWN,
            ),
        ],
    )

    assert ensure_mxaccess_success("subscribe bulk", reply) is reply


def test_ensure_mxaccess_success_raises_on_populated_failure_status() -> None:
    """A populated failure status (success == 0 with a failure category) raises."""

    reply = pb.MxCommandReply(
        session_id="session-1",
        kind=pb.MX_COMMAND_KIND_WRITE,
        protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
        statuses=[
            pb.MxStatusProxy(
                success=0,
                category=pb.MX_STATUS_CATEGORY_COMMUNICATION_ERROR,
            ),
        ],
    )

    with pytest.raises(MxAccessError):
        ensure_mxaccess_success("write", reply)


def test_ensure_mxaccess_success_passes_when_status_reports_success() -> None:
    reply = pb.MxCommandReply(
        session_id="session-1",
        kind=pb.MX_COMMAND_KIND_WRITE,
        protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
        statuses=[
            pb.MxStatusProxy(success=1, category=pb.MX_STATUS_CATEGORY_OK),
        ],
    )

    assert ensure_mxaccess_success("write", reply) is reply