Add support for creating terminals via GET

notebook/terminal/__init__.py

@@ -10,7 +10,7 @@
 from ipython_genutils.py3compat import which
 from notebook.utils import url_path_join as ujoin
 from .terminalmanager import TerminalManager
-from .handlers import TerminalHandler, TermSocket
+from .handlers import TerminalHandler, TermSocket, NewTerminalHandler
 from . import api_handlers
 
 
@@ -45,6 +45,7 @@ def initialize(nb_app):
         (ujoin(base_url, r"/terminals/(\w+)"), TerminalHandler),
         (ujoin(base_url, r"/terminals/websocket/(\w+)"), TermSocket,
              {'term_manager': terminal_manager}),
+        (ujoin(base_url, r"/terminals/new/(\w+)"), NewTerminalHandler),
         (ujoin(base_url, r"/api/terminals"), api_handlers.TerminalRootHandler),
         (ujoin(base_url, r"/api/terminals/(\w+)"), api_handlers.TerminalHandler),
     ]

notebook/terminal/handlers.py

@@ -3,6 +3,7 @@
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
 
+import json
 from tornado import web
 import terminado
 from notebook._tz import utcnow
@@ -14,8 +15,31 @@ class TerminalHandler(IPythonHandler):
     """Render the terminal interface."""
     @web.authenticated
     def get(self, term_name):
-        self.write(self.render_template('terminal.html',
-                   ws_path="terminals/websocket/%s" % term_name))
+        if term_name == 'new':
+            model = self.terminal_manager.create()
+            term_name = model['name']
+            new_path = self.request.path.replace("terminals/new", "terminals/" + term_name)
+            self.redirect(new_path)
+        else:
+            self.write(self.render_template('terminal.html',
+                       ws_path="terminals/websocket/%s" % term_name))
+
+
+class NewTerminalHandler(IPythonHandler):
+    """Renders a new terminal interface using the named argument."""
+    @web.authenticated
+    def get(self, term_name):
+        if term_name == 'new':
+            raise web.HTTPError(400, "Terminal name 'new' is reserved.")
+        new_path = self.request.path.replace("new/{}".format(term_name), term_name)
+        if term_name in self.terminal_manager.terminals:
+            self.set_header('Location', new_path)
+            self.set_status(302)
+            self.finish(json.dumps(self.terminal_manager.get_terminal_model(term_name)))
+            return
+
+        self.terminal_manager.create_with_name(term_name)
+        self.redirect(new_path)
 
 
 class TermSocket(WebSocketMixin, IPythonHandler, terminado.TermSocket):

notebook/terminal/terminalmanager.py

@@ -44,6 +44,16 @@ def __init__(self, *args, **kwargs):
     def create(self):
         """Create a new terminal."""
         name, term = self.new_named_terminal()
+        return self._finish_create(name, term)
+
+    def create_with_name(self, name):
+        """Create a new terminal."""
+        if name in self.terminals:
+            raise web.HTTPError(409, "A terminal with name '{}' already exists.".format(name))
+        term = self.get_terminal(name)
+        return self._finish_create(name, term)
+
+    def _finish_create(self, name, term):
         # Monkey-patch last-activity, similar to kernels.  Should we need
         # more functionality per terminal, we can look into possible sub-
         # classing or containment then.

notebook/terminal/tests/test_terminals_api.py

@@ -54,7 +54,7 @@ def tearDown(self):
             self.term_api.shutdown(k['name'])
 
     def test_no_terminals(self):
-        # Make sure there are no terminals running at the start
+        # Make sure there are no terminals are running at the start
         terminals = self.term_api.list().json()
         self.assertEqual(terminals, [])
 
@@ -65,6 +65,74 @@ def test_create_terminal(self):
         self.assertEqual(r.status_code, 200)
         self.assertIsInstance(term1, dict)
 
+    def test_create_terminal_via_get(self):
+        # Test creation of terminal via GET against terminals/new/<name>
+        r = self.term_api._req('GET', 'terminals/new')
+        self.assertEqual(r.status_code, 200)
+
+        r = self.term_api.get('1')
+        term1 = r.json()
+        self.assertEqual(r.status_code, 200)
+        self.assertIsInstance(term1, dict)
+        self.assertEqual(term1['name'], '1')
+
+        # hit the same endpoint a second time and ensure a second named terminal is created
+        r = self.term_api._req('GET', 'terminals/new')
+        self.assertEqual(r.status_code, 200)
+
+        r = self.term_api.get('2')
+        term2 = r.json()
+        self.assertEqual(r.status_code, 200)
+        self.assertIsInstance(term2, dict)
+        self.assertEqual(term2['name'], '2')
+
+        r = self.term_api.shutdown('2')
+        self.assertEqual(r.status_code, 204)
+
+        # Make sure there is 1 terminal running
+        terminals = self.term_api.list().json()
+        self.assertEqual(len(terminals), 1)
+
+        r = self.term_api.shutdown('1')
+        self.assertEqual(r.status_code, 204)
+
+        # Make sure there are no terminals are running
+        terminals = self.term_api.list().json()
+        self.assertEqual(len(terminals), 0)
+
+    def test_create_terminal_with_name(self):
+        # Test creation of terminal via GET against terminals/new/<name>
+        r = self.term_api._req('GET', 'terminals/new/foo')
+        self.assertEqual(r.status_code, 200)
+
+        r = self.term_api.get('foo')
+        foo_term = r.json()
+        self.assertEqual(r.status_code, 200)
+        self.assertIsInstance(foo_term, dict)
+        self.assertEqual(foo_term['name'], 'foo')
+
+        # hit the same endpoint a second time and ensure 302 with Location is returned
+        r = self.term_api._req('GET', 'terminals/new/foo')
+        # Access the "interesting" response from the history
+        self.assertEqual(len(r.history), 1)
+        r = r.history[0]
+        foo_term = r.json()
+        self.assertEqual(r.status_code, 302)
+        self.assertEqual(r.headers['Location'], self.url_prefix + "terminals/foo")
+        self.assertIsInstance(foo_term, dict)
+        self.assertEqual(foo_term['name'], 'foo')
+
+        r = self.term_api.shutdown('foo')
+        self.assertEqual(r.status_code, 204)
+
+        # Make sure there are no terminals are running
+        terminals = self.term_api.list().json()
+        self.assertEqual(len(terminals), 0)
+
+        # hit terminals/new/new and ensure that 400 is raised
+        with assert_http_error(400):
+            self.term_api._req('GET', 'terminals/new/new')
+
     def test_terminal_root_handler(self):
         # POST request
         r = self.term_api.start()