Python asyncio 模块，open_unix_connection() 实例源码

def client(loop, core_socket):
    reader, writer = loop.run_until_complete(
        asyncio.open_unix_connection(path=core_socket))

    class CoreClient:
        def __init__(self, reader, writer):
            self.reader = reader
            self.writer = writer

        async def send_cmd(self, cmd, **kwargs):
            msg = {'cmd': cmd, **kwargs}
            raw_msg = ejson_dumps(msg).encode()
            self.writer.write(raw_msg)
            self.writer.write(b'\n')
            raw_resp = await self.reader.readline()
            return ejson_loads(raw_resp.decode())

    yield CoreClient(reader, writer)
    writer.close()

def repl(socket_path):
    from parsec import __version__
    print('Parsec shell version: %s' % __version__)
    print('Connecting to: %s' % socket_path)
    open_conn = partial(asyncio.open_unix_connection, path=socket_path)
    reader, writer = await open_conn()
    quit = False
    while not quit:
        data = input('>>> ')
        if data in ('quit', 'q'):
            writer.close()
            return
        elif data in ('help', 'h'):
            print('No help for the braves !')
            continue
        elif data in ('reload', 'r'):
            writer.close()
            reader, writer = await open_conn()
            continue
        writer.write(data.encode())
        writer.write(b'\n')
        raw_resp = await reader.readline()
        resp = ejson_loads(raw_resp.decode())
        print('Received: %r' % resp)

def test_unix_server_auto_close_connection(event_loop, tempcwd):
    """Connections are closed automatically when sessions finish."""

    # Intentionally do not close the writer.
    async def noop(reader, writer):
        pass

    path = './echo.sock'
    async with UnixSocketServer(path, noop, loop=event_loop) as server:
        reader, writer = await asyncio.open_unix_connection(
            path=path,
            loop=event_loop,
        )
        try:
            with pytest.raises(asyncio.IncompleteReadError):
                await asyncio.wait_for(
                    reader.readexactly(1), timeout=0.1,
                )
            server.close()
            await server.wait_closed()
        finally:
            writer.close()

def _open_connection(*, loop, addr, params: _ConnectionParameters):
    if isinstance(addr, str):
        r, w = await asyncio.open_unix_connection(addr, loop=loop)
    else:
        if params.ssl:
            sock = await _get_ssl_ready_socket(*addr, loop=loop)

            try:
                r, w = await asyncio.open_connection(
                    sock=sock,
                    loop=loop,
                    ssl=params.ssl,
                    server_hostname=addr[0])
            except Exception:
                sock.close()
                raise

        else:
            r, w = await asyncio.open_connection(*addr, loop=loop)
            _set_nodelay(_get_socket(w.transport))

    return r, w

def _setUp(self):
        self.r, self.w = yield from asyncio.open_unix_connection(self.path)
        config = H2Configuration(header_encoding='utf-8')
        self.conn = H2Connection(config=config)
        self.conn.initiate_connection()
        self.w.write(self.conn.data_to_send())
        events = yield from self._expect_events(3)
        self.assertIsInstance(events[0], RemoteSettingsChanged)
        self.assertIsInstance(events[1], RemoteSettingsChanged)
        self.assertIsInstance(events[2], SettingsAcknowledged)

        self.assertIsInstance((yield from self.server.events.get()),
                              RemoteSettingsChanged)
        self.assertIsInstance((yield from self.server.events.get()),
                              SettingsAcknowledged)
        self.assertIsInstance((yield from self.server.events.get()),
                              SettingsAcknowledged)

def test_open_unix_connection(self):
        with test_utils.run_test_unix_server() as httpd:
            conn_fut = asyncio.open_unix_connection(httpd.address,
                                                    loop=self.loop)
            self._basetest_open_connection(conn_fut)

def test_open_unix_connection_no_loop_ssl(self):
        with test_utils.run_test_unix_server(use_ssl=True) as httpd:
            conn_fut = asyncio.open_unix_connection(
                httpd.address,
                ssl=test_utils.dummy_ssl_context(),
                server_hostname='',
                loop=self.loop)

            self._basetest_open_connection_no_loop_ssl(conn_fut)

