test first element of sys.path in Kernel

both when `--profile-dir` is specified, and when it is not.

IPython/kernel/tests/test_kernel.py

@@ -1,170 +1,186 b''
 """test the IPython Kernel"""
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2013  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import sys
 
 import nose.tools as nt
 
 from IPython.testing import decorators as dec, tools as tt
 from IPython.utils import py3compat
+from IPython.utils.path import locate_profile
 
 from .utils import new_kernel, kernel, TIMEOUT, assemble_output, execute, flush_channels
 
 #-------------------------------------------------------------------------------
 # Tests
 #-------------------------------------------------------------------------------
 
 
 def _check_mp_mode(kc, expected=False, stream="stdout"):
     execute(kc=kc, code="import sys")
     flush_channels(kc)
     msg_id, content = execute(kc=kc, code="print (sys.%s._check_mp_mode())" % stream)
     stdout, stderr = assemble_output(kc.iopub_channel)
     nt.assert_equal(eval(stdout.strip()), expected)
 
 
 # printing tests
 
 def test_simple_print():
     """simple print statement in kernel"""
     with kernel() as kc:
         iopub = kc.iopub_channel
         msg_id, content = execute(kc=kc, code="print ('hi')")
         stdout, stderr = assemble_output(iopub)
         nt.assert_equal(stdout, 'hi\n')
         nt.assert_equal(stderr, '')
         _check_mp_mode(kc, expected=False)
 
 
+def test_sys_path():
+    """test that sys.path doesn't get messed up by default"""
+    with kernel() as kc:
+        msg_id, content = execute(kc=kc, code="import sys; print (repr(sys.path[0]))")
+        stdout, stderr = assemble_output(kc.iopub_channel)
+        nt.assert_equal(stdout, "''\n")
+
+def test_sys_path_profile_dir():
+    """test that sys.path doesn't get messed up when `--profile-dir` is specified"""
+    
+    with new_kernel(['--profile-dir', locate_profile('default')]) as kc:
+        msg_id, content = execute(kc=kc, code="import sys; print (repr(sys.path[0]))")
+        stdout, stderr = assemble_output(kc.iopub_channel)
+        nt.assert_equal(stdout, "''\n")
+
 @dec.knownfailureif(sys.platform == 'win32', "subprocess prints fail on Windows")
 def test_subprocess_print():
     """printing from forked mp.Process"""
     with new_kernel() as kc:
         iopub = kc.iopub_channel
         
         _check_mp_mode(kc, expected=False)
         flush_channels(kc)
         np = 5
         code = '\n'.join([
             "from __future__ import print_function",
             "import multiprocessing as mp",
             "pool = [mp.Process(target=print, args=('hello', i,)) for i in range(%i)]" % np,
             "for p in pool: p.start()",
             "for p in pool: p.join()"
         ])
         
         expected = '\n'.join([
             "hello %s" % i for i in range(np)
         ]) + '\n'
         
         msg_id, content = execute(kc=kc, code=code)
         stdout, stderr = assemble_output(iopub)
         nt.assert_equal(stdout.count("hello"), np, stdout)
         for n in range(np):
             nt.assert_equal(stdout.count(str(n)), 1, stdout)
         nt.assert_equal(stderr, '')
         _check_mp_mode(kc, expected=False)
         _check_mp_mode(kc, expected=False, stream="stderr")
 
 
 def test_subprocess_noprint():
     """mp.Process without print doesn't trigger iostream mp_mode"""
     with kernel() as kc:
         iopub = kc.iopub_channel
         
         np = 5
         code = '\n'.join([
             "import multiprocessing as mp",
             "pool = [mp.Process(target=range, args=(i,)) for i in range(%i)]" % np,
             "for p in pool: p.start()",
             "for p in pool: p.join()"
         ])
         
         msg_id, content = execute(kc=kc, code=code)
         stdout, stderr = assemble_output(iopub)
         nt.assert_equal(stdout, '')
         nt.assert_equal(stderr, '')
 
         _check_mp_mode(kc, expected=False)
         _check_mp_mode(kc, expected=False, stream="stderr")
 
 
 @dec.knownfailureif(sys.platform == 'win32', "subprocess prints fail on Windows")
 def test_subprocess_error():
     """error in mp.Process doesn't crash"""
     with new_kernel() as kc:
         iopub = kc.iopub_channel
         
         code = '\n'.join([
             "import multiprocessing as mp",
             "p = mp.Process(target=int, args=('hi',))",
             "p.start()",
             "p.join()",
         ])
         
         msg_id, content = execute(kc=kc, code=code)
         stdout, stderr = assemble_output(iopub)
         nt.assert_equal(stdout, '')
         nt.assert_true("ValueError" in stderr, stderr)
 
         _check_mp_mode(kc, expected=False)
         _check_mp_mode(kc, expected=False, stream="stderr")
 
 # raw_input tests
 
 def test_raw_input():
     """test [raw_]input"""
     with kernel() as kc:
         iopub = kc.iopub_channel
         
         input_f = "input" if py3compat.PY3 else "raw_input"
         theprompt = "prompt> "
         code = 'print({input_f}("{theprompt}"))'.format(**locals())
         msg_id = kc.execute(code, allow_stdin=True)
         msg = kc.get_stdin_msg(block=True, timeout=TIMEOUT)
         nt.assert_equal(msg['header']['msg_type'], u'input_request')
         content = msg['content']
         nt.assert_equal(content['prompt'], theprompt)
         text = "some text"
         kc.input(text)
         reply = kc.get_shell_msg(block=True, timeout=TIMEOUT)
         nt.assert_equal(reply['content']['status'], 'ok')
         stdout, stderr = assemble_output(iopub)
         nt.assert_equal(stdout, text + "\n")
 
 
 @dec.skipif(py3compat.PY3)
 def test_eval_input():
     """test input() on Python 2"""
     with kernel() as kc:
         iopub = kc.iopub_channel
         
         input_f = "input" if py3compat.PY3 else "raw_input"
         theprompt = "prompt> "
         code = 'print(input("{theprompt}"))'.format(**locals())
         msg_id = kc.execute(code, allow_stdin=True)
         msg = kc.get_stdin_msg(block=True, timeout=TIMEOUT)
         nt.assert_equal(msg['header']['msg_type'], u'input_request')
         content = msg['content']
         nt.assert_equal(content['prompt'], theprompt)
         kc.input("1+1")
         reply = kc.get_shell_msg(block=True, timeout=TIMEOUT)
         nt.assert_equal(reply['content']['status'], 'ok')
         stdout, stderr = assemble_output(iopub)
         nt.assert_equal(stdout, "2\n")
 
 
 def test_help_output():
     """ipython kernel --help-all works"""
     tt.help_all_output_test('kernel')
 

