
    
8d                         d dl Z d dlmZ d dlZd dlmZ  G d dej                  Zedk    r ej	                     dS dS )    N)Empty)QtKernelManagerc                   .    e Zd Zd Zd ZddZd Zd ZdS )	Testsc                    t                      | _        | j                                         | j                                        | _        | j                            dd           | j                                        | _        | j                            dd           | j        j        | _        | j                            d           	 | 	                                 | 	                                 dS # t          $ rF | j                            d           | 	                                 | 	                                 Y dS w xY w)zOpen a kernel.T)shelliopubzprint(0)N)r   kernel_managerstart_kernelclientkernel_clientstart_channelsblocking_clientcomm_managerexecute_get_next_msgTimeoutErrorselfs    :lib/python3.11/site-packages/qtconsole/tests/test_comms.pysetUpzTests.setUp   s@   -//((***!07799))D)AAA#1AACC++$d+CCC .; 	$$Z000	!         	! 	! 	! ((444         		!s   (C7 7AEEc                     | j         r| j                             d           | j        r| j                                         dS dS )zClose the kernel.T)nowN)r
   shutdown_kernelr   shutdownr   s    r   tearDownzTests.tearDown!   sW     	://D/999 	*'')))))	* 	*    
   c                     t          j                     |z   }d}|dk    r_|t          j                     k     rt          	 | j                            d          }|d         d         }n# t          $ r Y nw xY w|dk    _|S )Nstatus   )timeoutheadermsg_type)timer   r   get_iopub_msgr   )r   r"   timeout_timer$   msgs        r   r   zTests._get_next_msg(   s    y{{W,(""dikk))""*888CCx=4    ("" 
s   )A( (
A54A5c                    | j         | j        } G fdd          } |            }|                    d           |                                 }|d         d         dk    sJ |                                 }|d         d         dk    sJ                     |           |j        dk    sJ |j        j        |d	         d
         k    sJ |                                 }|d         d         dk    sJ                     |           |j        dk    sJ |j        j        |d	         d
         k    sJ |                                 }|d         d         dk    sJ                     |           |j        dk    sJ |j        j        |d	         d
         k    sJ |                                 }|d         d         dk    sJ dS )z,Communicate from the kernel to the frontend.c                   &    e Zd Z fdZd Zd ZdS )7Tests.test_kernel_to_frontend.<locals>.DummyCommHandlerc                 L                         d| j                   d | _        d S )Ntest_api)register_target	comm_openlast_msg)r   r   s    r   __init__z@Tests.test_kernel_to_frontend.<locals>.DummyCommHandler.__init__<   s&    ,,ZHHH $r   c                     |                     | j                   |                    | j                   |d         d         | _        || _        d S Ncontentdata)on_msgcomm_messageon_closer0   comm)r   r9   r(   s      r   r/   zATests.test_kernel_to_frontend.<locals>.DummyCommHandler.comm_open@   sG    D-...d/000 #Iv 6 			r   c                 ,    |d         d         | _         d S r3   )r0   )r   r(   s     r   r7   zDTests.test_kernel_to_frontend.<locals>.DummyCommHandler.comm_messageF   s     #Iv 6r   N)__name__
__module____qualname__r1   r/   r7   )r   s   r   DummyCommHandlerr+   ;   sL        % % % % %! ! !7 7 7 7 7r   r>   zfrom ipykernel.comm import Comm
comm = Comm(target_name='test_api', data='open')
comm.send('message')
comm.close('close')
del comm
print('Done')
r#   r$   execute_inputr/   openr4   comm_idcomm_msgmessage
comm_closeclosestreamN)r   r   r   r   	_dispatchr0   r9   rA   )r   r   r>   handlerr(   r   s        @r   test_kernel_to_frontendzTests.test_kernel_to_frontend6   s&   (.	7 	7 	7 	7 	7 	7 	7 	7 	7 	7 #"$$		
 	
 	
   ""8}Z(O;;;;  ""8}Z(K7777s###6))))|#s9~i'@@@@@  ""8}Z(J6666s###9,,,,|#s9~i'@@@@@  ""8}Z(L8888s###7****|#s9~i'@@@@@  ""8}Z(H444444r   c                    | j         }| j        }|                    d           |                                 }|d         d         dk    sJ |                    dd          }|                                 }|d         d         dk    sJ |d	         d
         dk    sJ |                    d           |                                 }|d         d         dk    sJ |d	         d
         dk    sJ |                    d           |                                 }|d         d         dk    sJ |d         dk    sJ |d	         d
         dk    sJ dS )z,Communicate from the frontend to the kernel.a  class DummyCommHandler():
    def __init__(self):
        get_ipython().kernel.comm_manager.register_target(
            'test_api', self.comm_open)
    def comm_open(self, comm, msg):
        comm.on_msg(self.comm_message)
        comm.on_close(self.comm_message)
        print(msg['content']['data'])
    def comm_message(self, msg):
        print(msg['content']['data'])
dummy = DummyCommHandler()
r#   r$   r?   r-   r@   )r5   rF   r4   textzopen
rC   zmessage
rE   parent_headerrD   zclose
N)r   r   r   r   new_commsendrE   )r   r   r   r(   r9   s        r   test_frontend_to_kernelzTests.test_frontend_to_kernelk   s   (.
+	
 	
 	
   ""8}Z(O;;;;$$Zf$==  ""8}Z(H44449~f%1111		)  ""8}Z(H44449~f%4444

7  "" ?#J/<????:(****9~f%222222r   N)r   )r;   r<   r=   r   r   r   rI   rO    r   r   r   r   	   se        ! ! !,* * *   35 35 35j&3 &3 &3 &3 &3r   r   __main__)
r%   queuer   unittestqtconsole.managerr   TestCaser   r;   mainrP   r   r   <module>rW      s            - - - - - -H3 H3 H3 H3 H3H H3 H3 H3T zHMOOOOO r   