Add test of 'jupyter kernel'

Author: takluyver

jupyter_client/kernelapp.py

@@ -35,6 +35,7 @@ def initialize(self, argv=None):
         cf_basename = 'kernel-%s.json' % uuid.uuid4()
         self.km.connection_file = os.path.join(self.runtime_dir, cf_basename)
         self.loop = IOLoop.current()
+        self.loop.add_callback(self._record_started)
 
     def setup_signals(self):
         """Shutdown on SIGTERM or SIGINT (Ctrl-C)"""
@@ -56,6 +57,16 @@ def log_connection_info(self):
         self.log.info('Connection file: %s', cf)
         self.log.info("To connect a client: --existing %s", os.path.basename(cf))
 
+    def _record_started(self):
+        """For tests, create a file to indicate that we've started
+
+        Do not rely on this except in our own tests!
+        """
+        fn = os.environ.get('JUPYTER_CLIENT_TEST_RECORD_STARTUP_PRIVATE')
+        if fn is not None:
+            with open(fn, 'wb'):
+                pass
+
     def start(self):
         self.log.info('Starting kernel %r', self.kernel_name)
         try:

jupyter_client/tests/test_kernelapp.py

@@ -0,0 +1,57 @@
+from __future__ import division
+
+import os
+import shutil
+from subprocess import Popen, PIPE
+import sys
+from tempfile import mkdtemp
+import time
+
+def _launch(extra_env):
+    env = os.environ.copy()
+    env.update(extra_env)
+    return Popen([sys.executable, '-c',
+                  'from jupyter_client.kernelapp import main; main()'],
+                 env=env, stderr=PIPE)
+
+WAIT_TIME = 10
+POLL_FREQ = 10
+
+def test_kernelapp_lifecycle():
+    # Check that 'jupyter kernel' starts and terminates OK.
+    runtime_dir = mkdtemp()
+    startup_dir = mkdtemp()
+    started = os.path.join(startup_dir, 'started')
+    try:
+        p = _launch({'JUPYTER_RUNTIME_DIR': runtime_dir,
+                     'JUPYTER_CLIENT_TEST_RECORD_STARTUP_PRIVATE': started,
+                    })
+        # Wait for start
+        for _ in range(WAIT_TIME * POLL_FREQ):
+            if os.path.isfile(started):
+                break
+            time.sleep(1 / POLL_FREQ)
+        else:
+            raise AssertionError("No started file created in {} seconds"
+                                 .format(WAIT_TIME))
+
+        # Connection file should be there by now
+        files = os.listdir(runtime_dir)
+        assert len(files) == 1
+        cf = files[0]
+        assert cf.startswith('kernel')
+        assert cf.endswith('.json')
+
+        # Read the first three lines from stderr. This will hang if there are
+        # fewer lines to read; I don't see any way to avoid that without lots
+        # of extra complexity.
+        b = b''.join(p.stderr.readline() for _ in range(2)).decode('utf-8', 'replace')
+        assert cf in b
+
+        # Send SIGTERM to shut down
+        p.terminate()
+        p.wait(timeout=10)
+    finally:
+        shutil.rmtree(runtime_dir)
+        shutil.rmtree(startup_dir)
+