# Path: paco/throttle.py
# @decorate
# def throttle(coro, limit=1, timeframe=1,
#              return_value=None, raise_exception=False):
#     """
#     Creates a throttled coroutine function that only invokes
#     ``coro`` at most once per every time frame of seconds or milliseconds.
# 
#     Provide options to indicate whether func should be invoked on the
#     leading and/or trailing edge of the wait timeout.
# 
#     Subsequent calls to the throttled coroutine
#     return the result of the last coroutine invocation.
# 
#     This function can be used as decorator.
# 
#     Arguments:
#         coro (coroutinefunction):
#             coroutine function to wrap with throttle strategy.
#         limit (int):
#             number of coroutine allowed execution in the given time frame.
#         timeframe (int|float):
#             throttle limit time frame in seconds.
#         return_value (mixed):
#             optional return if the throttle limit is reached.
#             Returns the latest returned value by default.
#         raise_exception (bool):
#             raise exception if throttle limit is reached.
# 
#     Raises:
#         RuntimeError: if cannot throttle limit reached (optional).
# 
#     Returns:
#         coroutinefunction
# 
#     Usage::
# 
#         async def mul_2(num):
#             return num * 2
# 
#         # Use as simple wrapper
#         throttled = paco.throttle(mul_2, limit=1, timeframe=2)
#         await throttled(2)
#         # => 4
#         await throttled(3)  # ignored!
#         # => 4
#         await asyncio.sleep(2)
#         await throttled(3)  # executed!
#         # => 6
# 
#         # Use as decorator
#         @paco.throttle(limit=1, timeframe=2)
#         async def mul_2(num):
#             return num * 2
# 
#         await mul_2(2)
#         # => 4
#         await mul_2(3)  # ignored!
#         # => 4
#         await asyncio.sleep(2)
#         await mul_2(3)  # executed!
#         # => 6
# 
#     """
#     assert_corofunction(coro=coro)
# 
#     # Store execution limits
#     limit = max(int(limit), 1)
#     remaning = limit
# 
#     # Turn seconds in milliseconds
#     timeframe = timeframe * 1000
# 
#     # Keep call state
#     last_call = now()
#     # Cache latest retuned result
#     result = None
# 
#     def stop():
#         if raise_exception:
#             raise RuntimeError('paco: coroutine throttle limit exceeded')
#         if return_value:
#             return return_value
#         return result
# 
#     def elapsed():
#         return now() - last_call
# 
#     @asyncio.coroutine
#     def wrapper(*args, **kw):
#         nonlocal result
#         nonlocal remaning
#         nonlocal last_call
# 
#         if elapsed() > timeframe:
#             # Reset reamining calls counter
#             remaning = limit
#             # Update last call time
#             last_call = now()
#         elif elapsed() < timeframe and remaning <= 0:
#             return stop()
# 
#         # Decrease remaining limit
#         remaning -= 1
# 
#         # Schedule coroutine passing arguments and cache result
#         result = yield from coro(*args, **kw)
#         return result
# 
#     return wrapper
# 
# Path: tests/helpers.py
# def run_in_loop(coro, *args, **kw):
#     loop = asyncio.get_event_loop()
#     if asyncio.iscoroutinefunction(coro) or isfunction(coro):
#         coro = coro(*args, **kw)
#     return loop.run_until_complete(coro)


import time
import pytest
import asyncio
from paco import throttle
from .helpers import run_in_loop

def test_throttle_return_value():
    task = throttle(coro, limit=2, timeframe=0.2, return_value='ignored')
    assert run_in_loop(task, 1) == 1
    assert run_in_loop(task, 2) == 2
    assert run_in_loop(task, 3) == 'ignored'
    assert run_in_loop(task, 4) == 'ignored'
    time.sleep(0.3)
    assert run_in_loop(task, 5) == 5
    assert run_in_loop(task, 6) == 6
    assert run_in_loop(task, 7) == 'ignored'
    assert run_in_loop(task, 8) == 'ignored'


def test_throttle_raise_exception():
    task = throttle(coro, limit=1, timeframe=1, raise_exception=True)
    assert run_in_loop(task, 1) == 1

    with pytest.raises(RuntimeError):
        run_in_loop(task, 2)


def test_throttle_invalid_coro():
    with pytest.raises(TypeError):
        throttle(None)


def test_decorator():
    task = throttle(limit=2, timeframe=0.2)(coro)
    assert run_in_loop(task, 1) == 1

