ó Û¤[c@sdZddlZyddlmZWn!ek rIddlmZnXddlmZddlZddlm Z ddl Z ddl m Z ddl mZdefd „ƒYZd efd „ƒYZd efd „ƒYZdejfd„ƒYZedkrejƒndS(s* Tests for zmq shell / display publisher. iÿÿÿÿN(tQueue(tThread(tInt(tZMQDisplayPublisher(tSessiontNoReturnDisplayHookcBseZdZdZd„ZRS(s¢ 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. icCs|jd7_dS(Ni(t call_count(tselftobj((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyt__call__s(t__name__t __module__t__doc__RR (((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyRstReturnDisplayHookcBseZdZd„ZRS(s’ A dummy DisplayHook with the same counting ability as its base class, but which also returns the same message when it is called. cCstt|ƒj|ƒ|S(N(tsuperR R (RR((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyR )s(R R R R (((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyR #stCounterSessioncBs#eZdZedƒZd„ZRS(s This is a simple subclass to allow us to count the calls made to the session object by the display publisher. icOs,|jd7_tt|ƒj||ŽdS(su A trivial override to just augment the existing call with an increment to the send counter. iN(t send_countRRtsend(Rtargstkwargs((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyR6s(R R R RRR(((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyR.s tZMQDisplayPublisherTestscBsVeZdZd„Zd„Zd„Zd„Zd„Zd„Zd„Z d„Z RS( s6 Tests the ZMQDisplayPublisher in zmqshell.py cCsUtjƒ|_|jjtjƒ|_tƒ|_td|jd|jƒ|_dS(Ntsessiont pub_socket( tzmqtContexttcontexttsockettPUBRRRtdisp_pub(R((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pytsetUpDs   cCs+|jjƒ|jjƒ|jjƒdS(s¿ 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. N(Rt clear_outputRtcloseRtterm(R((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyttearDownNs  cCs:|jj|jkst‚|jj|jks6t‚dS(s Since there's no explicit constructor, here we confirm that keyword args get assigned correctly, and override the defaults. N(RRtAssertionErrorRR(R((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyttest_display_publisher_creationYscs¨ˆjjgkst‚d„}ˆjj|ƒˆjj|gksLt‚tƒ‰‡‡fd†}td|ƒ}|jƒˆjddƒ}|gks¤t‚dS(s„ Confirms that the thread_local attribute is correctly initialised with an empty list for the display hooks cSs|S(N((tmsg((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pythookhscsˆjˆjjƒdS(N(tputRt_hooks((tqR(s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pytset_thread_hooksnsttargetttimeouti N(RR'R"t register_hookRRtstarttget(RR%R)ttt thread_hooks((R(Rs=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyttest_thread_local_hooksbs   cCsStddƒ}|jjdks't‚|jj|ƒ|jjdksOt‚dS(sc Publish should prepare the message and eventually call `send` by default. taiiN(tdictRRR"Rtpublish(Rtdata((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyt test_publishuscCs–tddƒ}tƒ}|jj|ƒ|jdks=t‚|jjdksUt‚|jj|ƒ|jdkszt‚|jjdks’t‚dS(sé 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. R2iiN( R3RRR,RR"RRR4(RR5R%((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyttest_display_hook_halts_sends cCs–tddƒ}tƒ}|jj|ƒ|jdks=t‚|jjdksUt‚|jj|ƒ|jdkszt‚|jjdks’t‚dS(sã 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. R2iiN( R3R RR,RR"RRR4(RR5R%((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyt#test_display_hook_return_calls_send’s cCstddƒ}tƒ}|jj|ƒ|jdks=t‚|jjdksUt‚|jj|ƒ|jdkszt‚|jjdks’t‚|jj |ƒ}|jj|ƒ|j |ƒ|jdksÖt‚|jjdksît‚|jj |ƒ}|j |ƒdS(s` Once a hook is unregistered, it should not be called during `publish`. R2iiN( R3RRR,RR"RRR4tunregister_hookt assertTruet assertFalse(RR5R%tfirsttsecond((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyttest_unregister_hook¥s   ( R R R RR!R#R1R6R7R8R>(((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyR?s   t__main__(R tostqueueRt ImportErrort threadingRtunittestt traitletsRRtipykernel.zmqshellRtjupyter_client.sessionRtobjectRR RtTestCaseRR tmain(((s=lib/python2.7/site-packages/ipykernel/tests/test_zmq_shell.pyts"