Source

araldo / test / app_test.py

Full commit
import gevent
from gevent.queue import Queue
import pytest
import contextlib
from mock import Mock, patch, call
import araldo.app as sut
from araldo.plugins import PluginManager
from araldo.config import Config
from geventwebsocket.exceptions import WebSocketError


@pytest.fixture
def conf1_name():
    return "testfiles/conf1.yaml"


@pytest.fixture
def app(conf1_name):
    config = Config(conf1_name)
    PluginManager.create(config)

    queue = Queue()
    app = sut.WebSocketApp(queue)
    return app


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


@pytest.fixture
def invalid_socket():
    sock = Mock()

    def f(*args):
        raise WebSocketError()
    sock.send.side_effect = f
    return sock


@pytest.fixture
def web_socket():
    web_socket = Mock()
    web_socket.__len__ = Mock(return_value=10)
    return web_socket


@pytest.fixture
def message():
    message = Mock()
    message.__len__ = Mock(return_value=10)
    return message


def test_too_long_exception():
    e = sut.TooLong()
    assert str(e) == sut.TooLong._msg


def test_app_exception():
    e = sut.AppException("testmessage")
    assert str(e) == "testmessage"


## def test_handle_outbound_success(app, valid_socket):
##     app._websockets = {"websocket_1": valid_socket}
##
##     start_response = Mock()
##     environ = {
##         "QUERY_STRING": "xyz"
##     }
##     app._handle_outbound(environ, start_response)
##     assert start_response.call_args == call(
##         '200 OK',
##         [('Content-Type', 'text/html')])
##     assert valid_socket.send.called
##
##
## def test_handle_outbound_close_on_error(app, invalid_socket):
##     app._sockets = set([invalid_socket])
##
##     start_response = Mock()
##     environ = {
##         "QUERY_STRING": "xyz"
##     }
##     app._handle_outbound(environ, start_response)
##     assert invalid_socket.close.called


def _call_with_mocked_handlers(app, route, handler_name):
    environ = {
        "PATH_INFO": route
    }
    start_response = Mock()
    with contextlib.nested(
            patch.object(app, "_handle_websocket")) as (hw,):
            #patch.object(app, "_handle_outbound")) as (hw, ho):
        app(environ, start_response)
        assert getattr(app, handler_name).called


@pytest.mark.parametrize(("route", "handler_name"), [
    ("/websocket/websocket1", "_handle_websocket"),
    #("/outbound", "_handle_outbound"),
])
def test_routing(route, handler_name, app):
        _call_with_mocked_handlers(app, route, handler_name)


def test_not_found(app):
    environ = {
        "PATH_INFO": "/nonexisting"
    }
    start_response = Mock()
    app(environ, start_response)
    assert start_response.call_args == call(
        '404 NOT FOUND', [('Content-Type', 'text/html')])


def test_communicate_with_websocket_success(app, valid_socket):
    message = Mock()
    message.__len__ = Mock(return_value=10)
    app._websockets["websocket_1"] = valid_socket
    valid_socket.receive.return_value = message
    app._communicate_with_websocket(valid_socket, "websocket_1")
    assert "websocket_1" in app._websockets
    assert app._queue.qsize() == 1


def test_communicate_with_websocket_null_message(app, valid_socket):
    valid_socket.receive.return_value = None
    app._websockets = {"websocket_1": valid_socket}
    app._communicate_with_websocket(valid_socket, "websocket_1")
    assert app._queue.qsize() == 0
    assert len(app._websockets) == 0


def test_handle_websocket_termination(app, web_socket, message):
    web_socket.receive.return_value = message
    env = {"wsgi.websocket": web_socket}
    with patch.object(app, "_communicate_with_websocket") as cws:

        def f(web_socket, web_socket_name):
            app._websockets.clear()
        cws.side_effect = f
        app._handle_websocket(env, "websocket_1")


def test_handle_websocket_timeout(app, web_socket):
    env = {"wsgi.websocket": web_socket}
    with patch.object(app, "_communicate_with_websocket") as cws:

        def f(web_socket, websocket_name):
            app._websockets.clear()
            raise gevent.Timeout()
        cws.side_effect = f
        app._handle_websocket(env, "websocket_1")


@pytest.mark.parametrize(("path_info", "expected_websocket_name"), [
    ("/websocket/websocket1", "websocket1"),
    ("/websocket/websocket1/", "websocket1"),

])
def test_parse_websocket_name(path_info, expected_websocket_name):
    websocket_name = sut.WebSocketApp._parse_websocket_name(path_info)
    assert websocket_name == expected_websocket_name


@pytest.mark.parametrize(("path_info"), [
    ("/webesocket",),
    ("/webesocket/a/b",),
    ("/webesocket/a/b/",),
    ("/",),
    ("/x",)
])
def _parse_websocket_name_error():
    with pytest.raises(sut.ApplicationException):
        sut.WebSocketApp._parse_websocket_name("/websocket")