IPython/kernel/tests/utils.py

@@ -1,167 +1,170 b''
 """utilities for testing IPython kernels"""
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2013  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import atexit
 
 from contextlib import contextmanager
 from subprocess import PIPE
 from Queue import Empty
 
 import nose.tools as nt
 
 from IPython.kernel import KernelManager
 
 #-------------------------------------------------------------------------------
 # Globals
 #-------------------------------------------------------------------------------
 
 STARTUP_TIMEOUT = 60
 TIMEOUT = 15
 
 KM = None
 KC = None
 
 #-------------------------------------------------------------------------------
 # code
 #-------------------------------------------------------------------------------
 
 
-def start_new_kernel():
+def start_new_kernel(argv=None):
     """start a new kernel, and return its Manager and Client"""
     km = KernelManager()
-    km.start_kernel(stdout=PIPE, stderr=PIPE)
+    kwargs = dict(stdout=PIPE, stderr=PIPE)
+    if argv:
+        kwargs['extra_arguments'] = argv
+    km.start_kernel(**kwargs)
     kc = km.client()
     kc.start_channels()
     
     msg_id = kc.kernel_info()
     kc.get_shell_msg(block=True, timeout=STARTUP_TIMEOUT)
     flush_channels(kc)
     return km, kc
 
 def flush_channels(kc=None):
     """flush any messages waiting on the queue"""
     from .test_message_spec import validate_message
     
     if kc is None:
         kc = KC
     for channel in (kc.shell_channel, kc.iopub_channel):
         while True:
             try:
                 msg = channel.get_msg(block=True, timeout=0.1)
             except Empty:
                 break
             else:
                 validate_message(msg)
 
 
 def execute(code='', kc=None, **kwargs):
     """wrapper for doing common steps for validating an execution request"""
     from .test_message_spec import validate_message
     if kc is None:
         kc = KC
     msg_id = kc.execute(code=code, **kwargs)
     reply = kc.get_shell_msg(timeout=TIMEOUT)
     validate_message(reply, 'execute_reply', msg_id)
     busy = kc.get_iopub_msg(timeout=TIMEOUT)
     validate_message(busy, 'status', msg_id)
     nt.assert_equal(busy['content']['execution_state'], 'busy')
     
     if not kwargs.get('silent'):
         pyin = kc.get_iopub_msg(timeout=TIMEOUT)
         validate_message(pyin, 'pyin', msg_id)
         nt.assert_equal(pyin['content']['code'], code)
     
     return msg_id, reply['content']
 
 def start_global_kernel():
     """start the global kernel (if it isn't running) and return its client"""
     global KM, KC
     if KM is None:
         KM, KC = start_new_kernel()
         atexit.register(stop_global_kernel)
     return KC
 
 @contextmanager
 def kernel():
     """Context manager for the global kernel instance
     
     Should be used for most kernel tests
     
     Returns
     -------
     kernel_client: connected KernelClient instance
     """
     yield start_global_kernel()
 
 def uses_kernel(test_f):
     """Decorator for tests that use the global kernel"""
     def wrapped_test():
         with kernel() as kc:
             test_f(kc)
     wrapped_test.__doc__ = test_f.__doc__
     wrapped_test.__name__ = test_f.__name__
     return wrapped_test
 
 def stop_global_kernel():
     """Stop the global shared kernel instance, if it exists"""
     global KM, KC
     KC.stop_channels()
     KC = None
     if KM is None:
         return
     KM.shutdown_kernel(now=True)
     KM = None
 
 @contextmanager
-def new_kernel():
+def new_kernel(argv=None):
     """Context manager for a new kernel in a subprocess
     
     Should only be used for tests where the kernel must not be re-used.
     
     Returns
     -------
     kernel_client: connected KernelClient instance
     """
-    km, kc = start_new_kernel()
+    km, kc = start_new_kernel(argv)
     try:
         yield kc
     finally:
         kc.stop_channels()
         km.shutdown_kernel(now=True)
 
 
 def assemble_output(iopub):
     """assemble stdout/err from an execution"""
     stdout = ''
     stderr = ''
     while True:
         msg = iopub.get_msg(block=True, timeout=1)
         msg_type = msg['msg_type']
         content = msg['content']
         if msg_type == 'status' and content['execution_state'] == 'idle':
             # idle message signals end of output
             break
         elif msg['msg_type'] == 'stream':
             if content['name'] == 'stdout':
                 stdout += content['data']
             elif content['name'] == 'stderr':
                 stderr += content['data']
             else:
                 raise KeyError("bad stream: %r" % content['name'])
         else:
             # other output, ignored
             pass
     return stdout, stderr