araldo / test / app_test.py

from gevent.queue import Queue
import pytest
import contextlib
from mock import Mock, patch, call
import araldo.app as sut
from araldo.plugin_manager import PluginManager
from araldo.config import Config


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


@pytest.fixture
def config(conf1_name):
    return Config(conf1_name)


@pytest.fixture
def plugin_manager(config):
    return PluginManager(config)


@pytest.fixture
def app(plugin_manager):
    queue = Queue()
    app = sut.WebSocketApp(config, queue, plugin_manager)
    return app


@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')])


@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")
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.