1. Bernhard Biskup
  2. araldo

Source

araldo / test / endpoints_test.py

import pytest
from gevent import Timeout
from mock import Mock
from gevent.queue import Queue
from araldo.endpoints import PluginException
from araldo.endpoints import mock_endpoint


@pytest.fixture
def queue():
    return Queue()


@pytest.fixture
def marshal_json():
    return Mock()


@pytest.fixture
def plugin_manager(marshal_json):
    plugin_manager = Mock()
    plugin_manager.plugin_instances.return_value = {
        "araldo.marshalling": {
            "marshal-json": marshal_json
        }
    }
    return plugin_manager


@pytest.fixture
def conf():
    return {
        "channel": "channel_1"
    }


@pytest.fixture
def endpoint(conf, queue, plugin_manager, marshal_json):
    return mock_endpoint.EndPoint(
        name="testname",
        plugin_manager=plugin_manager,
        config=conf,
        gevent_queue=queue)


def test_plugin_exception():
    exception = PluginException("testmessage")
    assert "testmessage" == str(exception)


def test_endpoint(endpoint, conf, queue, plugin_manager, marshal_json):
    assert "testname" == endpoint.name()
    assert conf == endpoint.config()
    assert queue == endpoint.gevent_queue
    assert plugin_manager == endpoint.plugin_manager()
    assert endpoint.marshalling() == marshal_json


def test_run(endpoint):
    timeout = Timeout(.2)
    timeout.start()
    with pytest.raises(Timeout):
        endpoint.start()
        endpoint.join()


def test_check(endpoint):
    endpoint.check()


def test_send(endpoint):
    message = Mock()
    endpoint.send(message)