Merge pull request #88 from andfoy/server_tests

PR: Add server tests

circle.yml

@@ -11,7 +11,7 @@ machine:
     TRAVIS_OS_NAME: "linux"
     CONDA_DEPENDENCIES_FLAGS: "--quiet"
     CONDA_DEPENDENCIES: "pytest pytest-cov"
-    PIP_DEPENDENCIES: "coveralls coloredlogs pytest-qt pytest-xvfb"
+    PIP_DEPENDENCIES: "coveralls coloredlogs pytest-qt pytest-xvfb flaky"
 
 dependencies:
   pre:

spyder_terminal/server/main.py

@@ -26,6 +26,8 @@
 
 LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
               '-35s %(lineno) -5d: %(message)s')
+
+logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
 LOGGER = logging.getLogger(__name__)
 coloredlogs.install(level='info')
 
@@ -34,19 +36,24 @@
     clr = 'cls'
 
 
-def main(port, shell):
-    """Create and setup a new tornado server."""
-    logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
+def create_app(shell, close_future=None):
+    """Create and return a tornado Web Application instance."""
     settings = {"static_path": os.path.join(
         os.path.dirname(__file__), "static")}
-    application = tornado.web.Application(routes.ROUTES,
+    application = tornado.web.Application(routes.gen_routes(close_future),
                                           debug=True,
                                           serve_traceback=True,
                                           autoreload=True, **settings)
-    LOGGER.info("Server is now at: 127.0.0.1:{}".format(port))
-    LOGGER.info('Shell: {0}'.format(shell))
     application.term_manager = term_manager.TermManager(shell)
     application.logger = LOGGER
+    return application
+
+
+def main(port, shell):
+    """Create and setup a new tornado server."""
+    LOGGER.info("Server is now at: 127.0.0.1:{}".format(port))
+    LOGGER.info('Shell: {0}'.format(shell))
+    application = create_app(shell)
     ioloop = tornado.ioloop.IOLoop.instance()
     application.listen(port, address='127.0.0.1')
     try:

spyder_terminal/server/rest/term_rest.py

@@ -9,15 +9,6 @@
 class MainHandler(tornado.web.RequestHandler):
     """Handles creation of new terminals."""
 
-    def initialize(self, db=None):
-        """Stump initialization function."""
-        self.db = db
-
-    @tornado.gen.coroutine
-    def get(self):
-        """GET verb."""
-        self.status_code(403)
-
     @tornado.gen.coroutine
     def post(self):
         """POST verb: Create a new terminal."""
@@ -30,15 +21,6 @@ def post(self):
 class ResizeHandler(tornado.web.RequestHandler):
     """Handles resizing of terminals."""
 
-    def initialize(self, db=None):
-        """Stump initialization function."""
-        self.db = db
-
-    @tornado.gen.coroutine
-    def get(self):
-        """GET verb: Forbidden."""
-        self.status_code(403)
-
     @tornado.gen.coroutine
     def post(self, pid):
         """POST verb: Resize a terminal."""

spyder_terminal/server/routes.py

@@ -36,3 +36,14 @@
 ]
 
 ROUTES = REST + WS + WEB
+
+
+def gen_routes(close_future):
+    """Return a list of HTML redirection routes."""
+    if close_future is not None:
+        ws = []
+        for route in WS:
+            ws.append((route[0], route[1],
+                       dict(close_future=close_future)))
+        return REST + ws + WEB
+    return ROUTES

spyder_terminal/server/tests/__init__.py

@@ -0,0 +1,16 @@
+"""
+tests module.
+
+=========
+
+Provides:
+    1. Websocket and HTTP server methods tests.
+
+How to use the documentation
+----------------------------
+Documentation is available in one form: docstrings provided
+with the code
+
+Copyright (c) 2016, Edgar A. Margffoy.
+MIT, see LICENSE for more details.
+"""