def test_open_unix_connection_error(self):
        with test_utils.run_test_unix_server() as httpd:
            conn_fut = asyncio.open_unix_connection(httpd.address,
                                                    loop=self.loop)
            self._basetest_open_connection_error(conn_fut)

def connect(self) -> None:
        if self.host.startswith('/'):
            self.reader, self.writer = await asyncio.open_unix_connection(
                path=self.host, loop=self.loop
            )
        else:
            self.reader, self.writer = await asyncio.open_connection(
                host=self.host, port=self.port, loop=self.loop
            )

        sock = self.writer.transport.get_extra_info('socket')
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, int(self.options.pool_options.socket_keepalive))

        if self.host.startswith('/'):
            endpoint = self.host
        else:
            endpoint = '{}:{}'.format(self.host, self.port)
        logger.debug('Established connection to {}'.format(endpoint))
        self.read_loop_task = asyncio.ensure_future(self.read_loop(), loop=self.loop)

        ismaster = IsMaster(await self.command(
            'admin', SON([('ismaster', 1)]), ReadPreference.PRIMARY, DEFAULT_CODEC_OPTIONS
        ))

        self.is_mongos = ismaster.server_type == SERVER_TYPE.Mongos
        self.max_wire_version = ismaster.max_wire_version
        if ismaster.max_bson_size:
            self.max_bson_size = ismaster.max_bson_size
        if ismaster.max_message_size:
            self.max_message_size = ismaster.max_message_size
        if ismaster.max_write_batch_size:
            self.max_write_batch_size = ismaster.max_write_batch_size
        self.is_writable = ismaster.is_writable

        self.slave_ok = not self.is_mongos and self.options.read_preference != ReadPreference.PRIMARY

        if self.options.credentials:
            await self._authenticate()

        # Notify waiters that connection has been established
        self.__connected.set()

def test_socket_communication(loop):
    async def on_connection(reader, writer):
        cmd = await reader.readline()
        assert cmd == b'ping\n'
        writer.write(b'pong\n')

    async with run_unix_socket_server(on_connection=on_connection, loop=loop) as server:
        reader, writer = await asyncio.open_unix_connection(server.socket_path, loop=loop)
        try:
            writer.write(b'ping\n')
            resp = await reader.readline()
            assert resp == b'pong\n'
        finally:
            writer.close()

def test_run_core(core, core_socket):
    reader, writer = await asyncio.open_unix_connection(path=core_socket)
    writer.write(b'{"cmd": "ping", "ping": "hello"}\n')
    raw_resp = await reader.readline()
    assert raw_resp == b'{"pong": "hello", "status": "ok"}\n'

def signup(socket, identity, key_size):
    while True:
        password = getpass('Password:')
        repassword = getpass('Confirm password:')
        if password == repassword:
            break
        print('Passwords missmatch, please retry')

    import asyncio
    from parsec.tools import ejson_loads, ejson_dumps

    async def run():
        try:
            reader, writer = await asyncio.open_unix_connection(path=socket)
        except (FileNotFoundError, ConnectionRefusedError):
            raise SystemExit('ERROR: Cannot connect to parsec core at %s' % socket)
        msg = {
            'cmd': 'identity_signup',
            'id': identity,
            'password': password,
            'key_size': key_size
        }
        writer.write(ejson_dumps(msg).encode())
        writer.write(b'\n')
        raw_resp = await reader.readline()
        resp = ejson_loads(raw_resp.decode())
        writer.close()
        print(resp)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())

def test_ipc_requests(ipc_pipe_path, request_msg, expected, event_loop):
    assert wait_for(ipc_pipe_path), "IPC server did not successfully start with IPC file"
    reader, writer = await asyncio.open_unix_connection(ipc_pipe_path, loop=event_loop)
    writer.write(request_msg)
    await writer.drain()
    try:
        result_bytes = await asyncio.tasks.wait_for(reader.readuntil(b'}'), 0.25, loop=event_loop)
        result = json.loads(result_bytes.decode())
    except asyncio.TimeoutError:
        result = None
    assert result == expected
    writer.close()

