|
Server : Apache System : Linux iad1-shared-b8-43 6.6.49-grsec-jammy+ #10 SMP Thu Sep 12 23:23:08 UTC 2024 x86_64 User : dh_edsupp ( 6597262) PHP Version : 8.2.26 Disable Function : NONE Directory : /lib/python3/dist-packages/jeepney/io/tests/__pycache__/ |
Upload File : |
o
���`�
� @ s� d dl Z d dlmZmZmZ d dlmZmZ d dlm Z m
Z
ddl m Z e j
je dd�Ze jd d
� �Zd d � Zed
ddd�Zdd� Zdd� Zdd� Zdd� Zdd� ZdS )� N)�new_method_call� MessageType� DBusAddress)� message_bus� MatchRule)�open_dbus_router�Proxy� )�have_session_busz Tests require DBus session bus)�reasonc c s8 � t dd�� } | V W d � d S 1 sw Y d S )N�SESSION)�bus)r )�conn� r �A/usr/lib/python3/dist-packages/jeepney/io/tests/test_threading.py�router s � "�r c C s | j �d�sJ �d S )N�:)� unique_name�
startswith)r r r r � test_connect s r zorg.freedesktop.DBusz/org/freedesktop/DBuszorg.freedesktop.DBus.Peer)�bus_name� object_path� interfacec C s< t td�}| j|dd�}|jjtjksJ �|jdks J �d S )N�Ping� ��timeoutr )r �bus_peer�send_and_get_reply�header� message_typer �
method_return�body)r � ping_call�replyr r r �test_send_and_get_reply s
r% c C sH t t| dd�}d}|�|�}|dv sJ �|j|dd�\}|du 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
r1 c C s� t t| �}d}tdtjtjdtjd�}|�d|� |�|� | �|��&}|� |�\}|dks0J �|j
dd�}|j |d | j fksAJ �W d � d S 1 sLw Y d S )
Nz7io.gitlab.takluyver.jeepney.tests.threading_test_filter�signal�NameOwnerChanged)�type�senderr �member�pathr r g @r � )
r r r r r r �add_arg_condition�AddMatch�filterr+ �getr" r )r r
r. �
match_rule�queuer/ �
signal_msgr r r � test_filter* s"
�
"�r@ c C s� t | d�}tddd��}|j|dd�}W d � n1 s w Y |jjtju s+J �|jd �d��}|� � d ks<J �W d � d S 1 sGw Y d S )
N�GetFDr T�r
�
enable_fdsr r r zw+�readme)
r r r r r r r! r" �to_file�read)�respond_with_fd�
getfd_callr r$ �fr r r � test_recv_fdB s
�"�rJ c C sz | \}}t |dd|f�}tddd��}|j|dd�}W d � n1 s$w Y |jjtju s2J �|jd |ks;J �d S ) N�ReadFD�hr TrB 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_fdL s �rR )�pytest�jeepneyr r r �jeepney.bus_messagesr r �jeepney.io.threadingr r �utilsr
�mark�skipif�
pytestmark�fixturer r r r% r1 r@ rJ rR r r r r �<module> s* �
