/usr/lib/python3/dist-packages/ipykernel/tests/test_zmq_shell.py is in python3-ipykernel 4.8.2-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | # -*- coding: utf-8 -*-
""" Tests for zmq shell / display publisher. """
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import os
try:
from queue import Queue
except ImportError:
# py2
from Queue import Queue
from threading import Thread
import unittest
from traitlets import Int
import zmq
from ipykernel.zmqshell import ZMQDisplayPublisher
from jupyter_client.session import Session
class NoReturnDisplayHook(object):
"""
A dummy DisplayHook which allows us to monitor
the number of times an object is called, but which
does *not* return a message when it is called.
"""
call_count = 0
def __call__(self, obj):
self.call_count += 1
class ReturnDisplayHook(NoReturnDisplayHook):
"""
A dummy DisplayHook with the same counting ability
as its base class, but which also returns the same
message when it is called.
"""
def __call__(self, obj):
super(ReturnDisplayHook, self).__call__(obj)
return obj
class CounterSession(Session):
"""
This is a simple subclass to allow us to count
the calls made to the session object by the display
publisher.
"""
send_count = Int(0)
def send(self, *args, **kwargs):
"""
A trivial override to just augment the existing call
with an increment to the send counter.
"""
self.send_count += 1
super(CounterSession, self).send(*args, **kwargs)
class ZMQDisplayPublisherTests(unittest.TestCase):
"""
Tests the ZMQDisplayPublisher in zmqshell.py
"""
def setUp(self):
self.context = zmq.Context()
self.socket = self.context.socket(zmq.PUB)
self.session = CounterSession()
self.disp_pub = ZMQDisplayPublisher(
session = self.session,
pub_socket = self.socket
)
def tearDown(self):
"""
We need to close the socket in order to proceed with the
tests.
TODO - There is still an open file handler to '/dev/null',
presumably created by zmq.
"""
self.disp_pub.clear_output()
self.socket.close()
self.context.term()
def test_display_publisher_creation(self):
"""
Since there's no explicit constructor, here we confirm
that keyword args get assigned correctly, and override
the defaults.
"""
assert self.disp_pub.session == self.session
assert self.disp_pub.pub_socket == self.socket
def test_thread_local_hooks(self):
"""
Confirms that the thread_local attribute is correctly
initialised with an empty list for the display hooks
"""
assert self.disp_pub._hooks == []
def hook(msg):
return msg
self.disp_pub.register_hook(hook)
assert self.disp_pub._hooks == [hook]
q = Queue()
def set_thread_hooks():
q.put(self.disp_pub._hooks)
t = Thread(target=set_thread_hooks)
t.start()
thread_hooks = q.get(timeout=10)
assert thread_hooks == []
def test_publish(self):
"""
Publish should prepare the message and eventually call
`send` by default.
"""
data = dict(a = 1)
assert self.session.send_count == 0
self.disp_pub.publish(data)
assert self.session.send_count == 1
def test_display_hook_halts_send(self):
"""
If a hook is installed, and on calling the object
it does *not* return a message, then we assume that
the message has been consumed, and should not be
processed (`sent`) in the normal manner.
"""
data = dict(a = 1)
hook = NoReturnDisplayHook()
self.disp_pub.register_hook(hook)
assert hook.call_count == 0
assert self.session.send_count == 0
self.disp_pub.publish(data)
assert hook.call_count == 1
assert self.session.send_count == 0
def test_display_hook_return_calls_send(self):
"""
If a hook is installed and on calling the object
it returns a new message, then we assume that this
is just a message transformation, and the message
should be sent in the usual manner.
"""
data = dict(a=1)
hook = ReturnDisplayHook()
self.disp_pub.register_hook(hook)
assert hook.call_count == 0
assert self.session.send_count == 0
self.disp_pub.publish(data)
assert hook.call_count == 1
assert self.session.send_count == 1
def test_unregister_hook(self):
"""
Once a hook is unregistered, it should not be called
during `publish`.
"""
data = dict(a = 1)
hook = NoReturnDisplayHook()
self.disp_pub.register_hook(hook)
assert hook.call_count == 0
assert self.session.send_count == 0
self.disp_pub.publish(data)
assert hook.call_count == 1
assert self.session.send_count == 0
#
# After unregistering the `NoReturn` hook, any calls
# to publish should *not* got through the DisplayHook,
# but should instead hit the usual `session.send` call
# at the end.
#
# As a result, the hook call count should *not* increase,
# but the session send count *should* increase.
#
first = self.disp_pub.unregister_hook(hook)
self.disp_pub.publish(data)
self.assertTrue(first)
assert hook.call_count == 1
assert self.session.send_count == 1
#
# If a hook is not installed, `unregister_hook`
# should return false.
#
second = self.disp_pub.unregister_hook(hook)
self.assertFalse(second)
if __name__ == '__main__':
unittest.main()
|