spyder_terminal/server/tests/print_size.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python
+
+"""
+Print console size on UNIX and Windows systems.
+
+Taken from: https://gist.github.com/jtriley/1108174
+"""
+
+import os
+import shlex
+import struct
+import platform
+import subprocess
+
+
+def get_terminal_size():
+    """
+    Get width and height of console.
+
+    Works on Linux, OS X, Windows and Cygwin
+    Based on:
+    http://stackoverflow.com/questions/566746/how-to-get-console-window-width-in-python
+    """
+    current_os = platform.system()
+    tuple_xy = None
+    if current_os == 'Windows':
+        tuple_xy = _get_terminal_size_windows()
+        if tuple_xy is None:
+            tuple_xy = _get_terminal_size_tput()
+            # needed for window's python in cygwin's xterm!
+    if current_os in ['Linux', 'Darwin'] or current_os.startswith('CYGWIN'):
+        tuple_xy = _get_terminal_size_linux()
+    if tuple_xy is None:
+        print("default")
+        tuple_xy = (80, 25)      # default value
+    return tuple_xy
+
+
+def _get_terminal_size_windows():
+    try:
+        from ctypes import windll, create_string_buffer
+        # stdin handle is -10
+        # stdout handle is -11
+        # stderr handle is -12
+        h = windll.kernel32.GetStdHandle(-12)
+        csbi = create_string_buffer(22)
+        res = windll.kernel32.GetConsoleScreenBufferInfo(h, csbi)
+        if res:
+            (bufx, bufy, curx, cury, wattr,
+             left, top, right, bottom,
+             maxx, maxy) = struct.unpack("hhhhHhhhhhh", csbi.raw)
+            sizex = right - left + 1
+            sizey = bottom - top + 1
+            return sizex, sizey
+    except:
+        pass
+
+
+def _get_terminal_size_tput():
+    """Get terminal width.
+
+    src: http://stackoverflow.com/questions/263890/
+         how-do-i-find-the-width-height-of-a-terminal-window
+    """
+    try:
+        cols = int(subprocess.check_call(shlex.split('tput cols')))
+        rows = int(subprocess.check_call(shlex.split('tput lines')))
+        return (cols, rows)
+    except:
+        pass
+
+
+def _get_terminal_size_linux():
+    def ioctl_GWINSZ(fd):
+        try:
+            import fcntl
+            import termios
+            cr = struct.unpack('hh',
+                               fcntl.ioctl(fd, termios.TIOCGWINSZ, '1234'))
+            return cr
+        except:
+            pass
+    cr = ioctl_GWINSZ(0) or ioctl_GWINSZ(1) or ioctl_GWINSZ(2)
+    if not cr:
+        try:
+            fd = os.open(os.ctermid(), os.O_RDONLY)
+            cr = ioctl_GWINSZ(fd)
+            os.close(fd)
+        except:
+            pass
+    if not cr:
+        try:
+            cr = (os.environ['LINES'], os.environ['COLUMNS'])
+        except:
+            return None
+    return int(cr[1]), int(cr[0])
+
+
+if __name__ == "__main__":
+    print(get_terminal_size())

spyder_terminal/server/tests/test_server.py

