|
|
"""Tests for the session manager."""
|
|
|
|
|
|
from unittest import TestCase
|
|
|
|
|
|
from tornado import web
|
|
|
|
|
|
from ..sessionmanager import SessionManager
|
|
|
from IPython.html.services.kernels.kernelmanager import MappingKernelManager
|
|
|
|
|
|
class DummyKernel(object):
|
|
|
def __init__(self, kernel_name='python'):
|
|
|
self.kernel_name = kernel_name
|
|
|
|
|
|
class DummyMKM(MappingKernelManager):
|
|
|
"""MappingKernelManager interface that doesn't start kernels, for testing"""
|
|
|
def __init__(self, *args, **kwargs):
|
|
|
super(DummyMKM, self).__init__(*args, **kwargs)
|
|
|
self.id_letters = iter(u'ABCDEFGHIJK')
|
|
|
|
|
|
def _new_id(self):
|
|
|
return next(self.id_letters)
|
|
|
|
|
|
def start_kernel(self, kernel_id=None, path=None, kernel_name='python', **kwargs):
|
|
|
kernel_id = kernel_id or self._new_id()
|
|
|
self._kernels[kernel_id] = DummyKernel(kernel_name=kernel_name)
|
|
|
return kernel_id
|
|
|
|
|
|
def shutdown_kernel(self, kernel_id, now=False):
|
|
|
del self._kernels[kernel_id]
|
|
|
|
|
|
class TestSessionManager(TestCase):
|
|
|
|
|
|
def test_get_session(self):
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
session_id = sm.create_session(name='test.ipynb', path='/path/to/',
|
|
|
kernel_name='bar')['id']
|
|
|
model = sm.get_session(session_id=session_id)
|
|
|
expected = {'id':session_id,
|
|
|
'notebook':{'name':u'test.ipynb', 'path': u'/path/to/'},
|
|
|
'kernel': {'id':u'A', 'name': 'bar'}}
|
|
|
self.assertEqual(model, expected)
|
|
|
|
|
|
def test_bad_get_session(self):
|
|
|
# Should raise error if a bad key is passed to the database.
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
session_id = sm.create_session(name='test.ipynb', path='/path/to/',
|
|
|
kernel_name='foo')['id']
|
|
|
self.assertRaises(TypeError, sm.get_session, bad_id=session_id) # Bad keyword
|
|
|
|
|
|
def test_get_session_dead_kernel(self):
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
session = sm.create_session(name='test1.ipynb', path='/path/to/1/', kernel_name='python')
|
|
|
# kill the kernel
|
|
|
sm.kernel_manager.shutdown_kernel(session['kernel']['id'])
|
|
|
with self.assertRaises(KeyError):
|
|
|
sm.get_session(session_id=session['id'])
|
|
|
# no sessions left
|
|
|
listed = sm.list_sessions()
|
|
|
self.assertEqual(listed, [])
|
|
|
|
|
|
def test_list_sessions(self):
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
sessions = [
|
|
|
sm.create_session(name='test1.ipynb', path='/path/to/1/', kernel_name='python'),
|
|
|
sm.create_session(name='test2.ipynb', path='/path/to/2/', kernel_name='python'),
|
|
|
sm.create_session(name='test3.ipynb', path='/path/to/3/', kernel_name='python'),
|
|
|
]
|
|
|
sessions = sm.list_sessions()
|
|
|
expected = [{'id':sessions[0]['id'], 'notebook':{'name':u'test1.ipynb',
|
|
|
'path': u'/path/to/1/'}, 'kernel':{'id':u'A', 'name':'python'}},
|
|
|
{'id':sessions[1]['id'], 'notebook': {'name':u'test2.ipynb',
|
|
|
'path': u'/path/to/2/'}, 'kernel':{'id':u'B', 'name':'python'}},
|
|
|
{'id':sessions[2]['id'], 'notebook':{'name':u'test3.ipynb',
|
|
|
'path': u'/path/to/3/'}, 'kernel':{'id':u'C', 'name':'python'}}]
|
|
|
self.assertEqual(sessions, expected)
|
|
|
|
|
|
def test_list_sessions_dead_kernel(self):
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
sessions = [
|
|
|
sm.create_session(name='test1.ipynb', path='/path/to/1/', kernel_name='python'),
|
|
|
sm.create_session(name='test2.ipynb', path='/path/to/2/', kernel_name='python'),
|
|
|
]
|
|
|
# kill one of the kernels
|
|
|
sm.kernel_manager.shutdown_kernel(sessions[0]['kernel']['id'])
|
|
|
listed = sm.list_sessions()
|
|
|
expected = [
|
|
|
{
|
|
|
'id': sessions[1]['id'],
|
|
|
'notebook': {
|
|
|
'name': u'test2.ipynb',
|
|
|
'path': u'/path/to/2/',
|
|
|
},
|
|
|
'kernel': {
|
|
|
'id': u'B',
|
|
|
'name':'python',
|
|
|
}
|
|
|
}
|
|
|
]
|
|
|
self.assertEqual(listed, expected)
|
|
|
|
|
|
def test_update_session(self):
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
session_id = sm.create_session(name='test.ipynb', path='/path/to/',
|
|
|
kernel_name='julia')['id']
|
|
|
sm.update_session(session_id, name='new_name.ipynb')
|
|
|
model = sm.get_session(session_id=session_id)
|
|
|
expected = {'id':session_id,
|
|
|
'notebook':{'name':u'new_name.ipynb', 'path': u'/path/to/'},
|
|
|
'kernel':{'id':u'A', 'name':'julia'}}
|
|
|
self.assertEqual(model, expected)
|
|
|
|
|
|
def test_bad_update_session(self):
|
|
|
# try to update a session with a bad keyword ~ raise error
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
session_id = sm.create_session(name='test.ipynb', path='/path/to/',
|
|
|
kernel_name='ir')['id']
|
|
|
self.assertRaises(TypeError, sm.update_session, session_id=session_id, bad_kw='test.ipynb') # Bad keyword
|
|
|
|
|
|
def test_delete_session(self):
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
sessions = [
|
|
|
sm.create_session(name='test1.ipynb', path='/path/to/1/', kernel_name='python'),
|
|
|
sm.create_session(name='test2.ipynb', path='/path/to/2/', kernel_name='python'),
|
|
|
sm.create_session(name='test3.ipynb', path='/path/to/3/', kernel_name='python'),
|
|
|
]
|
|
|
sm.delete_session(sessions[1]['id'])
|
|
|
new_sessions = sm.list_sessions()
|
|
|
expected = [{'id':sessions[0]['id'], 'notebook':{'name':u'test1.ipynb',
|
|
|
'path': u'/path/to/1/'}, 'kernel':{'id':u'A', 'name':'python'}},
|
|
|
{'id':sessions[2]['id'], 'notebook':{'name':u'test3.ipynb',
|
|
|
'path': u'/path/to/3/'}, 'kernel':{'id':u'C', 'name':'python'}}]
|
|
|
self.assertEqual(new_sessions, expected)
|
|
|
|
|
|
def test_bad_delete_session(self):
|
|
|
# try to delete a session that doesn't exist ~ raise error
|
|
|
sm = SessionManager(kernel_manager=DummyMKM())
|
|
|
sm.create_session(name='test.ipynb', path='/path/to/', kernel_name='python')
|
|
|
self.assertRaises(TypeError, sm.delete_session, bad_kwarg='23424') # Bad keyword
|
|
|
self.assertRaises(web.HTTPError, sm.delete_session, session_id='23424') # nonexistant
|
|
|
|
|
|
|
