/usr/lib/python3/dist-packages/jeepney/io/tests/__pycache__
Edit: /usr/lib/python3/dist-packages/jeepney/io/tests/__pycache__/conftest.cpython-310.pyc (2402B)
o
`
@ s d dl mZ d dlZd dlZd dlmZmZmZmZm Z m
Z
d dl m Z m
Z
mZ e dd Ze dd Ze d d
ZdS ) )
TemporaryFileN) DBusAddress HeaderFields message_bus MessageType new_errornew_method_return)open_dbus_connection
DBusRouterProxyc # d} t | dd}tddd? t }tt|| \}W d n1 s(w Y |dks3J fdd }tj|dd
}| |V W d n1 sQw Y | d S ) Nz1io.gitlab.takluyver.jeepney.tests.respond_with_fd/bus_name object_pathSESSIONTbus
enable_fds c s } | jjtju rJ| jjtj dkrBtd }| d |
d t | d|f} | W d d S 1 s
._reply_oncetargetdaemon
r r r
r r RequestName threadingThreadstartjoinnameaddrrouterstatusr* reply_threadr( r& r) respond_with_fd s
r: c # r ) Nz.io.gitlab.takluyver.jeepney.tests.read_from_fdr
r r Tr r c s } | jjtju rJ| jjtj dkrB| jd d}|
d | }W d n1 s1w Y t
| d|f d S t| d q)NTReadFDr rbayr )r r r r r r r r bodyto_filer! readr" r r )r# fbr& r( r) r* 5 s
z!read_from_fd.._reply_oncer+ r. r4 r( r& r) read_from_fd+ s
rC c c sZ d} t d }|| | |d || fV W d d S 1 s&w Y d S )Ns abc123zw+br )r r flushr! )datar$ r( r( r) temp_file_and_contentsI s
"rF )tempfiler r0 pytestjeepneyr r r r r r jeepney.io.threadingr r
r fixturer: rC rF r( r( r( r) s