def open_connection(self):
        reader, writer = await open_ipc_connection(
            path=self.path,
            loop=self.loop,
        )

        logger.debug("Connection to %s established.", self.path)

        try:
            async with StreamConnection(
                reader=reader,
                writer=writer,
                address=self.path,
                zap_client=self.zap_client,
                socket_type=self.socket_type,
                identity=self.identity,
                mechanism=self.mechanism,
                on_ready=self.on_connection_ready.emit,
                on_lost=self.on_connection_lost.emit,
                on_failure=self.on_connection_failure,
                loop=self.loop,
            ) as connection:
                self.register_child(connection)
                await connection.wait_closed()

            return connection.get_error()
        finally:
            logger.debug("Connection to %s closed.", self.path)

def test_unix_server(event_loop, tempcwd):
    """Basic connectivity check."""

    async def echo(reader, writer):
        try:
            chunk = await reader.read(1024)
            while chunk:
                writer.write(chunk)
                chunk = await reader.read(1024)
        finally:
            writer.close()

    path = './echo.sock'
    async with UnixSocketServer(path, echo, event_loop) as server:
        assert server.path == path
        reader, writer = await asyncio.open_unix_connection(
            path=path,
            loop=event_loop,
        )
        try:
            req = b'FOO'
            writer.write(req)
            rep = await reader.read(1024)
            assert req == rep
        finally:
            writer.close()

def test_open_unix_connection(self):
        with test_utils.run_test_unix_server() as httpd:
            conn_fut = asyncio.open_unix_connection(httpd.address,
                                                    loop=self.loop)
            self._basetest_open_connection(conn_fut)

def test_open_unix_connection_no_loop_ssl(self):
        with test_utils.run_test_unix_server(use_ssl=True) as httpd:
            conn_fut = asyncio.open_unix_connection(
                httpd.address,
                ssl=test_utils.dummy_ssl_context(),
                server_hostname='',
                loop=self.loop)

            self._basetest_open_connection_no_loop_ssl(conn_fut)

def test_open_unix_connection_error(self):
        with test_utils.run_test_unix_server() as httpd:
            conn_fut = asyncio.open_unix_connection(httpd.address,
                                                    loop=self.loop)
            self._basetest_open_connection_error(conn_fut)

def _connect(self):
        reader, writer = await exec_with_timeout(
            asyncio.open_unix_connection(path=self.path,
                                         ssl=self.ssl_context,
                                         loop=self.loop),
            self._connect_timeout,
            loop=self.loop
        )
        self._reader = reader
        self._writer = writer
        await self.on_connect()

def test_open_unix_connection(self):
        with test_utils.run_test_unix_server() as httpd:
            conn_fut = asyncio.open_unix_connection(httpd.address,
                                                    loop=self.loop)
            self._basetest_open_connection(conn_fut)

def test_open_unix_connection_no_loop_ssl(self):
        with test_utils.run_test_unix_server(use_ssl=True) as httpd:
            conn_fut = asyncio.open_unix_connection(
                httpd.address,
                ssl=test_utils.dummy_ssl_context(),
                server_hostname='',
                loop=self.loop)

            self._basetest_open_connection_no_loop_ssl(conn_fut)

def test_open_unix_connection_error(self):
        with test_utils.run_test_unix_server() as httpd:
            conn_fut = asyncio.open_unix_connection(httpd.address,
                                                    loop=self.loop)
            self._basetest_open_connection_error(conn_fut)

def xtest_open_unix_connection(self):
    with test_utils.run_test_unix_server() as httpd:
        conn_fut = asyncio.open_unix_connection(httpd.address,
                                                loop=self.loop)
        self._basetest_open_connection(conn_fut)

def xtest_open_unix_connection_no_loop_ssl(self):
    with test_utils.run_test_unix_server(use_ssl=True) as httpd:
        conn_fut = asyncio.open_unix_connection(
            httpd.address,
            ssl=test_utils.dummy_ssl_context(),
            server_hostname='',
            loop=self.loop)

        self._basetest_open_connection_no_loop_ssl(conn_fut)

def xtest_open_unix_connection_error(self):
    with test_utils.run_test_unix_server() as httpd:
        conn_fut = asyncio.open_unix_connection(httpd.address,
                                                loop=self.loop)
        self._basetest_open_connection_error(conn_fut)

