async def wait_stopped(self):
    """Waits for the task to finish unwinding after ``stop()``. A no-op if
    the task never started or is the current task."""
    task = self.__task
    if task is not None and task is not asyncio.current_task():
        # asyncio.wait does not cancel the task and does not raise the
        # task's exception. Cancellation of the caller propagates normally.
        await asyncio.wait({task})

"""
Tests demonstrating that AsyncRepeatingTask.wait_stopped() mishandles
cancellation of its *caller*.

`await task` makes the polling task the awaiting coroutine's `_fut_waiter`, so
cancelling the caller of stop() (an ``asyncio.wait_for`` deadline, an
``asyncio.timeout`` block, a TaskGroup tearing down) has two effects:

1. the cancellation is forwarded *into* the polling task, aborting whatever
   cancellation cleanup it was doing (e.g. a persistent store finishing a
   write, aiohttp connection teardown), and
2. the caller's own cancellation is then absorbed by the blanket
   ``except asyncio.CancelledError``, so stop() keeps going, closes the
   transport out from under the still-unwinding poll, and reports success.

Both tests assert the *desired* behavior, so they FAIL on the current code and
pass once wait_stopped() waits without forwarding cancellation, e.g.::

    async def wait_stopped(self):
        task = self.__task
        if task is not None and task is not asyncio.current_task():
            await asyncio.wait({task})

(the same pattern join_handle() in this module already uses, for the reason
its comment explains).
"""

import asyncio
from unittest.mock import AsyncMock, MagicMock

import pytest

from ldclient.testing.impl.datasource.test_async_polling import make_processor


class TestStopUnderExternalCancellation:
    @pytest.mark.asyncio
    async def test_stop_under_deadline_reports_timeout_and_does_not_abandon_cleanup(self):
        # An application shutting down under a deadline:
        #   await asyncio.wait_for(client.close(), timeout=...)
        # If the in-flight poll's cancellation cleanup outlives the deadline,
        # the caller must see TimeoutError; the cleanup must not be aborted,
        # and the transport must not be closed under the live poll.
        events = []
        started = asyncio.Event()

        async def slow_poll():
            started.set()
            try:
                await asyncio.sleep(60)
            except asyncio.CancelledError:
                events.append('cleanup_started')
                # Simulates a store commit / connection teardown that takes
                # longer than the shutdown deadline below.
                await asyncio.sleep(0.3)
                events.append('cleanup_finished')
                raise

        async def close():
            events.append('transport_closed')

        requester = MagicMock()
        requester.get_all_data = slow_poll
        requester.close = close

        processor = make_processor(requester=requester)
        processor.start()
        await asyncio.wait_for(started.wait(), timeout=1.0)

        # Desired: the missed deadline is reported. Today wait_for() returns
        # normally, because wait_stopped() swallows the CancelledError that
        # wait_for delivers to stop().
        with pytest.raises(asyncio.TimeoutError):
            await asyncio.wait_for(processor.stop(), timeout=0.1)

        # Give the polling task time to finish unwinding on its own.
        await asyncio.sleep(0.4)

        # Desired: the cleanup ran to completion instead of being aborted by
        # the forwarded cancellation...
        assert events[:2] == ['cleanup_started', 'cleanup_finished']
        # ...and the transport was never closed while the poll was live.
        if 'transport_closed' in events:
            assert events.index('transport_closed') > events.index('cleanup_finished')

    @pytest.mark.asyncio
    async def test_cancelling_a_task_blocked_in_stop_actually_cancels_it(self):
        # A TaskGroup sibling failure or lifespan teardown cancels the task
        # that is running stop(). Desired: that task ends cancelled, and the
        # cancellation is not forwarded into the polling task's cleanup.
        cleanup_completed = asyncio.Event()
        started = asyncio.Event()

        async def slow_poll():
            started.set()
            try:
                await asyncio.sleep(60)
            except asyncio.CancelledError:
                await asyncio.sleep(0.2)
                cleanup_completed.set()
                raise

        requester = MagicMock()
        requester.get_all_data = slow_poll
        requester.close = AsyncMock()

        processor = make_processor(requester=requester)
        processor.start()
        await asyncio.wait_for(started.wait(), timeout=1.0)

        shutdown = asyncio.ensure_future(processor.stop())
        await asyncio.sleep(0.05)  # shutdown is now blocked inside wait_stopped()
        shutdown.cancel()

        # Desired: the cancellation propagates. Today stop() swallows it and
        # returns normally, so the application's cancellation is lost.
        with pytest.raises(asyncio.CancelledError):
            await shutdown
        assert shutdown.cancelled()

        # Desired: the polling task's cleanup still ran to completion.
        await asyncio.wait_for(cleanup_completed.wait(), timeout=1.0)