/tests/integration/concurrency/test_helpers.py

"""Unit tests for ``tests/integration/concurrency/helpers.py``.

Each helper has happy-path + at least one edge case. Helpers are
covered here (not via dogfooding in downstream fix tests) so a
helper-level regression surfaces before any consumer PR depends on it.
"""

from __future__ import annotations

import asyncio

import pytest

from .helpers import (
    assert_no_concurrent_overlap,
    assert_unique_outputs,
    with_simulated_cancel,
)

# pure helper-logic tests for tests/integration/concurrency/helpers.py;
# no HTTP, no cassette. Opt out of the tier-enforcement hook in
# tests/integration/conftest.py.
pytestmark = pytest.mark.allow_no_vcr

# ---------------------------------------------------------------------------
# assert_no_concurrent_overlap
# ---------------------------------------------------------------------------


def test_no_concurrent_overlap_serialized_passes() -> None:
    """A serialized stream (each enter followed by exit) passes."""
    events = [
        (0.0, "enter"),
        (0.1, "exit"),
        (0.2, "enter"),
        (0.3, "exit"),
    ]
    assert_no_concurrent_overlap(events)  # type: ignore[arg-type]


def test_no_concurrent_overlap_detects_overlap() -> None:
    """Two enters without an intervening exit raises with index/timestamp."""
    events = [
        (0.0, "enter"),
        (0.1, "enter"),  # overlap at index 1
        (0.2, "exit"),
        (0.3, "exit"),
    ]
    with pytest.raises(AssertionError, match="index 1.*timestamp=0.1"):
        assert_no_concurrent_overlap(events)  # type: ignore[arg-type]


def test_no_concurrent_overlap_unmatched_exit_raises() -> None:
    """An exit before any enter is a caller bug; raise with a clear msg."""
    events = [
        (0.0, "exit"),
    ]
    with pytest.raises(AssertionError, match="unmatched exit.*index 0"):
        assert_no_concurrent_overlap(events)  # type: ignore[arg-type]


def test_no_concurrent_overlap_dangling_enter_raises() -> None:
    """A stream that ends with inflight > 0 raises (caller bug)."""
    events = [
        (0.0, "enter"),
        # no matching exit
    ]
    with pytest.raises(AssertionError, match="ended with inflight=1"):
        assert_no_concurrent_overlap(events)  # type: ignore[arg-type]


def test_no_concurrent_overlap_unknown_kind_raises() -> None:
    """An unknown event kind is a caller bug."""
    events = [(0.0, "wat")]
    with pytest.raises(AssertionError, match="unknown event kind"):
        assert_no_concurrent_overlap(events)  # type: ignore[arg-type]


def test_no_concurrent_overlap_empty_stream_passes() -> None:
    """No events is vacuously serialized."""
    assert_no_concurrent_overlap([])


# ---------------------------------------------------------------------------
# assert_unique_outputs
# ---------------------------------------------------------------------------


async def test_unique_outputs_distinct_passes() -> None:
    """Each invocation returns a fresh value -> no duplicates."""
    counter = 0

    async def _next() -> int:
        nonlocal counter
        counter += 1
        return counter

    outputs = await assert_unique_outputs(_next, 50)
    assert len(outputs) == 50
    assert sorted(outputs) == list(range(1, 51))


async def test_unique_outputs_duplicate_raises_with_indices() -> None:
    """A constant-returning callable trips the duplicate check."""

    async def _const() -> str:
        return "same"

    with pytest.raises(AssertionError, match=r"duplicate output 'same'"):
        await assert_unique_outputs(_const, 3)


async def test_unique_outputs_zero_raises_value_error() -> None:
    """``n_concurrent < 1`` is a misuse and raises early."""

    async def _noop() -> None:
        return None

    with pytest.raises(ValueError, match="n_concurrent must be >= 1"):
        await assert_unique_outputs(_noop, 0)


