Gecko [ lorencats.com ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	998 B	-rw-r--r--
_asyncio.py	363 B	-rw-r--r--
test_connect.py	1.92 KB	-rw-r--r--
test_embed_kernel.py	4.53 KB	-rw-r--r--
test_eventloop.py	971 B	-rw-r--r--
test_io.py	1.06 KB	-rw-r--r--
test_jsonutil.py	3.18 KB	-rw-r--r--
test_kernel.py	9.38 KB	-rw-r--r--
test_kernelspec.py	3.63 KB	-rw-r--r--
test_message_spec.py	14.27 KB	-rw-r--r--
test_pickleutil.py	1.17 KB	-rw-r--r--
test_serialize.py	5.44 KB	-rw-r--r--
test_start_kernel.py	1.73 KB	-rw-r--r--
test_zmq_shell.py	5.75 KB	-rw-r--r--
utils.py	4.66 KB	-rw-r--r--

Code Editor : test_zmq_shell.py

# -*- 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()

${this.title}

Code Editor : test_zmq_shell.py