37 lines
743 B
Python
37 lines
743 B
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from zvk.bot.event_type import BotEventType
|
|
from zvk.event.event import Event
|
|
from zvk.event.periodic import periodic
|
|
from zvk.event.queue import EventQueue
|
|
from zvk.util.zlogging import logger
|
|
|
|
|
|
@periodic(period_secs=0.1)
|
|
async def periodic_f(counter):
|
|
counter[0] += 1
|
|
logger.debug('tick')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_periodic():
|
|
event_queue = EventQueue()
|
|
event_queue.register_consumer(periodic_f)
|
|
|
|
counter = [0]
|
|
|
|
starting_events = [Event(BotEventType.STARTUP, counter=counter)]
|
|
|
|
queue_task = asyncio.create_task(event_queue.run(starting_events))
|
|
|
|
await asyncio.sleep(0.45)
|
|
assert counter[0] == 5
|
|
|
|
event_queue.omae_wa_mou_shindeiru()
|
|
|
|
assert counter[0] == 5
|
|
|
|
await queue_task
|