def uri_compile(uri):
    url = urllib.parse.urlparse(uri)
    rawprotos = url.scheme.split('+')
    err_str, protos = proto.get_protos(rawprotos)
    if err_str:
        raise argparse.ArgumentTypeError(err_str)
    if 'ssl' in rawprotos or 'secure' in rawprotos:
        import ssl
        if not hasattr(ssl, 'Purpose'):
            raise argparse.ArgumentTypeError('ssl support is available for Python 3.4 and above')
        sslserver = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        sslclient = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        if 'ssl' in rawprotos:
            sslclient.check_hostname = False
            sslclient.verify_mode = ssl.CERT_NONE
    else:
        sslserver = None
        sslclient = None
    cipher, _, loc = url.netloc.rpartition('@')
    if cipher:
        from pproxy.cipher import get_cipher
        err_str, cipher = get_cipher(cipher)
        if err_str:
            raise argparse.ArgumentTypeError(err_str)
    match = pattern_compile(url.query) if url.query else None
    if loc:
        host, _, port = loc.partition(':')
        port = int(port) if port else 8080
        connect = functools.partial(asyncio.open_connection, host=host, port=port, ssl=sslclient)
        server = functools.partial(asyncio.start_server, host=host, port=port, ssl=sslserver)
    else:
        connect = functools.partial(asyncio.open_unix_connection, path=url.path, ssl=sslclient, server_hostname='' if sslclient else None)
        server = functools.partial(asyncio.start_unix_server, path=url.path, ssl=sslserver)
    return types.SimpleNamespace(protos=protos, rproto=protos[0], cipher=cipher, auth=url.fragment.encode(), match=match, server=server, connect=connect, bind=loc or url.path, sslclient=sslclient, sslserver=sslserver)

def test_unix_server_wait_closed_timeout(event_loop, tempcwd):
    """Not finished closing until all sessions complete."""

    # This session will (intentionally) never receive enough data.  After a
    # while, it will get cancelled and (intentionally) ignore the cancellation
    # request to simulate a badly designed session handler.  Once our test
    # verifies this behavior, it will close the client socket, which will
    # finally trigger the shutdown sequence.
    async def noop(reader, writer):
        try:
            await reader.readexactly(1024)
        except asyncio.CancelledError:
            try:
                await reader.readexactly(1024)
            except asyncio.IncompleteReadError:
                pass
        finally:
            writer.close()

    path = './echo.sock'
    server = UnixSocketServer(path, noop, loop=event_loop)
    server.start()
    await server.wait_started()
    try:
        reader, writer = await asyncio.open_unix_connection(
            path=path,
            loop=event_loop,
        )
        try:
            writer.write(b'REQ')
            with pytest.raises(asyncio.TimeoutError):
                await asyncio.wait_for(
                    reader.readexactly(1), timeout=0.1,
                )
            server.close()
            with pytest.raises(asyncio.TimeoutError):
                await server.wait_closed(timeout=0.1)
        finally:
            writer.close()
    finally:
        server.close()
        await server.wait_closed()

def _get_events_reader(self, vm=None) -> (asyncio.StreamReader, callable):
        '''Make connection to qubesd and return stream to read events from

        :param vm: Specific VM for which events should be handled, use None
        to handle events from all VMs (and non-VM objects)
        :return stream to read events from and a cleanup function
        (call it to terminate qubesd connection)'''
        if vm is not None:
            dest = vm.name
        else:
            dest = 'dom0'

        if self.app.qubesd_connection_type == 'socket':
            reader, writer = yield from asyncio.open_unix_connection(
                qubesadmin.config.QUBESD_SOCKET)
            writer.write(b'dom0\0')  # source
            writer.write(self._api_method.encode() + b'\0')  # method
            writer.write(dest.encode('ascii') + b'\0')  # dest
            writer.write(b'\0')  # arg
            writer.write_eof()

            def cleanup_func():
                '''Close connection to qubesd'''
                writer.close()
        elif self.app.qubesd_connection_type == 'qrexec':
            proc = yield from asyncio.create_subprocess_exec(
                'qrexec-client-vm', dest, self._api_method,
                stdin=subprocess.PIPE, stdout=subprocess.PIPE)

            proc.stdin.write_eof()
            reader = proc.stdout

            def cleanup_func():
                '''Close connection to qubesd'''
                try:
                    proc.kill()
                except ProcessLookupError:
                    pass
        else:
            raise NotImplementedError('Unsupported qubesd connection type: '
                                      + self.app.qubesd_connection_type)
        return reader, cleanup_func