import asyncio

import pytest

from livekit.agents.utils import aio

pytestmark = [pytest.mark.unit, pytest.mark.concurrent]


async def test_channel():
    tx = rx = aio.Chan[int]()
    sum = 0

    async def test_task():
        nonlocal sum
        while True:
            try:
                sum = sum + await rx.recv()
            except aio.ChanClosed:
                break

    t = asyncio.create_task(test_task())
    for _ in range(10):
        await tx.send(1)

    tx.close()
    await t
    assert sum == 10


async def test_interval():
    interval = aio.interval(0.1)

    _ = asyncio.get_event_loop()
    async for i in interval:
        if i == 3:
            break


async def test_sleep():
    await aio.sleep(0)

    sleep = aio.sleep(5)
    sleep.reset(0.1)
    await sleep


class TestAsyncAtomicCounter:
    async def test_basic_increment_decrement(self):
        counter = aio.AsyncAtomicCounter()
        assert await counter.get() == 0

        assert await counter.increment() == 1
        assert await counter.increment(5) == 6
        assert await counter.decrement() == 5
        assert await counter.decrement(3) == 2

    async def test_initial_value(self):
        counter = aio.AsyncAtomicCounter(initial=10)
        assert await counter.get() == 10

    async def test_set(self):
        counter = aio.AsyncAtomicCounter()
        await counter.set(42)
        assert await counter.get() == 42

    async def test_compare_and_swap_success(self):
        counter = aio.AsyncAtomicCounter(initial=5)
        assert await counter.compare_and_swap(5, 10) is True
        assert await counter.get() == 10

    async def test_compare_and_swap_failure(self):
        counter = aio.AsyncAtomicCounter(initial=5)
        assert await counter.compare_and_swap(99, 20) is False
        assert await counter.get() == 5

    async def test_get_and_reset(self):
        counter = aio.AsyncAtomicCounter(initial=7)
        prev = await counter.get_and_reset()
        assert prev == 7
        assert await counter.get() == 0

    async def test_get_and_reset_custom_value(self):
        counter = aio.AsyncAtomicCounter(initial=15)
        prev = await counter.get_and_reset(reset_value=3)
        assert prev == 15
        assert await counter.get() == 3

    async def test_get_nowait(self):
        counter = aio.AsyncAtomicCounter(initial=99)
        assert counter.get_nowait() == 99
        await counter.increment()
        assert counter.get_nowait() == 100

    async def test_concurrent_increments(self):
        counter = aio.AsyncAtomicCounter()
        num_tasks = 10
        increments_per_task = 100

        async def increment_many():
            for _ in range(increments_per_task):
                await counter.increment()

        await asyncio.gather(*(increment_many() for _ in range(num_tasks)))
        assert await counter.get() == num_tasks * increments_per_task