FilesExpand file tree

 master

/

test_multikernelmanager.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

78 lines (65 loc) · 2.39 KB

 master

/

test_multikernelmanager.py

Copy path

Top

File metadata and controls

• Code
• Blame

78 lines (65 loc) · 2.39 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

"""Tests for the notebook kernel and session manager."""

from subprocess import PIPE

import time

from unittest import TestCase

from IPython.testing import decorators as dec

from IPython.config.loader import Config

from IPython.utils.localinterfaces import localhost

from IPython.kernel import KernelManager

from IPython.kernel.multikernelmanager import MultiKernelManager

class TestKernelManager(TestCase):

def _get_tcp_km(self):

c = Config()

km = MultiKernelManager(config=c)

return km

def _get_ipc_km(self):

c = Config()

c.KernelManager.transport = 'ipc'

c.KernelManager.ip = 'test'

km = MultiKernelManager(config=c)

return km

def _run_lifecycle(self, km):

kid = km.start_kernel(stdout=PIPE, stderr=PIPE)

self.assertTrue(km.is_alive(kid))

self.assertTrue(kid in km)

self.assertTrue(kid in km.list_kernel_ids())

self.assertEqual(len(km),1)

km.restart_kernel(kid, now=True)

self.assertTrue(km.is_alive(kid))

self.assertTrue(kid in km.list_kernel_ids())

km.interrupt_kernel(kid)

k = km.get_kernel(kid)

self.assertTrue(isinstance(k, KernelManager))

km.shutdown_kernel(kid, now=True)

self.assertTrue(not kid in km)

def _run_cinfo(self, km, transport, ip):

kid = km.start_kernel(stdout=PIPE, stderr=PIPE)

k = km.get_kernel(kid)

cinfo = km.get_connection_info(kid)

self.assertEqual(transport, cinfo['transport'])

self.assertEqual(ip, cinfo['ip'])

self.assertTrue('stdin_port' in cinfo)

self.assertTrue('iopub_port' in cinfo)

stream = km.connect_iopub(kid)

stream.close()

self.assertTrue('shell_port' in cinfo)

stream = km.connect_shell(kid)

stream.close()

self.assertTrue('hb_port' in cinfo)

stream = km.connect_hb(kid)

stream.close()

km.shutdown_kernel(kid, now=True)

def test_tcp_lifecycle(self):

km = self._get_tcp_km()

self._run_lifecycle(km)

def test_tcp_cinfo(self):

km = self._get_tcp_km()

self._run_cinfo(km, 'tcp', localhost())

@dec.skip_win32

def test_ipc_lifecycle(self):

km = self._get_ipc_km()

self._run_lifecycle(km)

@dec.skip_win32

def test_ipc_cinfo(self):

km = self._get_ipc_km()

self._run_cinfo(km, 'ipc', 'test')