o
    •ºé`¥  ã                   @   sª   d dl Z d dlmZmZmZ d dlmZmZ d dlm	Z	m
Z
 ddlmZ e jje dd�Ze jd	d
„ ƒZdd„ Zedddd�Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ ZdS )é    N)Únew_method_callÚMessageTypeÚDBusAddress)Úmessage_busÚ	MatchRule)Úopen_dbus_connectionÚProxyé   )Úhave_session_buszTests require DBus session bus)Úreasonc                  c   s8   � t dd��} | V  W d   ƒ d S 1 sw   Y  d S )NÚSESSION)Úbus)r   )Úconn© r   ú@/usr/lib/python3/dist-packages/jeepney/io/tests/test_blocking.pyÚsession_conn   s   €"ÿr   c                 C   s   | j  d¡sJ ‚d S )Nú:)Úunique_nameÚ
startswith)r   r   r   r   Útest_connect   s   r   zorg.freedesktop.DBusz/org/freedesktop/DBuszorg.freedesktop.DBus.Peer)Úbus_nameÚobject_pathÚ	interfacec                 C   sd   t tdƒ}| j|ddd�}|jjtjksJ ‚|jdksJ ‚t tdƒ}| j|ddd�}|dks0J ‚d S )NÚPingé   F)ÚtimeoutÚunwrapr   T)r   Úbus_peerÚsend_and_get_replyÚheaderÚmessage_typer   Úmethod_returnÚbody)r   Ú	ping_callÚreplyÚ
reply_bodyr   r   r   Útest_send_and_get_reply   s   

r&   c                 C   sH   t t| dd�}d}| |¡}|dv sJ ‚|j|dd�\}|du s"J ‚d S )Nr   ©r   z+io.gitlab.takluyver.jeepney.examples.Server>   ©r	   ©é   é   )Ú_timeoutT)r   r   ÚRequestNameÚNameHasOwner)r   ÚproxyÚnameÚresÚ	has_ownerr   r   r   Ú
test_proxy%   s   
r3   c                 C   s¨   t t| ƒ}d}tdtjtjdtjd�}| d|¡ | |¡ |  |¡�'}| 	|¡\}|dks0J ‚| j
|dd�}|j|d	| jfksBJ ‚W d   ƒ d S 1 sMw   Y  d S )
Nz6io.gitlab.takluyver.jeepney.tests.blocking_test_filterÚsignalÚNameOwnerChanged)ÚtypeÚsenderr   ÚmemberÚpathr   r	   r*   r'   Ú )r   r   r   r   r   r   Úadd_arg_conditionÚAddMatchÚfilterr-   Úrecv_until_filteredr"   r   )r   r   r0   Ú
match_ruleÚmatchesr1   Ú
signal_msgr   r   r   Útest_filter.   s"   
û
"úrB   c                 C   sœ   t | dƒ}tddd��}|j|dd�}W d   ƒ n1 sw   Y  |jjtju s+J ‚|jd  d¡�}| 	¡ d	ks<J ‚W d   ƒ d S 1 sGw   Y  d S )
NÚGetFDr   T©r   Ú
enable_fdsr   r'   r   zw+Úreadme)
r   r   r   r   r    r   r!   r"   Úto_fileÚread)Úrespond_with_fdÚ
getfd_callr   r$   Úfr   r   r   Útest_recv_fdG   s   
ÿ"ÿrL   c                 C   sz   | \}}t |dd|fƒ}tddd��}|j|dd�}W d   ƒ n1 s$w   Y  |jjtju s2J ‚|jd |ks;J ‚d S )	NÚReadFDÚhr   TrD   r   r'   r   )r   r   r   r   r    r   r!   r"   )Útemp_file_and_contentsÚread_from_fdÚ	temp_fileÚdataÚreadfd_callr   r$   r   r   r   Útest_send_fdQ   s   ÿrT   )ÚpytestÚjeepneyr   r   r   Újeepney.bus_messagesr   r   Újeepney.io.blockingr   r   Úutilsr
   ÚmarkÚskipifÚ
pytestmarkÚfixturer   r   r   r&   r3   rB   rL   rT   r   r   r   r   Ú<module>   s*    ÿ
ý
	
