52 lines
1.0 KiB
Python
52 lines
1.0 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from zvk.bot.event_type import BotEventType
|
|
from zvk.event.consumer import event_consumer
|
|
from zvk.event.event import Event
|
|
from zvk.event.periodic import periodic
|
|
from zvk.event.queue import EventQueue
|
|
|
|
|
|
@periodic(period_secs=0.1)
|
|
async def loop1():
|
|
yield Event('inc')
|
|
|
|
|
|
@periodic(period_secs=0.1)
|
|
async def loop2():
|
|
yield Event('inc')
|
|
|
|
|
|
@event_consumer(consumes=['inc'])
|
|
async def inc(counter):
|
|
counter[0] += 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_event_cancellation():
|
|
event_queue = EventQueue()
|
|
|
|
counter = [0]
|
|
|
|
event_queue.register_consumer(loop1)
|
|
event_queue.register_consumer(loop2)
|
|
event_queue.register_consumer(inc)
|
|
|
|
starting_events = [Event(BotEventType.STARTUP, counter=counter)]
|
|
|
|
queue_task = asyncio.create_task(event_queue.run(starting_events))
|
|
|
|
await asyncio.sleep(0.15)
|
|
assert counter[0] == 4
|
|
|
|
event_queue.deregister_consumer(loop2)
|
|
await asyncio.sleep(0.2)
|
|
assert counter[0] == 6
|
|
|
|
event_queue.omae_wa_mou_shindeiru()
|
|
await queue_task
|
|
|
|
assert counter[0] == 6
|