Add tests and fix some issues

Tests taken from #6360
Thomas Kluyver 12 years ago
parent 0e109ee4be
commit b2737e668e

@ -189,7 +189,10 @@ class SessionManager(LoggingConfigurable):
"""Takes sqlite database session row and turns it into a dictionary"""
if row['kernel_id'] not in self.kernel_manager:
# The kernel was killed or died without deleting the session.
self.delete_session(row['session_id'])
# We can't use delete_session here because that tries to find
# and shut down the kernel.
self.cursor.execute("DELETE FROM session WHERE session_id=?",
(row['session_id'],))
raise KeyError
model = {
@ -207,7 +210,9 @@ class SessionManager(LoggingConfigurable):
the session database"""
c = self.cursor.execute("SELECT * FROM session")
result = []
for row in c:
# We need to use fetchall() here, because row_to_model can delete rows,
# which messes up the cursor if we're iterating over rows.
for row in c.fetchall():
try:
result.append(self.row_to_model(row))
except KeyError:

@ -47,6 +47,17 @@ class TestSessionManager(TestCase):
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 = [
@ -63,6 +74,30 @@ class TestSessionManager(TestCase):
'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/',

Loading…
Cancel
Save