@@ -0,0 +1,171 @@
+
+"""
+Tornado server-side tests.
+
+Note: This uses tornado.testing unittest style tests
+"""
+
+import os
+import sys
+import os.path as osp
+
+PY3 = sys.version_info[0] == 3
+
+if PY3:
+    from urllib.parse import urlencode
+else:
+    from urllib import urlencode
+
+import pytest
+from flaky import flaky
+from tornado import testing, websocket, gen
+from tornado.concurrent import Future
+from spyder.utils.programs import find_program
+
+sys.path.append(osp.realpath(osp.dirname(__file__) + "/.."))
+
+from main import create_app
+
+LOCATION = os.path.realpath(os.path.join(os.getcwd(),
+                                         os.path.dirname(__file__)))
+LOCATION_SLASH = LOCATION.replace('\\', '/')
+
+LINE_END = '\n'
+SHELL = '/usr/bin/env bash'
+WINDOWS = os.name == 'nt'
+
+if WINDOWS:
+    LINE_END = '\r\n'
+    SHELL = find_program('cmd.exe')
+
+
+class TerminalServerTests(testing.AsyncHTTPTestCase):
+    """Main server tests."""
+
+    def get_app(self):
+        """Return HTTP/WS server."""
+        self.close_future = Future()
+        return create_app(SHELL, self.close_future)
+
+    def _mk_connection(self, pid):
+        return websocket.websocket_connect(
+            'ws://127.0.0.1:{0}/terminals/{1}'.format(
+                self.get_http_port(), pid)
+        )
+
+    @gen.coroutine
+    def close(self, ws):
+        """
+        Close a websocket connection and wait for the server side.
+
+        If we don't wait here, there are sometimes leak warnings in the
+        tests.
+        """
+        ws.close()
+        yield self.close_future
+
+    @testing.gen_test
+    def test_main_get(self):
+        """Test if HTML source is rendered."""
+        response = yield self.http_client.fetch(
+            self.get_url('/'),
+            method="GET"
+        )
+        self.assertEqual(response.code, 200)
+
+    @testing.gen_test
+    def test_main_post(self):
+        """Test that POST requests to root are forbidden."""
+        try:
+            yield self.http_client.fetch(
+                self.get_url('/'),
+                method="POST",
+                body=''
+            )
+        except Exception:
+            pass
+
+    @testing.gen_test
+    def test_create_terminal(self):
+        """Test terminal creation."""
+        data = {'rows': '25', 'cols': '80'}
+        response = yield self.http_client.fetch(
+            self.get_url('/api/terminals'),
+            method="POST",
+            body=urlencode(data)
+        )
+        self.assertEqual(response.code, 200)
+
+    @flaky(max_runs=3)
+    @testing.gen_test
+    def test_terminal_communication(self):
+        """Test terminal creation."""
+        data = {'rows': '25', 'cols': '100'}
+        response = yield self.http_client.fetch(
+            self.get_url('/api/terminals'),
+            method="POST",
+            body=urlencode(data)
+        )
+        pid = response.body.decode('utf-8')
+        sock = yield self._mk_connection(pid)
+        msg = yield sock.read_message()
+        print(msg)
+        test_msg = 'pwd'
+        sock.write_message(' ' + test_msg)
+        msg = ''
+        while test_msg not in msg:
+            msg = yield sock.read_message()
+            print(msg)
+        self.assertTrue(test_msg in msg)
+
+    # @pytest.mark.skipif(os.name == 'nt', reason="It doesn't work on Windows")
+    @testing.gen_test
+    def test_terminal_closing(self):
+        """Test terminal destruction."""
+        data = {'rows': '25', 'cols': '80'}
+        response = yield self.http_client.fetch(
+            self.get_url('/api/terminals'),
+            method="POST",
+            body=urlencode(data)
+        )
+        pid = response.body.decode('utf-8')
+        sock = yield self._mk_connection(pid)
+        _ = yield sock.read_message()
+        yield self.close(sock)
+        try:
+            sock.write_message(' This shall not work')
+        except AttributeError:
+            pass
+
+    @flaky(max_runs=3)
+    @testing.gen_test
+    def test_terminal_resize(self):
+        """Test terminal resizing."""
+        data = {'rows': '25', 'cols': '80'}
+        response = yield self.http_client.fetch(
+            self.get_url('/api/terminals'),
+            method="POST",
+            body=urlencode(data)
+        )
+
+        pid = response.body.decode('utf-8')
+        sock = yield self._mk_connection(pid)
+        _ = yield sock.read_message()
+
+        data = {'rows': '23', 'cols': '73'}
+        response = yield self.http_client.fetch(
+            self.get_url('/api/terminals/{0}/size'.format(pid)),
+            method="POST",
+            body=urlencode(data)
+        )
+
+        sock.write_message('cd {0}{1}'.format(LOCATION_SLASH, LINE_END))
+
+        python_exec = 'python print_size.py' + LINE_END
+        sock.write_message(python_exec)
+
+        expected_size = '(73, 23)'
+        msg = ''
+        while expected_size not in msg:
+            msg = yield sock.read_message()
+        self.assertIn(expected_size, msg)