# Path: paco/throttle.py
# @decorate
# def throttle(coro, limit=1, timeframe=1,
#              return_value=None, raise_exception=False):
#     """
#     Creates a throttled coroutine function that only invokes
#     ``coro`` at most once per every time frame of seconds or milliseconds.
# 
#     Provide options to indicate whether func should be invoked on the
#     leading and/or trailing edge of the wait timeout.
# 
#     Subsequent calls to the throttled coroutine
#     return the result of the last coroutine invocation.
# 
#     This function can be used as decorator.
# 
#     Arguments:
#         coro (coroutinefunction):
#             coroutine function to wrap with throttle strategy.
#         limit (int):
#             number of coroutine allowed execution in the given time frame.
#         timeframe (int|float):
#             throttle limit time frame in seconds.
#         return_value (mixed):
#             optional return if the throttle limit is reached.
#             Returns the latest returned value by default.
#         raise_exception (bool):
#             raise exception if throttle limit is reached.
# 
#     Raises:
#         RuntimeError: if cannot throttle limit reached (optional).
# 
#     Returns:
#         coroutinefunction
# 
#     Usage::
# 
#         async def mul_2(num):
#             return num * 2
# 
#         # Use as simple wrapper
#         throttled = paco.throttle(mul_2, limit=1, timeframe=2)
#         await throttled(2)
#         # => 4
#         await throttled(3)  # ignored!
#         # => 4
#         await asyncio.sleep(2)
#         await throttled(3)  # executed!
#         # => 6
# 
#         # Use as decorator
#         @paco.throttle(limit=1, timeframe=2)
#         async def mul_2(num):
#             return num * 2
# 
#         await mul_2(2)
#         # => 4
#         await mul_2(3)  # ignored!
#         # => 4
#         await asyncio.sleep(2)
#         await mul_2(3)  # executed!
#         # => 6
# 
#     """
#     assert_corofunction(coro=coro)
# 
#     # Store execution limits
#     limit = max(int(limit), 1)
#     remaning = limit
# 
#     # Turn seconds in milliseconds
#     timeframe = timeframe * 1000
# 
#     # Keep call state
#     last_call = now()
#     # Cache latest retuned result
#     result = None
# 
#     def stop():
#         if raise_exception:
#             raise RuntimeError('paco: coroutine throttle limit exceeded')
#         if return_value:
#             return return_value
#         return result
# 
#     def elapsed():
#         return now() - last_call
# 
#     @asyncio.coroutine
#     def wrapper(*args, **kw):
#         nonlocal result
#         nonlocal remaning
#         nonlocal last_call
# 
#         if elapsed() > timeframe:
#             # Reset reamining calls counter
#             remaning = limit
#             # Update last call time
#             last_call = now()
#         elif elapsed() < timeframe and remaning <= 0:
#             return stop()
# 
#         # Decrease remaining limit
#         remaning -= 1
# 
#         # Schedule coroutine passing arguments and cache result
#         result = yield from coro(*args, **kw)
#         return result
# 
#     return wrapper
# 
# Path: tests/helpers.py
# def run_in_loop(coro, *args, **kw):
#     loop = asyncio.get_event_loop()
#     if asyncio.iscoroutinefunction(coro) or isfunction(coro):
#         coro = coro(*args, **kw)
#     return loop.run_until_complete(coro)


# Path: tests/throttle_test.py
import time
import pytest
import asyncio
from paco import throttle
from .helpers import run_in_loop


def test_throttle_return_value():
    task = throttle(coro, limit=2, timeframe=0.2, return_value='ignored')
    assert run_in_loop(task, 1) == 1
    assert run_in_loop(task, 2) == 2
    assert run_in_loop(task, 3) == 'ignored'
    assert run_in_loop(task, 4) == 'ignored'
    time.sleep(0.3)
    assert run_in_loop(task, 5) == 5
    assert run_in_loop(task, 6) == 6
    assert run_in_loop(task, 7) == 'ignored'
    assert run_in_loop(task, 8) == 'ignored'


def test_throttle_raise_exception():
    task = throttle(coro, limit=1, timeframe=1, raise_exception=True)
    assert run_in_loop(task, 1) == 1

    with pytest.raises(RuntimeError):
        run_in_loop(task, 2)


def test_throttle_invalid_coro():
    with pytest.raises(TypeError):
        throttle(None)


def test_decorator():
    task = throttle(limit=2, timeframe=0.2)(coro)
    assert run_in_loop(task, 1) == 1

    assert run_in_loop(task, 2) == 2