# SPDX-License-Identifier: LGPL-2.1-or-later

# Copyright (C) 2020-2023 igo95862

# This file is part of python-sdbus

# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.

# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.

# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
from __future__ import annotations

from asyncio import Event, get_running_loop
from asyncio import run as asyncio_run
from asyncio import sleep, wait_for
from asyncio.subprocess import create_subprocess_exec
from typing import TYPE_CHECKING
from unittest import SkipTest

from sdbus.exceptions import (
    DbusFailedError,
    DbusFileExistsError,
    DbusNoReplyError,
    DbusPropertyReadOnlyError,
    DbusUnknownObjectError,
    SdBusLibraryError,
    SdBusUnmappedMessageError,
)
from sdbus.sd_bus_internals import (
    DBUS_ERROR_TO_EXCEPTION,
    DbusPropertyEmitsChangeFlag,
)
from sdbus.unittest import IsolatedDbusTestCase
from sdbus.utils.parse import parse_properties_changed

from sdbus import (
    DbusInterfaceCommonAsync,
    DbusNoReplyFlag,
    dbus_method_async,
    dbus_method_async_override,
    dbus_property_async,
    dbus_property_async_override,
    dbus_signal_async,
    get_current_message,
)

if TYPE_CHECKING:
    from sdbus.dbus_proxy_async_interfaces import (
        DBUS_PROPERTIES_CHANGED_TYPING,
    )
else:
    DBUS_PROPERTIES_CHANGED_TYPING = None


class TestPing(IsolatedDbusTestCase):

    async def test_ping_with_busctl(self) -> None:
        try:
            busctl_process = await create_subprocess_exec(
                '/usr/bin/busctl',
                '--user',
                'call',
                'org.freedesktop.DBus', '/org/freedesktop/DBus',
                'org.freedesktop.DBus.Peer', 'Ping',
            )
        except FileNotFoundError:
            raise SkipTest('busctl not installed')

        return_code = await busctl_process.wait()
        self.assertEqual(return_code, 0)

    async def test_ping(self) -> None:
        m = self.bus.new_method_call_message(
            'org.freedesktop.DBus', '/org/freedesktop/DBus',
            'org.freedesktop.DBus.Peer', 'Ping',
        )
        r = await self.bus.call_async(m)
        self.assertIsNone(r.get_contents())


class TestRequestName(IsolatedDbusTestCase):
    async def test_request_name(self) -> None:
        await self.bus.request_name_async("org.example.test", 0)


TEST_INTERFACE_NAME = "org.test.test"


class TestInterface(
    DbusInterfaceCommonAsync,
    interface_name=TEST_INTERFACE_NAME,
):

    def __init__(self) -> None:
        super().__init__()
        self.test_string = 'test_property'
        self.test_string_read = 'read'
        self.test_no_reply_string = 'no'
        self.property_private = 100
        self.no_reply_sync = Event()

    @dbus_method_async("s", "s")
    async def upper(self, string: str) -> str:
        """Uppercase the input"""
        return string.upper()

    @dbus_method_async(result_signature='s')
    async def get_sender(self) -> str:
        message = get_current_message()
        return message.sender or ''

    @dbus_method_async(result_signature='x')
    async def test_int(self) -> int:
        return 1

    @dbus_method_async(result_signature='x', result_args_names=('an_int', ))
    async def int_annotated(self) -> int:
        return 1

    @dbus_property_async(
        "s",
        flags=DbusPropertyEmitsChangeFlag,
    )
    def test_property(self) -> str:
        """Test property"""
        return self.test_string

    @test_property.setter
    def test_property_set(self, new_property: str) -> None:
        self.test_string = new_property

    @dbus_property_async("s")
    def test_property_read_only(self) -> str:
        return self.test_string_read

    @dbus_property_async("x")
    def test_property_private(self) -> int:
        return self.property_private

    @test_property_private.setter_private
    def test_private_setter(self, new_value: int) -> None:
        self.property_private = new_value

    @dbus_method_async("sb", "s")
    async def kwargs_function(
            self,
            input: str = 'test',
            is_upper: bool = True) -> str:
        if is_upper:
            return input.upper()
        else:
            return input.lower()

    @dbus_method_async("sb", "s", 0, ('string_result', ))
    async def kwargs_function_annotated(
            self,
            input: str = 'test',
            is_upper: bool = True) -> str:
        if is_upper:
            return input.upper()
        else:
            return input.lower()

    @dbus_signal_async('ss')
    def test_signal(self) -> tuple[str, str]:
        """Test signal"""
        raise NotImplementedError

    @dbus_method_async()
    async def raise_base_exception(self) -> None:
        raise DbusFailedError('Test error')

    @dbus_method_async()
    async def raise_derived_exception(self) -> None:
        raise DbusFileExistsError('Test error 2')

    @dbus_method_async()
    async def raise_custom_error(self) -> None:
        raise DbusErrorTest('Custom')

    @dbus_method_async()
    async def raise_and_unmap_error(self) -> None:
        try:
            DBUS_ERROR_TO_EXCEPTION.pop('org.example.Nothing')
        except KeyError:
            ...

        raise DbusErrorUnmappedLater('Should be unmapped')

    @dbus_method_async()
    async def raise_python_exc(self) -> None:
        raise ValueError("Test!")

    @dbus_method_async('s', flags=DbusNoReplyFlag)
    async def no_reply_method(self, new_value: str) -> None:
        self.no_reply_sync.set()

    @dbus_property_async("s")
    def test_constant_property(self) -> str:
        return "a"

    @dbus_method_async(
        result_signature='(ss)'
    )
    async def test_struct_return(self) -> tuple[str, str]:
        return ('hello', 'world')

    @dbus_method_async(
        result_signature='(ss)'
    )
    async def test_struct_return_workaround(self) -> tuple[tuple[str, str]]:
        return (('hello', 'world'), )

    @dbus_method_async()
    async def looong_method(self) -> None:
        await sleep(100)

    @dbus_signal_async()
    def empty_signal(self) -> None:
        raise NotImplementedError

    @dbus_method_async()
    async def returns_none_method(self) -> None:
        return

    @dbus_method_async(
        input_signature="(iiii)",
        result_signature="i",
    )
    async def takes_struct_method(
        self,
        int_struct: tuple[int, int, int, int],
    ) -> int:
        a, b, c, d = int_struct
        return a*b*c*d

    @dbus_method_async("s", "x")
    async def return_length(self, input_str: str) -> int:
        return len(input_str)


class DbusErrorTest(DbusFailedError):
    dbus_error_name = 'org.example.Error'


class DbusErrorUnmappedLater(DbusFailedError):
    dbus_error_name = 'org.example.Nothing'


TEST_SERVICE_NAME = 'org.example.test'


def initialize_object() -> tuple[TestInterface, TestInterface]:
    test_object = TestInterface()
    test_object.export_to_dbus('/')

    test_object_connection = TestInterface.new_proxy(
        TEST_SERVICE_NAME, '/')

    return test_object, test_object_connection


