Gecko [ www.utparral.edu.mx ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
conftest.py	2.67 KB	-rw-r--r--
__init__.py	0 B	-rw-r--r--
test_asyncio.py	2.54 KB	-rw-r--r--
test_blocking.py	2.91 KB	-rw-r--r--
test_threading.py	2.64 KB	-rw-r--r--
test_trio.py	3.8 KB	-rw-r--r--
utils.py	79 B	-rw-r--r--

Code Editor : conftest.py

from tempfile import TemporaryFile
import threading

import pytest

from jeepney import (
    DBusAddress,  HeaderFields, message_bus, MessageType, new_error,
    new_method_return,
)
from jeepney.io.threading import open_dbus_connection, DBusRouter, Proxy

@pytest.fixture()
def respond_with_fd():
    name = "io.gitlab.takluyver.jeepney.tests.respond_with_fd"
    addr = DBusAddress(bus_name=name, object_path='/')

with open_dbus_connection(bus='SESSION', enable_fds=True) as conn:
        with DBusRouter(conn) as router:
            status, = Proxy(message_bus, router).RequestName(name)
        assert status == 1  # DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER

def _reply_once():
            while True:
                msg = conn.receive()
                if msg.header.message_type is MessageType.method_call:
                    if msg.header.fields[HeaderFields.member] == 'GetFD':
                        with TemporaryFile('w+') as tf:
                            tf.write('readme')
                            tf.seek(0)
                            rep = new_method_return(msg, 'h', (tf,))
                            conn.send(rep)
                            return
                    else:
                        conn.send(new_error(msg, 'NoMethod'))

reply_thread = threading.Thread(target=_reply_once, daemon=True)
        reply_thread.start()
        yield addr

reply_thread.join()

@pytest.fixture()
def read_from_fd():
    name = "io.gitlab.takluyver.jeepney.tests.read_from_fd"
    addr = DBusAddress(bus_name=name, object_path='/')

def _reply_once():
            while True:
                msg = conn.receive()
                if msg.header.message_type is MessageType.method_call:
                    if msg.header.fields[HeaderFields.member] == 'ReadFD':
                        with msg.body[0].to_file('rb') as f:
                            f.seek(0)
                            b = f.read()
                        conn.send(new_method_return(msg, 'ay', (b,)))
                        return
                    else:
                        conn.send(new_error(msg, 'NoMethod'))

reply_thread = threading.Thread(target=_reply_once, daemon=True)
        reply_thread.start()
        yield addr

reply_thread.join()

@pytest.fixture()
def temp_file_and_contents():
    data = b'abc123'
    with TemporaryFile('w+b') as tf:
        tf.write(data)
        tf.flush()
        tf.seek(0)
        yield tf, data

${this.title}

Code Editor : conftest.py