# Path: aiostream/manager.py
# class TaskGroup:
#     def __init__(self):
#         self._pending = set()
# 
#     async def __aenter__(self):
#         return self
# 
#     async def __aexit__(self, *args):
#         while self._pending:
#             task = self._pending.pop()
#             await self.cancel_task(task)
# 
#     def create_task(self, coro):
#         task = asyncio.ensure_future(coro)
#         self._pending.add(task)
#         return task
# 
#     async def wait_any(self, tasks):
#         done, _ = await asyncio.wait(tasks, return_when="FIRST_COMPLETED")
#         self._pending -= done
#         return done
# 
#     async def wait_all(self, tasks):
#         if not tasks:
#             return set()
#         done, _ = await asyncio.wait(tasks)
#         self._pending -= done
#         return done
# 
#     async def cancel_task(self, task):
#         try:
#             # The task is already cancelled
#             if task.cancelled():
#                 pass
#             # The task is already finished
#             elif task.done():
#                 # Discard the pending exception (if any).
#                 # This makes sense since we don't know in which context the exception
#                 # was meant to be processed. For instance, a `StopAsyncIteration`
#                 # might be raised to notify that the end of a streamer has been reached.
#                 task.exception()
#             # The task needs to be cancelled and awaited
#             else:
#                 task.cancel()
#                 try:
#                     await task
#                 except asyncio.CancelledError:
#                     pass
#                 # Silence any exception raised while cancelling the task.
#                 # This might happen if the `CancelledError` is silenced, and the
#                 # corresponding async generator returns, causing the `anext` call
#                 # to raise a `StopAsyncIteration`.
#                 except Exception:
#                     pass
#         finally:
#             self._pending.discard(task)


import asyncio
import pytest
from aiostream.manager import TaskGroup

async def task1():
    await asyncio.sleep(1)
    return 1


async def task2():
    await asyncio.sleep(0)
    return 2


async def task3():
    await asyncio.sleep(0)
    raise ValueError(3)


async def task4():
    await asyncio.sleep(1)
    raise ValueError(4)


async def task5():
    try:
        await asyncio.sleep(1)
    finally:
        raise ValueError(5)


@pytest.mark.asyncio
async def test_task_group_cleanup(event_loop):

# Path: aiostream/manager.py
# class TaskGroup:
#     def __init__(self):
#         self._pending = set()
# 
#     async def __aenter__(self):
#         return self
# 
#     async def __aexit__(self, *args):
#         while self._pending:
#             task = self._pending.pop()
#             await self.cancel_task(task)
# 
#     def create_task(self, coro):
#         task = asyncio.ensure_future(coro)
#         self._pending.add(task)
#         return task
# 
#     async def wait_any(self, tasks):
#         done, _ = await asyncio.wait(tasks, return_when="FIRST_COMPLETED")
#         self._pending -= done
#         return done
# 
#     async def wait_all(self, tasks):
#         if not tasks:
#             return set()
#         done, _ = await asyncio.wait(tasks)
#         self._pending -= done
#         return done
# 
#     async def cancel_task(self, task):
#         try:
#             # The task is already cancelled
#             if task.cancelled():
#                 pass
#             # The task is already finished
#             elif task.done():
#                 # Discard the pending exception (if any).
#                 # This makes sense since we don't know in which context the exception
#                 # was meant to be processed. For instance, a `StopAsyncIteration`
#                 # might be raised to notify that the end of a streamer has been reached.
#                 task.exception()
#             # The task needs to be cancelled and awaited
#             else:
#                 task.cancel()
#                 try:
#                     await task
#                 except asyncio.CancelledError:
#                     pass
#                 # Silence any exception raised while cancelling the task.
#                 # This might happen if the `CancelledError` is silenced, and the
#                 # corresponding async generator returns, causing the `anext` call
#                 # to raise a `StopAsyncIteration`.
#                 except Exception:
#                     pass
#         finally:
#             self._pending.discard(task)


# Path: tests/test_task_group.py
import asyncio
import pytest
from aiostream.manager import TaskGroup


async def task1():
    await asyncio.sleep(1)
    return 1


async def task2():
    await asyncio.sleep(0)
    return 2


async def task3():
    await asyncio.sleep(0)
    raise ValueError(3)


async def task4():
    await asyncio.sleep(1)
    raise ValueError(4)


async def task5():
    try:
        await asyncio.sleep(1)
    finally:
        raise ValueError(5)


@pytest.mark.asyncio
async def test_task_group_cleanup(event_loop):

    async with TaskGroup() as group: