2013-09-20 17:04:52 +01:00
|
|
|
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
|
|
|
|
# Copyright(c)2013 NTT corp. All Rights Reserved.
|
|
|
|
#
|
|
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
|
|
# not use this file except in compliance with the License. You may obtain
|
|
|
|
# a copy of the License at
|
|
|
|
#
|
|
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
#
|
|
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
|
|
# License for the specific language governing permissions and limitations
|
|
|
|
# under the License.
|
|
|
|
|
|
|
|
""" Unit tests for websocketproxy """
|
2013-11-13 12:01:05 +00:00
|
|
|
import os
|
|
|
|
import logging
|
2013-09-20 17:04:52 +01:00
|
|
|
import select
|
2013-11-13 12:01:05 +00:00
|
|
|
import shutil
|
|
|
|
import stubout
|
|
|
|
import subprocess
|
|
|
|
import tempfile
|
|
|
|
import time
|
|
|
|
import unittest
|
2013-09-20 17:04:52 +01:00
|
|
|
|
|
|
|
from websockify import websocketproxy
|
|
|
|
|
|
|
|
|
|
|
|
class MockSocket(object):
|
|
|
|
def __init__(*args, **kwargs):
|
|
|
|
pass
|
|
|
|
|
|
|
|
def shutdown(*args):
|
|
|
|
pass
|
|
|
|
|
|
|
|
def close(*args):
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
class WebSocketProxyTest(unittest.TestCase):
|
|
|
|
|
2013-11-13 12:01:05 +00:00
|
|
|
def _init_logger(self, tmpdir):
|
|
|
|
name = 'websocket-unittest'
|
|
|
|
logger = logging.getLogger(name)
|
|
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
logger.propagate = True
|
|
|
|
filename = "%s.log" % (name)
|
|
|
|
handler = logging.FileHandler(filename)
|
|
|
|
handler.setFormatter(logging.Formatter("%(message)s"))
|
|
|
|
logger.addHandler(handler)
|
|
|
|
|
2013-09-20 17:04:52 +01:00
|
|
|
def setUp(self):
|
|
|
|
"""Called automatically before each test."""
|
|
|
|
super(WebSocketProxyTest, self).setUp()
|
|
|
|
self.soc = ''
|
|
|
|
self.stubs = stubout.StubOutForTesting()
|
2013-11-13 12:01:05 +00:00
|
|
|
# Temporary dir for test data
|
|
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
|
|
# Put log somewhere persistent
|
|
|
|
self._init_logger('./')
|
|
|
|
# Mock this out cause it screws tests up
|
|
|
|
self.stubs.Set(os, 'chdir', lambda *args, **kwargs: None)
|
2013-09-20 17:04:52 +01:00
|
|
|
|
|
|
|
def tearDown(self):
|
|
|
|
"""Called automatically after each test."""
|
|
|
|
self.stubs.UnsetAll()
|
2013-11-13 12:01:05 +00:00
|
|
|
shutil.rmtree(self.tmpdir)
|
2013-09-20 17:04:52 +01:00
|
|
|
super(WebSocketProxyTest, self).tearDown()
|
|
|
|
|
2013-11-13 12:01:05 +00:00
|
|
|
def _get_websockproxy(self, **kwargs):
|
|
|
|
return websocketproxy.WebSocketProxy(key=self.tmpdir,
|
|
|
|
web=self.tmpdir,
|
|
|
|
record=self.tmpdir,
|
|
|
|
**kwargs)
|
|
|
|
|
2013-09-20 17:04:52 +01:00
|
|
|
def test_run_wrap_cmd(self):
|
2013-11-13 12:01:05 +00:00
|
|
|
web_socket_proxy = self._get_websockproxy()
|
2013-09-20 17:04:52 +01:00
|
|
|
web_socket_proxy.__dict__["wrap_cmd"] = "wrap_cmd"
|
|
|
|
|
|
|
|
def mock_Popen(*args, **kwargs):
|
|
|
|
return '_mock_cmd'
|
|
|
|
|
|
|
|
self.stubs.Set(subprocess, 'Popen', mock_Popen)
|
|
|
|
web_socket_proxy.run_wrap_cmd()
|
|
|
|
self.assertEquals(web_socket_proxy.spawn_message, True)
|
|
|
|
|
|
|
|
def test_started(self):
|
2013-11-13 12:01:05 +00:00
|
|
|
web_socket_proxy = self._get_websockproxy()
|
2013-09-20 17:04:52 +01:00
|
|
|
web_socket_proxy.__dict__["spawn_message"] = False
|
|
|
|
web_socket_proxy.__dict__["wrap_cmd"] = "wrap_cmd"
|
|
|
|
|
|
|
|
def mock_run_wrap_cmd(*args, **kwargs):
|
|
|
|
web_socket_proxy.__dict__["spawn_message"] = True
|
|
|
|
|
|
|
|
self.stubs.Set(web_socket_proxy, 'run_wrap_cmd', mock_run_wrap_cmd)
|
|
|
|
web_socket_proxy.started()
|
|
|
|
self.assertEquals(web_socket_proxy.__dict__["spawn_message"], True)
|
|
|
|
|
|
|
|
def test_poll(self):
|
2013-11-13 12:01:05 +00:00
|
|
|
web_socket_proxy = self._get_websockproxy()
|
2013-09-20 17:04:52 +01:00
|
|
|
web_socket_proxy.__dict__["wrap_cmd"] = "wrap_cmd"
|
|
|
|
web_socket_proxy.__dict__["wrap_mode"] = "respawn"
|
|
|
|
web_socket_proxy.__dict__["wrap_times"] = [99999999]
|
|
|
|
web_socket_proxy.__dict__["spawn_message"] = True
|
|
|
|
web_socket_proxy.__dict__["cmd"] = None
|
|
|
|
self.stubs.Set(time, 'time', lambda: 100000000.000)
|
|
|
|
web_socket_proxy.poll()
|
|
|
|
self.assertEquals(web_socket_proxy.spawn_message, False)
|
|
|
|
|
|
|
|
def test_new_client(self):
|
2013-11-13 12:01:05 +00:00
|
|
|
web_socket_proxy = self._get_websockproxy()
|
2013-09-20 17:04:52 +01:00
|
|
|
web_socket_proxy.__dict__["verbose"] = "verbose"
|
|
|
|
web_socket_proxy.__dict__["daemon"] = None
|
|
|
|
web_socket_proxy.__dict__["client"] = "client"
|
|
|
|
|
|
|
|
self.stubs.Set(web_socket_proxy, 'socket', MockSocket)
|
|
|
|
|
|
|
|
def mock_select(*args, **kwargs):
|
|
|
|
ins = None
|
|
|
|
outs = None
|
|
|
|
excepts = "excepts"
|
|
|
|
return ins, outs, excepts
|
|
|
|
|
|
|
|
self.stubs.Set(select, 'select', mock_select)
|
2013-11-28 12:33:28 +00:00
|
|
|
self.assertRaises(Exception, web_socket_proxy.new_websocket_client)
|