File: //usr/lib/python3/dist-packages/jeepney/io/tests/__pycache__/test_trio.cpython-310.pyc
o
���`4 � @ s� d dl Z d dlZd dlmZmZmZmZ d dlmZm Z d dl
m Z m Z m
Z
ddlmZ ejj ejje dd�gZd d
� Zed d d
d�Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zdd � ZdS ) � N)� DBusAddress�DBusErrorResponse� MessageType�new_method_call)� message_bus� MatchRule)�open_dbus_connection�open_dbus_router�Proxy� )�have_session_busz Tests require DBus session bus)�reasonc � s` �t dd�I d H } | 4 I d H � | j�d�sJ �W d �I d H d S 1 I d H s)w Y d S )N�SESSION��bus�:)r � unique_name�
startswith)�conn� r �</usr/lib/python3/dist-packages/jeepney/io/tests/test_trio.py� test_connect s
�.�r zorg.freedesktop.DBusz/org/freedesktop/DBuszorg.freedesktop.DBus.Peer)�bus_name� object_path� interfacec � s� �t td�} tdd�4 I d H �(}t�d�� |�| �I d H }W d � n1 s(w Y W d �I d H n 1 I d H s=w Y |jjtj ksKJ �|j
dksRJ �d S )N�Pingr r � r ) r �bus_peerr �trio�
fail_after�send_and_get_reply�header� message_typer �
method_return�body�� ping_call�req�replyr r r �test_send_and_get_reply s �
�(�r) c � s� �t td�} tdd�4 I d H �(}t�d�� |�| �I d H }W d � n1 s(w Y W d �I d H n 1 I d H s=w Y |jjtj ksKJ �d S )N�Snartr r r )
r r r r r r r! r" r �errorr% r r r � test_send_and_get_reply_error* s �
�(�r, c � s� �t dd�4 I d H �0} tt| �}d}|�|�I d H }|dv s J �|�|�I d H \}|du s/J �W d �I d H d S 1 I d H s@w Y d S )Nr r z+io.gitlab.takluyver.jeepney.examples.Server> �r �� T)r r
r � RequestName� NameHasOwner)r' �proxy�name�res� has_ownerr r r �
test_proxy3 s �
.�r6 c � s� �t dd�4 I d H �.} tt| �}t�t�� |�d�I d H W d � n1 s(w Y W d �I d H d S 1 I d H s>w Y d S )Nr r z:123)r r
r �pytest�raisesr r0 )r' r2 r r r �test_proxy_error> s �
�.�r9 c
� s2 �d} t dd�4 I d H ��}tt|�}tdtjtjdtjd�}|�d| � |�|�I d H |� |�4 I d H �A}|�
| �I d H \}|dksEJ �t � d �� |�
� I d H }W d � n1 s\w Y |j| d
|jfkslJ �W d �I d H n 1 I d H s|w Y W d �I d H d S 1 I d H s�w Y d S ) Nz2io.gitlab.takluyver.jeepney.tests.trio_test_filterr r �signal�NameOwnerChanged)�type�senderr �member�pathr r g @� )r r
r r r r r �add_arg_condition�AddMatch�filterr0 r r �receiver$ r )r3 �routerr �
match_rule�chanr4 �
signal_msgr r r � test_filterE s, �
� �*�.�rI c
� s� �t | d�}t�d��0 tddd�4 I d H �}|�|�I d H }W d �I d H n 1 I d H s/w Y W d � n1 s>w Y |jjtju sLJ �|j d �
d��}|� � dks]J �W d � d S 1 shw Y d S ) N�GetFDr r T�r �
enable_fdsr zw+�readme) r r r r r r! r" r r# r$ �to_file�read)�respond_with_fd�
getfd_callrE r( �fr r r � test_recv_fd_ s �
(�� �"�rS c
� s� �| \}}t |dd|f�}t�d��0 tddd�4 I d H �}|�|�I d H }W d �I d H n 1 I d H s6w Y W d � n1 sEw Y |jjtju sSJ �|j d |ks\J �d S )N�ReadFD�hr r TrK r )
r r r r r r! r" r r# r$ )�temp_file_and_contents� read_from_fd� temp_file�data� readfd_callrE r( r r r � test_send_fdj s � (�� �r[ ) r r7 �jeepneyr r r r �jeepney.bus_messagesr r �jeepney.io.trior r r
�utilsr �mark�skipif�
pytestmarkr r r) r, r6 r9 rI rS r[ r r r r �<module> s0 ��
