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

# Copyright (C) 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 get_running_loop, sleep, wait_for
from unittest import main

from sdbus.exceptions import (
    SdBusLibraryError,
    SdBusRequestNameAlreadyOwnerError,
    SdBusRequestNameError,
    SdBusRequestNameExistsError,
    SdBusRequestNameInQueueError,
)
from sdbus.sd_bus_internals import NameAllowReplacementFlag, NameQueueFlag
from sdbus.unittest import IsolatedDbusTestCase
from sdbus_async.dbus_daemon import FreedesktopDbus

from sdbus import (
    request_default_bus_name,
    request_default_bus_name_async,
    sd_bus_open_user,
)

TEST_BUS_NAME = 'com.example.test'
TEST_BUS_NAME_regex_match = TEST_BUS_NAME.replace('.', r'\.')


class TestRequestNameLowLevel(IsolatedDbusTestCase):
    def test_request_name_exception_tree(self) -> None:
        # Test that SdBusRequestNameError is super class
        # of other request name exceptions
        self.assertTrue(
            issubclass(
                SdBusRequestNameAlreadyOwnerError,
                SdBusRequestNameError,
            )
        )
        self.assertTrue(
            issubclass(
                SdBusRequestNameExistsError,
                SdBusRequestNameError,
            )
        )
        self.assertTrue(
            issubclass(
                SdBusRequestNameInQueueError,
                SdBusRequestNameError,
            )
        )
        # Test the opposite
        self.assertFalse(
            issubclass(
                SdBusRequestNameAlreadyOwnerError,
                SdBusRequestNameExistsError,
            )
        )
        self.assertFalse(
            issubclass(
                SdBusRequestNameInQueueError,
                SdBusRequestNameExistsError,
            )
        )
        self.assertFalse(
            issubclass(
                SdBusRequestNameInQueueError,
                SdBusRequestNameAlreadyOwnerError,
            )
        )

    async def test_name_exists_async(self) -> None:
        extra_bus = sd_bus_open_user()
        await self.bus.request_name_async(TEST_BUS_NAME, 0)

        with self.assertRaises(SdBusRequestNameExistsError):
            await wait_for(
                extra_bus.request_name_async(TEST_BUS_NAME, 0),
                timeout=1,
            )

    async def test_name_already_async(self) -> None:
        await self.bus.request_name_async(TEST_BUS_NAME, 0)

        with self.assertRaises(SdBusRequestNameAlreadyOwnerError):
            await wait_for(
                self.bus.request_name_async(TEST_BUS_NAME, 0),
                timeout=1,
            )

    async def test_name_queued_async(self) -> None:
        extra_bus = sd_bus_open_user()
        await self.bus.request_name_async(TEST_BUS_NAME, 0)

        with self.assertRaises(SdBusRequestNameInQueueError):
            await wait_for(
                extra_bus.request_name_async(TEST_BUS_NAME, NameQueueFlag),
                timeout=1,
            )

    async def test_name_other_error_async(self) -> None:
        extra_bus = sd_bus_open_user()
        extra_bus.close()

        with self.assertRaises(SdBusLibraryError):
            await wait_for(
                extra_bus.request_name_async(TEST_BUS_NAME, 0),
                timeout=1,
            )

    def test_name_exists_block(self) -> None:
        extra_bus = sd_bus_open_user()
        self.bus.request_name(TEST_BUS_NAME, 0)

        with self.assertRaisesRegex(
            SdBusRequestNameExistsError,
            TEST_BUS_NAME_regex_match,
        ):
            extra_bus.request_name(TEST_BUS_NAME, 0)

    def test_name_already_block(self) -> None:
        self.bus.request_name(TEST_BUS_NAME, 0)

        with self.assertRaisesRegex(
            SdBusRequestNameAlreadyOwnerError,
            TEST_BUS_NAME_regex_match,
        ):
            self.bus.request_name(TEST_BUS_NAME, 0)

    def test_name_queued_block(self) -> None:
        extra_bus = sd_bus_open_user()
        self.bus.request_name(TEST_BUS_NAME, 0)

        with self.assertRaisesRegex(
            SdBusRequestNameInQueueError,
            TEST_BUS_NAME_regex_match,
        ):
            extra_bus.request_name(TEST_BUS_NAME, NameQueueFlag)

    def test_name_other_error_block(self) -> None:
        extra_bus = sd_bus_open_user()
        extra_bus.close()
        with self.assertRaises(SdBusLibraryError):
            extra_bus.request_name(TEST_BUS_NAME, 0)


class TestRequestNameBlock(IsolatedDbusTestCase):
    def test_request_name_replacement(self) -> None:
        extra_bus = sd_bus_open_user()
        extra_bus.request_name(TEST_BUS_NAME, NameAllowReplacementFlag)

        with self.assertRaises(SdBusRequestNameExistsError):
            request_default_bus_name(TEST_BUS_NAME)

        request_default_bus_name(
            TEST_BUS_NAME,
            replace_existing=True,
        )


class TestRequestNameAsync(IsolatedDbusTestCase):
    async def test_request_name_replacement(self) -> None:
        extra_bus = sd_bus_open_user()
        await extra_bus.request_name_async(
            TEST_BUS_NAME,
            NameAllowReplacementFlag,
        )

        with self.assertRaises(SdBusRequestNameExistsError):
            await request_default_bus_name_async(TEST_BUS_NAME)

        await request_default_bus_name_async(
            TEST_BUS_NAME,
            replace_existing=True,
        )

    async def test_request_name_queue(self) -> None:
        extra_bus = sd_bus_open_user()
        await extra_bus.request_name_async(TEST_BUS_NAME, 0)

        with self.assertRaises(SdBusRequestNameInQueueError):
            await request_default_bus_name_async(
                TEST_BUS_NAME,
                queue=True,
            )

        async def catch_owner_changed() -> str:
            dbus = FreedesktopDbus()
            async for name, old, new in dbus.name_owner_changed:
                if name != TEST_BUS_NAME:
                    continue

                if old and new:
                    return new

            raise RuntimeError

        loop = get_running_loop()
        owner_changed_task = loop.create_task(catch_owner_changed())
        await sleep(0)

        extra_bus.close()

        await wait_for(owner_changed_task, timeout=0.5)

        with self.assertRaises(SdBusRequestNameAlreadyOwnerError):
            await request_default_bus_name_async(TEST_BUS_NAME)


if __name__ == '__main__':
    main()