2
2
3
3
from ipykernel .comm import Comm , CommManager
4
4
from ipykernel .ipkernel import IPythonKernel
5
+ from ipykernel .kernelbase import Kernel
5
6
6
7
7
- def test_comm (kernel ) :
8
+ def test_comm (kernel : Kernel ) -> None :
8
9
manager = CommManager (kernel = kernel )
9
- kernel .comm_manager = manager
10
+ kernel .comm_manager = manager # type:ignore
10
11
11
12
c = Comm (kernel = kernel , target_name = "bar" )
12
13
msgs = []
13
14
15
+ assert kernel is c .kernel # type:ignore
16
+
14
17
def on_close (msg ):
15
18
msgs .append (msg )
16
19
@@ -28,7 +31,7 @@ def on_message(msg):
28
31
assert c .target_name == "bar"
29
32
30
33
31
- def test_comm_manager (kernel ) :
34
+ def test_comm_manager (kernel : Kernel ) -> None :
32
35
manager = CommManager (kernel = kernel )
33
36
msgs = []
34
37
@@ -48,14 +51,19 @@ def on_msg(msg):
48
51
manager .register_target ("foo" , foo )
49
52
manager .register_target ("fizz" , fizz )
50
53
51
- kernel .comm_manager = manager
54
+ kernel .comm_manager = manager # type:ignore
52
55
with unittest .mock .patch .object (Comm , "publish_msg" ) as publish_msg :
53
56
comm = Comm ()
54
57
comm .on_msg (on_msg )
55
58
comm .on_close (on_close )
56
59
manager .register_comm (comm )
57
60
assert publish_msg .call_count == 1
58
61
62
+ # make sure that when we don't pass a kernel, the 'default' kernel is taken
63
+ Kernel ._instance = kernel # type:ignore
64
+ assert comm .kernel is kernel # type:ignore
65
+ Kernel .clear_instance ()
66
+
59
67
assert manager .get_comm (comm .comm_id ) == comm
60
68
assert manager .get_comm ('foo' ) is None
61
69
0 commit comments