Source

araldo-websocket / test / websocket_test.py

Full commit
import gevent
from gevent.queue import Queue
import pytest
from mock import Mock, patch
from araldo_websocket.endpoints import websocket_endpoint as sut
from araldo.message import Message


@pytest.fixture
def config():
    return {
        "host": "localhost",
        "websocket_url": "ws://localhost:12345/testsocket/",
        "trace": True
    }


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


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


@pytest.fixture
def create_connection():
    create_connection_mock = Mock()
    ws_mock = Mock()
    create_connection_mock.return_value = ws_mock
    return create_connection_mock


@pytest.fixture
def endpoint(plugin_manager, config, queue, create_connection):
    with patch.object(sut, "create_connection", create_connection):
        result = sut.EndPoint(
            name="ws_1",
            plugin_manager=plugin_manager,
            config=config,
            gevent_queue=queue)
        assert create_connection.called
        return result


def test_plugin_id(endpoint):
    assert endpoint.plugin_id() == "endpoint-websocketclient"


def test_get_next_message(endpoint, queue):
    recv_mock = Mock()
    recv_mock.return_value = "test_message"
    with patch.object(endpoint._websocket, "recv", recv_mock):
        endpoint._get_next_message()
        assert recv_mock.called
        assert queue.qsize() == 1


def test_run_exception(endpoint):
    with patch.object(endpoint, "_get_next_message") as gnm:
        def f():
            raise Exception("test_exception")

        gnm.side_effect = f

        class TooLong(Exception):
            pass

        # terminate test of endless loop using timeout
        with gevent.Timeout(.1, TooLong):
            try:
                endpoint._run()
            except TooLong:
                pass


def test_send(endpoint):
    message = Message(
        origin="test_origin",
        destination="test_destination",
        payload="test_payload")
    with patch.object(endpoint, "_websocket") as ws:
        endpoint.send(message)
        assert ws.send.called