class TestProxy(IsolatedDbusTestCase):
    async def asyncSetUp(self) -> None:
        await super().asyncSetUp()
        await self.bus.request_name_async(TEST_SERVICE_NAME, 0)

    async def test_method_kwargs(self) -> None:
        test_object, test_object_connection = initialize_object()

        self.assertEqual(
            'TEST',
            await test_object_connection.kwargs_function(
                'test', True)
        )

        self.assertEqual(
            'TEST',
            await test_object_connection.kwargs_function())

        self.assertEqual(
            'test',
            await test_object_connection.kwargs_function(
                is_upper=False))

        self.assertEqual(
            'ASD',
            await test_object_connection.kwargs_function('asd'))

        self.assertEqual(
            'ASD',
            await test_object_connection.kwargs_function(input='asd'))

        self.assertEqual(
            'asd',
            await test_object_connection.kwargs_function(
                'ASD', is_upper=False))

        self.assertEqual(
            'asd',
            await test_object_connection.kwargs_function(
                input='ASD', is_upper=False))

    async def test_method(self) -> None:
        test_object, test_object_connection = initialize_object()

        test_string = 'asdafsrfgdrtuhrytuj'

        self.assertEqual(test_string.upper(),
                         await test_object.upper(test_string))

        self.assertEqual(1, await test_object_connection.test_int())

        self.assertEqual(
            test_string.upper(),
            await test_object_connection.upper(test_string))

        self.assertEqual(
            await wait_for(test_object.test_struct_return(), 0.5),
            await wait_for(test_object_connection.test_struct_return(), 0.5),
        )

        self.assertEqual(
            (await wait_for(
                test_object.test_struct_return_workaround(), 0.5))[0],
            await wait_for(
                test_object_connection.test_struct_return_workaround(), 0.5),
        )

        self.assertTrue(await test_object_connection.get_sender())

        with self.subTest("Test method that returns None"):
            self.assertIsNone(
                await test_object
                .returns_none_method()  # type: ignore[func-returns-value]
            )
            self.assertIsNone(
                await test_object_connection
                .returns_none_method()   # type: ignore[func-returns-value]
            )

        with self.subTest("Test method that takes a single struct"):
            self.assertEqual(
                await test_object.takes_struct_method((2, 3, 4, 5)),
                120,
            )
            self.assertEqual(
                await test_object_connection.takes_struct_method((9, 8, 7, 6)),
                3024,
            )

    async def test_subclass(self) -> None:
        test_object, test_object_connection = initialize_object()

        test_var = ['asdasd']

        class TestInheritence(TestInterface):
            @dbus_method_async_override()
            async def test_int(self) -> int:
                return 2

            @dbus_property_async_override()
            def test_property(self) -> str:
                return test_var[0]

            @test_property.setter
            def test_property_setter(self, var: str) -> None:
                test_var.insert(0, var)

        test_subclass = TestInheritence()

        test_subclass.export_to_dbus('/subclass', self.bus)

        self.assertEqual(await test_subclass.test_int(), 2)

        test_subclass_connection = TestInheritence.new_proxy(
            TEST_SERVICE_NAME, '/subclass', self.bus)

        self.assertEqual(await test_subclass_connection.test_int(), 2)

        self.assertEqual(test_var[0], await test_subclass.test_property)

        await test_subclass.test_property.set_async('12345')

        self.assertEqual(test_var[0], await test_subclass.test_property)
        self.assertEqual('12345', await test_subclass.test_property)

        with self.subTest('Test dbus to python mapping'):
            dbus_elements_map = (
                {
                    interface_name: meta.dbus_member_to_python_attr
                    for interface_name, meta in
                    TestInterface._dbus_iter_interfaces_meta()
                }
            )
            self.assertIn(
                "TestInt",
                dbus_elements_map[TEST_INTERFACE_NAME],
            )

            self.assertIn(
                "TestInt",
                dbus_elements_map[TEST_INTERFACE_NAME],
            )

            self.assertIn(
                "TestProperty",
                dbus_elements_map[TEST_INTERFACE_NAME],
            )

        with self.subTest('Tripple subclass'):
            class TestInheritenceTri(TestInheritence):
                @dbus_method_async_override()
                async def test_int(self) -> int:
                    return 3

                @dbus_property_async_override()
                def test_property(self) -> str:
                    return 'tri'

            test_subclass_tri = TestInheritenceTri()

            test_subclass_tri.export_to_dbus('/subclass/tri', self.bus)

            self.assertEqual(await test_subclass_tri.test_int(), 3)

            test_subclass_tri_connection = TestInheritenceTri.new_proxy(
                TEST_SERVICE_NAME, '/subclass/tri', self.bus)

            self.assertEqual(await test_subclass_tri_connection.test_int(), 3)

            self.assertEqual(await test_subclass_tri.test_property, 'tri')
            self.assertEqual(
                await test_subclass_tri_connection.test_property, 'tri')

    async def test_properties(self) -> None:
        test_object, test_object_connection = initialize_object()

        self.assertEqual(
            'test_property',
            await test_object.test_property.get_async())

        self.assertEqual(
            'test_property', await test_object.test_property)

        self.assertEqual(
            await wait_for(test_object_connection.test_property, 0.5),
            await test_object.test_property)

        self.assertEqual(
            'test_property',
            await wait_for(test_object_connection.test_property, 0.5))

        self.assertEqual(
            await test_object.test_property_read_only,
            await wait_for(
                test_object_connection.test_property_read_only, 0.5))

        new_string = 'asdsgrghdthdth'

        await wait_for(
            test_object_connection.test_property.set_async(
                new_string),
            0.5)

        self.assertEqual(
            new_string, await test_object.test_property)

        self.assertEqual(
            new_string,
            await wait_for(test_object_connection.test_property, 0.5)
        )

    async def test_signal(self) -> None:
        test_object, test_object_connection = initialize_object()

        test_tuple = ('sgfsretg', 'asd')

        async with self.assertDbusSignalEmits(
                test_object.test_signal
            ) as local_signals_record, self.assertDbusSignalEmits(
                test_object_connection.test_signal
        ) as remote_signals_record:
            test_object.test_signal.emit(test_tuple)

        async with self.assertDbusSignalEmits(
                test_object.test_signal
            ) as local_signals_record, self.assertDbusSignalEmits(
                test_object_connection.test_signal
        ) as remote_signals_record:
            test_object.test_signal.emit(test_tuple)

        self.assertEqual([test_tuple], local_signals_record.output)
        self.assertEqual([test_tuple], remote_signals_record.output)

    async def test_signal_catch_anywhere(self) -> None:
        test_object, test_object_connection = initialize_object()

        loop = get_running_loop()

        test_tuple = ('sgfsretg', 'asd')

        with self.subTest('Catch anywhere over D-Bus object'):
            async def catch_anywhere_oneshot_dbus(
            ) -> tuple[str, tuple[str, str]]:
                async for x in test_object_connection.test_signal\
                        .catch_anywhere():
                    return x

                raise RuntimeError

            catch_anywhere_over_dbus_task \
                = loop.create_task(catch_anywhere_oneshot_dbus())

            await sleep(0)

            test_object.test_signal.emit(test_tuple)

            self.assertEqual(
                ('/', test_tuple),
                await wait_for(catch_anywhere_over_dbus_task, timeout=1),
            )

        with self.subTest('Catch anywhere over D-Bus class'):
            async def catch_anywhere_oneshot_from_class(
            ) -> tuple[str, tuple[str, str]]:
                async for x in TestInterface.test_signal.catch_anywhere(
                        TEST_SERVICE_NAME, self.bus):
                    return x

                raise RuntimeError

            catch_anywhere_from_class_task \
                = loop.create_task(catch_anywhere_oneshot_from_class())

            await sleep(0)

            test_object.test_signal.emit(test_tuple)

            self.assertEqual(
                ('/', test_tuple),
                await wait_for(catch_anywhere_from_class_task, timeout=1),
            )

        with self.subTest('Catch anywhere over local object'):
            async def catch_anywhere_oneshot_local(
            ) -> tuple[str, tuple[str, str]]:
                async for x in test_object.test_signal.catch_anywhere():
                    return x

                raise RuntimeError

            catch_anywhere_over_local_task \
                = loop.create_task(catch_anywhere_oneshot_local())

            with self.assertRaises(NotImplementedError):
                await wait_for(
                    catch_anywhere_over_local_task,
                    timeout=1,
                )

    async def test_signal_multiple_readers(self) -> None:
        test_object, test_object_connection = initialize_object()

        loop = get_running_loop()

        test_tuple = ('sgfsretg', 'asd')

        async def reader_one() -> tuple[str, str]:
            async for x in test_object_connection.test_signal.catch():
                return test_tuple

            raise RuntimeError

        async def reader_two() -> tuple[str, str]:
            async for x in test_object_connection.test_signal.catch():
                return test_tuple

            raise RuntimeError

        t1 = loop.create_task(reader_one())
        t2 = loop.create_task(reader_two())

        loop.call_at(0, test_object.test_signal.emit, test_tuple)

        self.assertEqual(test_tuple, await wait_for(t1, timeout=1))
        self.assertEqual(test_tuple, await wait_for(t2, timeout=1))

    async def test_exceptions(self) -> None:
        test_object, test_object_connection = initialize_object()

        async def first_test() -> None:
            await test_object_connection.raise_base_exception()

        with self.assertRaises(DbusFailedError):
            await wait_for(first_test(), timeout=1)

        async def second_test() -> None:
            await test_object_connection.raise_derived_exception()

        with self.assertRaises(DbusFileExistsError):
            await wait_for(second_test(), timeout=1)

        async def third_test() -> None:
            await test_object_connection.raise_custom_error()

        with self.assertRaises(DbusErrorTest):
            await wait_for(third_test(), timeout=1)

        def bad_exception_error_name_used() -> None:
            class BadDbusError(DbusFailedError):
                dbus_error_name = 'org.freedesktop.DBus.Error.NoMemory'

        self.assertRaises(ValueError, bad_exception_error_name_used)

        def bad_exception_no_error_name() -> None:
            class BadDbusError(DbusFailedError):
                ...

        self.assertRaises(TypeError, bad_exception_no_error_name)

        async def test_unmapped_expection() -> None:
            await test_object_connection.raise_and_unmap_error()

        with self.assertRaises(SdBusUnmappedMessageError):
            await wait_for(test_unmapped_expection(), timeout=1)

    async def test_no_reply_method(self) -> None:
        test_object, test_object_connection = initialize_object()

        await wait_for(test_object_connection.no_reply_method('yes'),
                       timeout=1)

        await wait_for(test_object.no_reply_sync.wait(), timeout=1)

    async def test_interface_remove(self) -> None:
        test_object, test_object_connection = initialize_object()

        from gc import collect

        del test_object

        collect()

        with self.assertRaises(DbusUnknownObjectError):
            await wait_for(test_object_connection.dbus_introspect(),
                           timeout=0.2)

    def test_docstring(self) -> None:
        test_object, test_object_connection = initialize_object()

        from pydoc import getdoc

        self.assertTrue(getdoc(test_object.upper))

        self.assertTrue(getdoc(test_object_connection.upper))

        self.assertTrue(getdoc(test_object.test_property))

        self.assertTrue(
            getdoc(test_object_connection.test_property))

        self.assertTrue(getdoc(test_object.test_signal))

        self.assertTrue(
            getdoc(test_object_connection.test_signal))

    async def test_emits_properties_changed(self) -> None:
        test_object, test_object_connection = initialize_object()

        test_str = 'should_be_emited'

        loop = get_running_loop()

        async def catch_property_emit_connection() -> str:
            async for x in test_object_connection.properties_changed.catch():
                for v in x[1].values():
                    probably_str = v[1]
                    if isinstance(probably_str, str):
                        return probably_str
                    else:
                        raise TypeError
            raise ValueError

        async def catch_property_emit_local() -> str:
            async for x in test_object.properties_changed.catch():
                for v in x[1].values():
                    probably_str = v[1]
                    if isinstance(probably_str, str):
                        return probably_str
                    else:
                        raise TypeError
            raise ValueError

        t1 = loop.create_task(catch_property_emit_connection())
        t2 = loop.create_task(catch_property_emit_local())

        await test_object_connection.test_property.set_async(test_str)

        t1_result = await wait_for(t1, timeout=0.2)
        t2_result = await wait_for(t2, timeout=0.2)

        self.assertEqual(t1_result, test_str)
        self.assertEqual(t2_result, test_str)

    async def test_bus_close(self) -> None:
        test_object, test_object_connection = initialize_object()

        async def too_long_wait() -> None:
            await test_object_connection.looong_method()

        self.bus.close()

        with self.assertRaises(SdBusLibraryError):
            await wait_for(too_long_wait(), timeout=1)

    async def test_bus_timerfd(self) -> None:
        test_object, test_object_connection = initialize_object()

        self.bus.method_call_timeout_usec = 10_000  # 0.01 seconds

        loop = get_running_loop()

        start = loop.time()

        with self.assertRaises(DbusNoReplyError):
            await wait_for(test_object_connection.looong_method(), timeout=1)

        self.assertAlmostEqual(loop.time() - start, 0.01, delta=0.01)

    async def test_signal_queue_wildcard_match(self) -> None:
        test_object, test_object_connection = initialize_object()

        loop = get_running_loop()
        future = loop.create_future()

        slot = await self.bus.match_signal_async(
            TEST_SERVICE_NAME,
            None, None, None,
            future.set_result)

        try:
            test_object.test_signal.emit(('test', 'signal'))

            await wait_for(future, timeout=1)
            message = future.result()
            self.assertEqual(message.member, "TestSignal")
        finally:
            slot.close()

    async def test_class_with_string_subclass_parameter(self) -> None:
        from enum import Enum

        class InterfaceNameEnum(str, Enum):
            FOO = 'org.example.foo'
            BAR = 'org.example.bar'

        class ObjectPathEnum(str, Enum):
            FOO = '/foo'
            BAR = '/bar'

        class EnumedInterfaceAsync(
            DbusInterfaceCommonAsync,
            interface_name=InterfaceNameEnum.BAR,
        ):

            @dbus_property_async('s')
            def hello_world(self) -> str:
                return 'Hello World!'

        test_object = EnumedInterfaceAsync()
        test_object.export_to_dbus(ObjectPathEnum.FOO)

    async def test_properties_get_all_dict(self) -> None:
        test_object, test_object_connection = initialize_object()

        dbus_dict = await test_object_connection._properties_get_all(
            TEST_INTERFACE_NAME)

        self.assertEqual(
            await test_object.test_property,
            dbus_dict['TestProperty'][1],
        )

        self.assertEqual(
            await test_object.test_property,
            (
                await test_object_connection.properties_get_all_dict()
            )['test_property'],
        )

    async def test_empty_signal(self) -> None:
        test_object, test_object_connection = initialize_object()

        async with self.assertDbusSignalEmits(
                test_object.empty_signal
            ) as local_signals_record, self.assertDbusSignalEmits(
                test_object_connection.empty_signal
        ) as remote_signals_record:
            test_object.empty_signal.emit(None)

        self.assertEqual([None], local_signals_record.output)
        self.assertEqual([None], remote_signals_record.output)

    async def test_properties_changed(self) -> None:
        test_object, test_object_connection = initialize_object()

        test_str = 'should_be_emited'

        async with self.assertDbusSignalEmits(
            test_object_connection.properties_changed
        ) as properties_changed_catch:
            await test_object_connection.test_property.set_async(test_str)

        properties_changed_data = properties_changed_catch.output[0]

        parsed_dict_from_class = parse_properties_changed(
            TestInterface, properties_changed_data)
        self.assertEqual(
            test_str,
            parsed_dict_from_class['test_property'],
        )

        parsed_dict_from_object = parse_properties_changed(
            test_object_connection, properties_changed_data)
        self.assertEqual(
            test_str,
            parsed_dict_from_object['test_property'],
        )

        properties_changed_data[2].append('invalidated_property')
        parsed_dict_with_invalidation = parse_properties_changed(
            test_object, properties_changed_data,
            on_unknown_member='reuse',
        )
        self.assertIsNone(
            parsed_dict_with_invalidation['invalidated_property'])

    async def test_property_private_setter(self) -> None:
        test_object, test_object_connection = initialize_object()

        new_value = 200
        self.assertNotEqual(
            await test_object_connection.test_property_private,
            new_value
        )

        with self.assertRaises(DbusPropertyReadOnlyError):
            await test_object_connection.test_property_private.set_async(
                new_value)

        async with self.assertDbusSignalEmits(
            test_object_connection.properties_changed
        ) as properties_changed_catch:
            await test_object.test_property_private.set_async(new_value)

        changed_properties = properties_changed_catch.output[0]

        self.assertEqual(
            await test_object_connection.test_property_private,
            new_value
        )

        self.assertIn('TestPropertyPrivate', changed_properties[1])

    async def test_property_override_setter_private(self) -> None:

        test_int = 1

        class TestInterfacePrivateSetter(TestInterface):
            @dbus_property_async_override()
            def test_property_private(self) -> int:
                return test_int

            @test_property_private.setter_private
            def _private_setter(self, new_value: int) -> None:
                nonlocal test_int
                test_int = new_value

        test_object = TestInterfacePrivateSetter()
        test_object.export_to_dbus('/')
        test_object_connection = TestInterface.new_proxy(
            TEST_SERVICE_NAME, '/')

        self.assertEqual(
            await test_object_connection.test_property_private,
            test_int,
        )

        async def catch_properties_changed() -> int:
            async for x in test_object_connection.properties_changed:
                changed_attr = parse_properties_changed(
                    TestInterface, x)["test_property_private"]

                if not isinstance(changed_attr, int):
                    raise TypeError

                return changed_attr

            raise RuntimeError

        catch_changed_task = get_running_loop(
        ).create_task(catch_properties_changed())

        with self.assertRaises(DbusPropertyReadOnlyError):
            await test_object_connection.test_property_private.set_async(10)

        await test_object.test_property_private.set_async(10)

        self.assertEqual(
            await test_object_connection.test_property_private,
            10,
        )

        self.assertEqual(
            await test_object.test_property_private,
            test_int,
        )

        self.assertEqual(
            await wait_for(catch_changed_task, timeout=1),
            10,
        )

    async def test_interface_composition(self) -> None:
        class OneInterface(
            DbusInterfaceCommonAsync,
            interface_name="org.example.one",
        ):
            @dbus_method_async(result_signature="x")
            async def one(self) -> int:
                raise NotImplementedError

        class TwoInterface(
            DbusInterfaceCommonAsync,
            interface_name="org.example.two",
        ):
            @dbus_method_async(result_signature="t")
            async def two(self) -> int:
                return 2

        class CombinedInterface(TwoInterface, OneInterface):
            ...

        test_combined = CombinedInterface()
        test_combined_interfaces = [
            iface for iface, _ in test_combined._dbus_iter_interfaces_meta()
        ]

        # Verify the order of reported interfaces on the combined class.
        self.assertEqual(test_combined_interfaces, [
            "org.freedesktop.DBus.Peer",
            "org.freedesktop.DBus.Introspectable",
            "org.freedesktop.DBus.Properties",
            "org.example.one",
            "org.example.two",
        ])

    async def test_extremely_large_string(self) -> None:
        test_object, test_object_connection = initialize_object()

        extremely_large_string = "a" * 8423681

        remote_len = await wait_for(
            test_object_connection.return_length(
                extremely_large_string
            ),
            timeout=10,
        )

        self.assertEqual(
            remote_len,
            len(extremely_large_string),
        )

        # Check that calling regular methods still works.
        for _ in range(5):
            await test_object_connection.returns_none_method()

    async def test_export_handle(self) -> None:
        test_object = TestInterface()
        test_object_connection = TestInterface.new_proxy(
            TEST_SERVICE_NAME, '/',
        )
        with self.assertRaises(DbusUnknownObjectError):
            await test_object_connection.returns_none_method()

        with test_object.export_to_dbus("/"):
            await test_object_connection.returns_none_method()

        with self.assertRaises(DbusUnknownObjectError):
            await test_object_connection.returns_none_method()

        test_object2 = TestInterface()
        handle = test_object2.export_to_dbus("/")
        await test_object_connection.returns_none_method()
        handle.stop()

        with self.assertRaises(DbusUnknownObjectError):
            await test_object_connection.returns_none_method()

    def test_asyncio_run_different_loops(self) -> None:
        bus = self.bus

        async def test() -> None:
            dbus_object = DbusInterfaceCommonAsync.new_proxy(
                "org.freedesktop.DBus",
                "/org/freedesktop/DBus",
                bus,
            )
            await wait_for(dbus_object.dbus_ping(), timeout=1)

        with self.assertRaisesRegex(RuntimeError, "different loop"):
            asyncio_run(test())

    async def test_python_exc(self) -> None:
        test_object, test_object_connection = initialize_object()

        with self.assertRaisesRegex(ValueError, "Test!"):
            await test_object.raise_python_exc()

        with self.assertRaisesRegex(ValueError, "Test!"):
            await test_object_connection.raise_python_exc()

    async def test_empty_dbus_interface(self) -> None:
        class Empty(
            DbusInterfaceCommonAsync,
            interface_name="org.empty",
        ):
            ...

        empty_local = Empty()
        empty_local.export_to_dbus("/")
        empty_proxy = Empty.new_proxy(TEST_SERVICE_NAME, "/")

        intro = await empty_proxy.dbus_introspect()
        self.assertIn('<interface name="org.empty">', intro)