mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2024-11-12 19:27:17 +00:00
148 lines
3.9 KiB
Python
148 lines
3.9 KiB
Python
# pylint: disable=redefined-outer-name
|
|
from secrets import token_urlsafe
|
|
from random import randint
|
|
|
|
import pytest
|
|
|
|
from .event_handler_factory import EventHandlerFactoryBase
|
|
from .event_handler import EventHandler
|
|
|
|
|
|
class MockEventHandlerFactory(EventHandlerFactoryBase):
|
|
def _build_server_connector(self):
|
|
return MockServerConnector()
|
|
|
|
|
|
class MockServerConnector:
|
|
def __init__(self):
|
|
self.keys = []
|
|
self._on_message = None
|
|
|
|
def register_callback(self, callback):
|
|
self._on_message = callback
|
|
|
|
def simulate_message(self, message):
|
|
self._on_message(message)
|
|
|
|
def subscribe(self, keys):
|
|
self.keys.extend(keys)
|
|
|
|
def unsubscribe(self, keys):
|
|
for key in keys:
|
|
self.keys.remove(key)
|
|
|
|
def send_message(self, message, scope=None):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class MockEventHandler:
|
|
def __init__(self):
|
|
self.cleaned_up = False
|
|
|
|
def handle_event(self, message, server_connector):
|
|
pass
|
|
|
|
def cleanup(self):
|
|
self.cleaned_up = True
|
|
|
|
|
|
@pytest.fixture
|
|
def test_msg():
|
|
yield token_urlsafe(randint(16, 64))
|
|
|
|
|
|
@pytest.fixture
|
|
def test_keys():
|
|
yield [
|
|
token_urlsafe(randint(2, 8))
|
|
for _ in range(randint(16, 32))
|
|
]
|
|
|
|
|
|
def test_build_from_object(test_keys, test_msg):
|
|
mock_eh = MockEventHandler()
|
|
def test_handle_event(message, server_connector):
|
|
raise RuntimeError(message, server_connector.keys)
|
|
mock_eh.handle_event = test_handle_event
|
|
eh = MockEventHandlerFactory().build(mock_eh, keys=test_keys)
|
|
|
|
with pytest.raises(RuntimeError) as err:
|
|
eh.server_connector.simulate_message(test_msg)
|
|
msg, keys = err.args
|
|
assert msg == test_msg
|
|
assert keys == test_keys
|
|
assert not mock_eh.cleaned_up
|
|
eh.stop()
|
|
assert mock_eh.cleaned_up
|
|
|
|
|
|
def test_build_from_object_with_keys(test_keys):
|
|
mock_eh = MockEventHandler()
|
|
mock_eh.keys = test_keys # pylint: disable=attribute-defined-outside-init
|
|
eh = MockEventHandlerFactory().build(mock_eh)
|
|
|
|
assert not mock_eh.cleaned_up
|
|
EventHandler.stop_all_instances()
|
|
assert mock_eh.cleaned_up
|
|
assert eh.server_connector.keys == test_keys
|
|
|
|
|
|
def test_build_from_callable(test_keys, test_msg):
|
|
class SomeCallable:
|
|
def __init__(self):
|
|
self.message = None
|
|
self.cleaned_up = False
|
|
def __call__(self, message, server_connector):
|
|
self.message = message
|
|
def cleanup(self):
|
|
self.cleaned_up = True
|
|
|
|
mock_eh = SomeCallable()
|
|
eh = MockEventHandlerFactory().build(mock_eh, keys=test_keys)
|
|
|
|
assert eh.server_connector.keys == test_keys
|
|
assert not mock_eh.message
|
|
eh.server_connector.simulate_message(test_msg)
|
|
assert mock_eh.message == test_msg
|
|
assert not mock_eh.cleaned_up
|
|
eh.stop()
|
|
assert mock_eh.cleaned_up
|
|
|
|
|
|
def test_build_from_function(test_keys, test_msg):
|
|
def some_function(message, server_connector):
|
|
raise RuntimeError(message, server_connector.keys)
|
|
eh = MockEventHandlerFactory().build(some_function, keys=test_keys)
|
|
|
|
assert eh.server_connector.keys == test_keys
|
|
with pytest.raises(RuntimeError) as err:
|
|
eh.server_connector.simulate_message(test_msg)
|
|
msg, keys = err.args
|
|
assert msg == test_msg
|
|
assert keys == test_keys
|
|
|
|
|
|
def test_build_from_lambda(test_keys, test_msg):
|
|
def assert_messages_equal(msg):
|
|
assert msg == test_msg
|
|
fun = lambda msg, sc: assert_messages_equal(msg)
|
|
eh = MockEventHandlerFactory().build(fun, keys=test_keys)
|
|
eh.server_connector.simulate_message(test_msg)
|
|
|
|
|
|
def test_build_raises_if_no_key():
|
|
eh = MockEventHandler()
|
|
with pytest.raises(ValueError):
|
|
MockEventHandlerFactory().build(eh)
|
|
|
|
def test_handle_event(*_):
|
|
pass
|
|
with pytest.raises(ValueError):
|
|
MockEventHandlerFactory().build(test_handle_event)
|
|
|
|
with pytest.raises(ValueError):
|
|
MockEventHandlerFactory().build(lambda msg, sc: None)
|