baseimage-tutorial-framework/lib/tfw/event_handlers/test_event_handler.py

187 lines
5.1 KiB
Python
Raw Normal View History

2019-07-11 16:52:15 +00:00
# pylint: disable=redefined-outer-name,attribute-defined-outside-init
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 simulate_message(self, message):
self._on_message(message)
2019-07-11 16:52:15 +00:00
def register_callback(self, callback):
self._on_message = callback
def subscribe(self, *keys):
self.keys.extend(keys)
2019-07-11 16:52:15 +00:00
def unsubscribe(self, *keys):
for key in keys:
self.keys.remove(key)
def send_message(self, message, scope=None):
pass
def close(self):
pass
2019-07-11 16:52:15 +00:00
class MockEventHandlerStub:
def __init__(self):
2019-07-11 16:52:15 +00:00
self.server_connector = None
self.last_message = None
self.cleaned_up = False
2019-07-11 16:52:15 +00:00
self.started = False
2019-07-11 16:52:15 +00:00
def start(self):
self.started = True
def cleanup(self):
self.cleaned_up = True
2019-07-11 16:52:15 +00:00
class MockEventHandler(MockEventHandlerStub):
# pylint: disable=unused-argument
def handle_event(self, message, server_connector):
self.last_message = message
class MockCallable(MockEventHandlerStub):
def __call__(self, message, server_connector):
self.last_message = message
@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):
2019-07-11 16:52:15 +00:00
mock_eh = MockEventHandlerStub()
def handle_event(message, server_connector):
raise RuntimeError(message, server_connector.keys)
2019-07-11 16:52:15 +00:00
mock_eh.handle_event = handle_event
assert not mock_eh.started
eh = MockEventHandlerFactory().build(mock_eh, keys=test_keys)
2019-07-11 16:52:15 +00:00
assert mock_eh.started
assert mock_eh.server_connector is eh.server_connector
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
2019-07-11 16:52:15 +00:00
def test_build_from_object_with_keys(test_keys, test_msg):
mock_eh = MockEventHandler()
2019-07-11 16:52:15 +00:00
mock_eh.keys = test_keys
assert not mock_eh.started
eh = MockEventHandlerFactory().build(mock_eh)
2019-07-11 16:52:15 +00:00
assert mock_eh.server_connector.keys == test_keys
assert eh.server_connector is mock_eh.server_connector
assert mock_eh.started
assert not mock_eh.last_message
eh.server_connector.simulate_message(test_msg)
assert mock_eh.last_message == test_msg
assert not mock_eh.cleaned_up
EventHandler.stop_all_instances()
assert mock_eh.cleaned_up
2019-07-11 16:52:15 +00:00
def test_build_from_simple_object(test_keys, test_msg):
class SimpleMockEventHandler:
# pylint: disable=no-self-use
def handle_event(self, message, server_connector):
raise RuntimeError(message, server_connector)
mock_eh = SimpleMockEventHandler()
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
def test_build_from_callable(test_keys, test_msg):
2019-07-11 16:52:15 +00:00
mock_eh = MockCallable()
assert not mock_eh.started
eh = MockEventHandlerFactory().build(mock_eh, keys=test_keys)
2019-07-11 16:52:15 +00:00
assert mock_eh.started
assert mock_eh.server_connector is eh.server_connector
assert eh.server_connector.keys == test_keys
2019-07-11 16:52:15 +00:00
assert not mock_eh.last_message
eh.server_connector.simulate_message(test_msg)
2019-07-11 16:52:15 +00:00
assert mock_eh.last_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)
2019-07-11 16:52:15 +00:00
def test_build_raises_if_no_key(test_keys):
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)
2019-07-11 16:52:15 +00:00
eh.keys = test_keys
MockEventHandlerFactory().build(eh)