68 lines
1.7 KiB
Python
68 lines
1.7 KiB
Python
# pylint: disable=redefined-outer-name
|
|
from math import inf
|
|
from time import sleep
|
|
from os import urandom
|
|
from random import randint
|
|
|
|
import pytest
|
|
|
|
from .message_queue_handler import MessageQueueHandler
|
|
|
|
|
|
class MockConnector:
|
|
def __init__(self):
|
|
self.callback = None
|
|
self.messages = []
|
|
|
|
def raise_event(self, message):
|
|
self.callback(message, self)
|
|
sleep(0.01)
|
|
|
|
def send_message(self, message):
|
|
self.messages.append(message)
|
|
|
|
|
|
@pytest.fixture
|
|
def handler():
|
|
connector = MockConnector()
|
|
handler = MessageQueueHandler(inf)
|
|
handler.connector = connector
|
|
connector.callback = handler.handle_event
|
|
handler.start()
|
|
yield handler
|
|
handler.cleanup()
|
|
|
|
|
|
@pytest.fixture
|
|
def queue():
|
|
yield {
|
|
'key': 'message.queue',
|
|
'messages': [
|
|
{'originator': urandom(4).hex(), 'message': urandom(16).hex()}
|
|
for _ in range(randint(5, 10))
|
|
]
|
|
}
|
|
|
|
|
|
def test_message_order(handler, queue):
|
|
handler.connector.raise_event(queue)
|
|
old_list = queue['messages']
|
|
new_list = handler.connector.messages
|
|
length = len(old_list)
|
|
assert len(new_list) == length
|
|
for i in range(length):
|
|
unpacked = new_list[i]
|
|
assert unpacked['key'] == 'message.send'
|
|
assert unpacked['originator'] == old_list[i]['originator']
|
|
assert unpacked['typing'] == (i < length-1)
|
|
|
|
|
|
def test_wpm(handler, queue):
|
|
handler.wpm = 10000
|
|
handler.connector.raise_event(queue)
|
|
assert len(handler.connector.messages) == 1
|
|
handler.wpm = 100000000
|
|
handler.connector.raise_event(queue)
|
|
sleep(0.25)
|
|
assert len(handler.connector.messages) == 2*len(queue['messages'])
|