async def test_unique_outputs_returns_outputs_for_caller_assertions() -> None:
    """Callers may add their own assertions on the returned list."""
    seq = iter(range(5))

    async def _seq() -> int:
        return next(seq)

    outputs = await assert_unique_outputs(_seq, 5)
    # Caller can check shape beyond uniqueness. Use ``sorted`` /
    # ``set`` rather than positional equality so the test stays
    # stable across asyncio event-loop scheduling implementations
    # (CPython runs zero-await coroutines FIFO; uvloop/others may not).
    assert sorted(outputs) == [0, 1, 2, 3, 4]
    assert set(outputs) == set(range(5))


# ---------------------------------------------------------------------------
# with_simulated_cancel
# ---------------------------------------------------------------------------


async def test_with_simulated_cancel_returns_result_when_fast_enough() -> None:
    """A coro that finishes before the cancel fires returns its result."""

    async def _quick() -> str:
        await asyncio.sleep(0.01)
        return "done"

    result = await with_simulated_cancel(_quick(), delay=1.0)
    assert result == "done"


async def test_with_simulated_cancel_returns_cancelled_error_on_timeout() -> None:
    """A coro slower than the cancel returns ``CancelledError``."""

    async def _slow() -> str:
        await asyncio.sleep(10.0)
        return "never"

    result = await with_simulated_cancel(_slow(), delay=0.05, label="slow")
    assert isinstance(result, asyncio.CancelledError)


async def test_with_simulated_cancel_returns_application_exception() -> None:
    """If the coro raises before the cancel, that exception is captured."""

    async def _raises() -> str:
        await asyncio.sleep(0.01)
        raise RuntimeError("boom")

    result = await with_simulated_cancel(_raises(), delay=1.0)
    assert isinstance(result, RuntimeError)
    assert str(result) == "boom"


async def test_with_simulated_cancel_negative_delay_raises() -> None:
    """``delay < 0`` is a misuse and raises early."""

    async def _noop() -> None:
        return None

    with pytest.raises(ValueError, match="delay must be >= 0"):
        await with_simulated_cancel(_noop(), delay=-1.0)


async def test_with_simulated_cancel_propagates_outer_cancellation_to_inner() -> None:
    """Outer cancel of the helper itself must cancel the wrapped task too.

    Otherwise the inner task becomes a stray background task and the
    cancellation is silently consumed. Regression coverage for the
    claude[bot] iter-2 finding.
    """
    inner_started = asyncio.Event()
    inner_finished = False

    async def _slow() -> str:
        nonlocal inner_finished
        inner_started.set()
        try:
            await asyncio.sleep(10.0)
            inner_finished = True
            return "never"
        except asyncio.CancelledError:
            # Re-raise so the cancellation actually takes effect on
            # the task (the helper expects to receive CancelledError
            # from awaiting the task).
            raise

    helper_task = asyncio.create_task(
        with_simulated_cancel(_slow(), delay=10.0, label="outer-cancel-test")
    )

    # Wait until the inner coro has started (so the task is real).
    await inner_started.wait()
    # Give the canceller a tick to schedule, then cancel the OUTER helper.
    await asyncio.sleep(0)
    helper_task.cancel()

    # The helper catches CancelledError and returns it as a value
    # (per its contract); we just verify the outer cancel did NOT
    # leave the inner running.
    try:
        result = await helper_task
        assert isinstance(result, asyncio.CancelledError)
    except asyncio.CancelledError:
        # Acceptable: depending on timing, the outer cancel may
        # propagate before the helper's except block runs.
        pass

    # Yield to let the inner task observe its own cancellation.
    await asyncio.sleep(0)
    assert not inner_finished, (
        "Inner task should have been cancelled when the outer helper was cancelled."
    )


async def test_with_simulated_cancel_cleans_up_canceller_task() -> None:
    """The canceller task should not leak after the helper returns.

    No direct API to inspect; we rely on no warnings about
    "Task was destroyed but it is pending" emerging at teardown. This
    test exists so a future regression that drops the cleanup `cancel()`
    will surface as a noisy test run.
    """

    async def _quick() -> int:
        return 42

    result = await with_simulated_cancel(_quick(), delay=0.5)
    assert result == 42
    # Yield once so any pending cancellation can complete cleanly.
    await asyncio.sleep(0)