|
1 |
| -from ipykernel.comm import Comm |
| 1 | +from ipykernel.comm import Comm, CommManager |
| 2 | +from ipykernel.ipkernel import IPythonKernel |
2 | 3 |
|
3 | 4 |
|
4 |
| -async def test_comm(kernel): |
5 |
| - c = Comm() |
6 |
| - c.kernel = kernel # type:ignore |
| 5 | +def test_comm(kernel): |
| 6 | + manager = CommManager(kernel=kernel) |
| 7 | + kernel.comm_manager = manager |
| 8 | + |
| 9 | + c = Comm(kernel=kernel) |
| 10 | + msgs = [] |
| 11 | + |
| 12 | + def on_close(msg): |
| 13 | + msgs.append(msg) |
| 14 | + |
| 15 | + def on_message(msg): |
| 16 | + msgs.append(msg) |
| 17 | + |
7 | 18 | c.publish_msg("foo")
|
| 19 | + c.open({}) |
| 20 | + c.on_msg(on_message) |
| 21 | + c.on_close(on_close) |
| 22 | + c.handle_msg({}) |
| 23 | + c.handle_close({}) |
| 24 | + c.close() |
| 25 | + assert len(msgs) == 2 |
| 26 | + |
| 27 | + |
| 28 | +def test_comm_manager(kernel): |
| 29 | + manager = CommManager(kernel=kernel) |
| 30 | + msgs = [] |
| 31 | + |
| 32 | + def foo(comm, msg): |
| 33 | + msgs.append(msg) |
| 34 | + comm.close() |
| 35 | + |
| 36 | + def fizz(comm, msg): |
| 37 | + raise RuntimeError('hi') |
| 38 | + |
| 39 | + def on_close(msg): |
| 40 | + msgs.append(msg) |
| 41 | + |
| 42 | + def on_msg(msg): |
| 43 | + msgs.append(msg) |
| 44 | + |
| 45 | + manager.register_target("foo", foo) |
| 46 | + manager.register_target("fizz", fizz) |
| 47 | + |
| 48 | + kernel.comm_manager = manager |
| 49 | + comm = Comm() |
| 50 | + comm.on_msg(on_msg) |
| 51 | + comm.on_close(on_close) |
| 52 | + manager.register_comm(comm) |
| 53 | + |
| 54 | + assert manager.get_comm(comm.comm_id) == comm |
| 55 | + assert manager.get_comm('foo') is None |
| 56 | + |
| 57 | + msg = dict(content=dict(comm_id=comm.comm_id, target_name='foo')) |
| 58 | + manager.comm_open(None, None, msg) |
| 59 | + assert len(msgs) == 1 |
| 60 | + msg['content']['target_name'] = 'bar' |
| 61 | + manager.comm_open(None, None, msg) |
| 62 | + assert len(msgs) == 1 |
| 63 | + msg = dict(content=dict(comm_id=comm.comm_id, target_name='fizz')) |
| 64 | + manager.comm_open(None, None, msg) |
| 65 | + assert len(msgs) == 1 |
| 66 | + |
| 67 | + manager.register_comm(comm) |
| 68 | + assert manager.get_comm(comm.comm_id) == comm |
| 69 | + msg = dict(content=dict(comm_id=comm.comm_id)) |
| 70 | + manager.comm_msg(None, None, msg) |
| 71 | + assert len(msgs) == 2 |
| 72 | + msg['content']['comm_id'] = 'foo' |
| 73 | + manager.comm_msg(None, None, msg) |
| 74 | + assert len(msgs) == 2 |
| 75 | + |
| 76 | + manager.register_comm(comm) |
| 77 | + assert manager.get_comm(comm.comm_id) == comm |
| 78 | + msg = dict(content=dict(comm_id=comm.comm_id)) |
| 79 | + manager.comm_close(None, None, msg) |
| 80 | + assert len(msgs) == 3 |
| 81 | + |
| 82 | + assert comm._closed |
| 83 | + |
| 84 | + |
| 85 | +def test_comm_in_manager(ipkernel: IPythonKernel) -> None: |
| 86 | + comm = Comm() |
| 87 | + assert comm.comm_id in ipkernel.comm_manager.comms |
0 commit comments