baseimage-tutorial-framework/tfw/internals/event_handling/test_event_handler.py

# pylint: disable=redefined-outer-name,attribute-defined-outside-init
from secrets import token_urlsafe
from random import randint

import pytest

from .event_handler_factory_base import EventHandlerFactoryBase
from .event_handler import EventHandler
from .control_event_handler import ControlEventHandler
from .fsm_aware_event_handler import FSMAwareEventHandler


class MockEventHandlerFactory(EventHandlerFactoryBase):
    def _build_connector(self):
        return MockConnector()


class MockConnector:
    def __init__(self):
        self.keys = []
        self._on_message = None

    def simulate_message(self, message):
        self._on_message(message)

    def register_callback(self, callback):
        self._on_message = callback

    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 MockEventHandlerStub:
    def __init__(self):
        self.connector = None
        self.last_message = None
        self.cleaned_up = False
        self.started = False

    def start(self):
        self.started = True

    def cleanup(self):
        self.cleaned_up = True


class MockEventHandler(MockEventHandlerStub):
    # pylint: disable=unused-argument
    def handle_event(self, message, connector):
        self.last_message = message


class MockCallable(MockEventHandlerStub):
    def __call__(self, message, 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):
    mock_eh = MockEventHandlerStub()
    def handle_event(message, connector):
        raise RuntimeError(message, connector.keys)
    mock_eh.handle_event = handle_event

    assert not mock_eh.started
    eh = MockEventHandlerFactory().build(mock_eh, keys=test_keys)

    assert mock_eh.started
    assert mock_eh.connector is eh.connector
    with pytest.raises(RuntimeError) as err:
        eh.connector.simulate_message(test_msg)
    msg, keys = err.value.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, test_msg):
    mock_eh = MockEventHandler()
    mock_eh.keys = test_keys

    assert not mock_eh.started
    eh = MockEventHandlerFactory().build(mock_eh)

    assert mock_eh.connector.keys == test_keys
    assert eh.connector is mock_eh.connector
    assert mock_eh.started
    assert not mock_eh.last_message
    eh.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


def test_build_from_simple_object(test_keys, test_msg):
    class SimpleMockEventHandler:
        # pylint: disable=no-self-use
        def handle_event(self, message, connector):
            raise RuntimeError(message)

    mock_eh = SimpleMockEventHandler()
    eh = MockEventHandlerFactory().build(mock_eh, keys=test_keys)

    with pytest.raises(RuntimeError) as err:
        eh.connector.simulate_message(test_msg)
    msg = err.value.args[0]
    assert msg == test_msg


def test_build_type():
    class SomeControlHandler:
        keys = ['cica']
        type_id = 'ControlEventHandler'
        # pylint: disable=no-self-use
        def handle_event(self, *_):
            pass

    eh = MockEventHandlerFactory().build(SomeControlHandler())
    assert isinstance(eh, ControlEventHandler)

    eh = MockEventHandlerFactory().build(SomeControlHandler(), event_handler_type=EventHandler)
    assert type(eh) == EventHandler  # pylint: disable=unidiomatic-typecheck

    SomeControlHandler.type_id = 'FSMAwareEventHandler'
    eh = MockEventHandlerFactory().build(SomeControlHandler())
    assert isinstance(eh, FSMAwareEventHandler)

    eh = MockEventHandlerFactory().build(lambda *_: None, keys=['cica'])
    assert isinstance(eh, EventHandler)


def test_build_from_callable(test_keys, test_msg):
    mock_eh = MockCallable()

    assert not mock_eh.started
    eh = MockEventHandlerFactory().build(mock_eh, keys=test_keys)

    assert mock_eh.started
    assert mock_eh.connector is eh.connector
    assert eh.connector.keys == test_keys
    assert not mock_eh.last_message
    eh.connector.simulate_message(test_msg)
    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, connector):
        raise RuntimeError(message, connector.keys)
    eh = MockEventHandlerFactory().build(some_function, keys=test_keys)

    assert eh.connector.keys == test_keys
    with pytest.raises(RuntimeError) as err:
        eh.connector.simulate_message(test_msg)
    msg, keys = err.value.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.connector.simulate_message(test_msg)


def test_build_raises_if_no_key(test_keys):
    eh = MockEventHandler()
    with pytest.raises(ValueError):
        MockEventHandlerFactory().build(eh)

    def handle_event(*_):
        pass
    with pytest.raises(ValueError):
        MockEventHandlerFactory().build(handle_event)

    with pytest.raises(ValueError):
        MockEventHandlerFactory().build(lambda msg, sc: None)

    WithKeysEventHandler = EventHandler
    WithKeysEventHandler.keys = test_keys
    MockEventHandlerFactory().build(eh, event_handler_type=WithKeysEventHandler)

    eh.keys = test_keys
    MockEventHandlerFactory().build(eh)