test_tcp_address.py

import asyncio
import os

import pytest

from dbus_fast import Message
from dbus_fast._private.address import parse_address
from dbus_fast.aio import MessageBus


@pytest.mark.asyncio
async def test_tcp_connection_with_forwarding(event_loop):
    closables = []
    host = "127.0.0.1"
    port = "55556"

    addr_info = parse_address(os.environ.get("DBUS_SESSION_BUS_ADDRESS"))
    assert addr_info
    assert "abstract" in addr_info[0][1]
    path = f'\0{addr_info[0][1]["abstract"]}'

    async def handle_connection(tcp_reader, tcp_writer):
        unix_reader, unix_writer = await asyncio.open_unix_connection(path)
        closables.append(tcp_writer)
        closables.append(unix_writer)

        async def handle_read():
            while True:
                data = await tcp_reader.read(1)
                if not data:
                    break
                unix_writer.write(data)

        async def handle_write():
            while True:
                data = await unix_reader.read(1)
                if not data:
                    break
                tcp_writer.write(data)

        asyncio.run_coroutine_threadsafe(handle_read(), event_loop)
        asyncio.run_coroutine_threadsafe(handle_write(), event_loop)

    server = await asyncio.start_server(handle_connection, host, port)
    closables.append(server)

    bus = await MessageBus(bus_address=f"tcp:host={host},port={port}").connect()

    # basic tests to see if it works
    result = await bus.call(
        Message(
            destination="org.freedesktop.DBus",
            path="/org/freedesktop/DBus",
            interface="org.freedesktop.DBus.Peer",
            member="Ping",
        )
    )
    assert result

    intr = await bus.introspect("org.freedesktop.DBus", "/org/freedesktop/DBus")
    obj = bus.get_proxy_object("org.freedesktop.DBus", "/org/freedesktop/DBus", intr)
    iface = obj.get_interface("org.freedesktop.DBus.Peer")
    await iface.call_ping()

    assert bus._sock.getpeername()[0] == host
    assert bus._sock.getsockname()[0] == host
    assert bus._sock.gettimeout() == 0
    assert bus._stream.closed is False

    for c in closables:
        c.close()