test_androidtv_async.py

import asyncio
import sys
import unittest
from unittest.mock import patch


sys.path.insert(0, "..")

from androidtv import constants
from androidtv.androidtv.androidtv_async import AndroidTVAsync

from . import async_patchers
from .async_wrapper import awaiter
from .patchers import patch_calls


STREAM_MUSIC_EMPTY = "- STREAM_MUSIC:\n \n- STREAM"

STREAM_MUSIC_OFF = """- STREAM_MUSIC:
   Muted: false
   Min: 0
   Max: 60
   Current: 2 (speaker): 20, 40000 (hmdi_arc): 27, 40000000 (default): 15
   Devices: speaker
- STREAM_ALARM:
   Muted: true
   Min: 0
   Max: 7
   Current: 2 (speaker): 3, 40000 (hmdi_arc): 3, 40000000 (default): 2
   Devices: speaker"""

STREAM_MUSIC_NO_VOLUME = """- STREAM_MUSIC:
   Muted: false
   Min: 0
   Max: 60
   Devices: speaker
- STREAM_ALARM:
   Muted: true
   Min: 0
   Max: 7
   Current: 2 (speaker): 3, 40000 (hmdi_arc): 3, 40000000 (default): 2
   Devices: speaker"""


STREAM_MUSIC_ON = """- STREAM_MUSIC:
   Muted: false
   Min: 0
   Max: 60
   Current: 2 (speaker): 20, 40000 (hmdi_arc): 22, 40000000 (default): 15
   Devices: hmdi_arc
- STREAM_ALARM:
   Muted: false
   Min: 0
   Max: 7
   Current: 2 (speaker): 3, 40000 (hmdi_arc): 3, 40000000 (default): 2
   Devices: speaker"""


class TestAndroidTVAsyncPython(unittest.TestCase):
    PATCH_KEY = "python"
    ADB_ATTR = "_adb"

    @awaiter
    async def setUp(self):
        with (
            async_patchers.PATCH_ADB_DEVICE_TCP,
            async_patchers.patch_connect(True)[self.PATCH_KEY],
            async_patchers.patch_shell("")[self.PATCH_KEY],
        ):
            self.atv = AndroidTVAsync("HOST", 5555)
            await self.atv.adb_connect()

    @awaiter
    async def test_turn_on_off(self):
        """Test that the ``AndroidTVAsync.turn_on`` and ``AndroidTVAsync.turn_off`` methods work correctly."""
        with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell("")[self.PATCH_KEY]:
            await self.atv.turn_on()
            self.assertEqual(
                getattr(self.atv._adb, self.ADB_ATTR).shell_cmd,
                constants.CMD_SCREEN_ON + " || input keyevent {0}".format(constants.KEY_POWER),
            )

            await self.atv.turn_off()
            self.assertEqual(
                getattr(self.atv._adb, self.ADB_ATTR).shell_cmd,
                constants.CMD_SCREEN_ON + " && input keyevent {0}".format(constants.KEY_POWER),
            )

    @awaiter
    async def test_start_intent(self):
        """Test that the ``start_intent`` method works correctly."""
        with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell("")[self.PATCH_KEY]:
            await self.atv.start_intent("TEST")
            self.assertEqual(
                getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "am start -a android.intent.action.VIEW -d TEST"
            )

    @awaiter
    async def test_running_apps(self):
        """Check that the ``running_apps`` property works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            with patch_calls(self.atv, self.atv._running_apps) as patched:
                await self.atv.running_apps()
                assert patched.called

    @awaiter
    async def test_stream_music_properties(self):
        """Check that the ``stream_music_properties`` method works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            with (
                patch_calls(self.atv, self.atv._audio_output_device) as audio_output_device,
                patch_calls(self.atv, self.atv._is_volume_muted) as is_volume_muted,
                patch_calls(self.atv, self.atv._volume) as volume,
                patch_calls(self.atv, self.atv._volume_level) as volume_level,
            ):
                await self.atv.stream_music_properties()
                assert audio_output_device.called
                assert is_volume_muted.called
                assert volume.called
                assert volume_level.called

            with patch_calls(self.atv, self.atv._audio_output_device) as audio_output_device:
                await self.atv.audio_output_device()
                assert audio_output_device.called

            with patch_calls(self.atv, self.atv._is_volume_muted) as is_volume_muted:
                await self.atv.is_volume_muted()
                assert is_volume_muted.called

            with patch_calls(self.atv, self.atv._volume) as volume:
                await self.atv.volume()
                assert volume.called

            with patch_calls(self.atv, self.atv._volume_level) as volume_level:
                await self.atv.volume_level()
                assert volume_level.called

    @awaiter
    async def test_set_volume_level(self):
        """Check that the ``set_volume_level`` method works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            new_volume_level = await self.atv.set_volume_level(0.5)
            self.assertIsNone(new_volume_level)

        with async_patchers.patch_shell("")[self.PATCH_KEY]:
            new_volume_level = await self.atv.set_volume_level(0.5)
            self.assertIsNone(new_volume_level)

        with async_patchers.patch_shell(STREAM_MUSIC_ON)[self.PATCH_KEY]:
            new_volume_level = await self.atv.set_volume_level(0.5)
            self.assertEqual(new_volume_level, 0.5)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "media volume --show --stream 3 --set 30")

        with async_patchers.patch_shell("")[self.PATCH_KEY]:
            new_volume_level = await self.atv.set_volume_level(30.0 / 60)
            self.assertEqual(new_volume_level, 0.5)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "media volume --show --stream 3 --set 30")

        with async_patchers.patch_shell("")[self.PATCH_KEY]:
            new_volume_level = await self.atv.set_volume_level(22.0 / 60)
            self.assertEqual(new_volume_level, 22.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "media volume --show --stream 3 --set 22")

    @awaiter
    async def test_volume_up(self):
        """Check that the ``volume_up`` method works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_up()
            self.assertIsNone(new_volume_level)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 24")

        with async_patchers.patch_shell("")[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_up()
            self.assertIsNone(new_volume_level)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 24")

        with async_patchers.patch_shell(STREAM_MUSIC_ON)[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_up()
            self.assertEqual(new_volume_level, 23.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 24")
            new_volume_level = await self.atv.volume_up(23.0 / 60)
            self.assertEqual(new_volume_level, 24.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 24")

        with async_patchers.patch_shell(STREAM_MUSIC_OFF)[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_up()
            self.assertEqual(new_volume_level, 21.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 24")
            new_volume_level = await self.atv.volume_up(21.0 / 60)
            self.assertEqual(new_volume_level, 22.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 24")

    @awaiter
    async def test_volume_down(self):
        """Check that the ``volume_down`` method works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_down()
            self.assertIsNone(new_volume_level)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 25")

        with async_patchers.patch_shell("")[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_down()
            self.assertIsNone(new_volume_level)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 25")

        with async_patchers.patch_shell(STREAM_MUSIC_ON)[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_down()
            self.assertEqual(new_volume_level, 21.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 25")
            new_volume_level = await self.atv.volume_down(21.0 / 60)
            self.assertEqual(new_volume_level, 20.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 25")

        with async_patchers.patch_shell(STREAM_MUSIC_OFF)[self.PATCH_KEY]:
            new_volume_level = await self.atv.volume_down()
            self.assertEqual(new_volume_level, 19.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 25")
            new_volume_level = await self.atv.volume_down(19.0 / 60)
            self.assertEqual(new_volume_level, 18.0 / 60)
            self.assertEqual(getattr(self.atv._adb, self.ADB_ATTR).shell_cmd, "input keyevent 25")

    @awaiter
    async def test_get_properties(self):
        """Check that ``get_properties()`` works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            with (
                patch_calls(self.atv, self.atv.screen_on_awake_wake_lock_size) as screen_on_awake_wake_lock_size,
                patch_calls(self.atv, self.atv.current_app_media_session_state) as current_app_media_session_state,
                patch_calls(self.atv, self.atv.stream_music_properties) as stream_music_properties,
                patch_calls(self.atv, self.atv.running_apps) as running_apps,
                patch_calls(self.atv, self.atv.get_hdmi_input) as get_hdmi_input,
            ):
                await self.atv.get_properties(lazy=True)
                assert screen_on_awake_wake_lock_size.called
                assert not current_app_media_session_state.called
                assert not running_apps.called
                assert not get_hdmi_input.called

            with (
                patch_calls(self.atv, self.atv.screen_on_awake_wake_lock_size) as screen_on_awake_wake_lock_size,
                patch_calls(self.atv, self.atv.current_app_media_session_state) as current_app_media_session_state,
                patch_calls(self.atv, self.atv.stream_music_properties) as stream_music_properties,
                patch_calls(self.atv, self.atv.running_apps) as running_apps,
                patch_calls(self.atv, self.atv.get_hdmi_input) as get_hdmi_input,
            ):
                await self.atv.get_properties(lazy=False, get_running_apps=True)
                assert screen_on_awake_wake_lock_size.called
                assert current_app_media_session_state.called
                assert running_apps.called
                assert get_hdmi_input.called

            with (
                patch_calls(self.atv, self.atv.screen_on_awake_wake_lock_size) as screen_on_awake_wake_lock_size,
                patch_calls(self.atv, self.atv.current_app_media_session_state) as current_app_media_session_state,
                patch_calls(self.atv, self.atv.stream_music_properties) as stream_music_properties,
                patch_calls(self.atv, self.atv.running_apps) as running_apps,
                patch_calls(self.atv, self.atv.get_hdmi_input) as get_hdmi_input,
            ):
                await self.atv.get_properties(lazy=False, get_running_apps=False)
                assert screen_on_awake_wake_lock_size.called
                assert current_app_media_session_state.called
                assert not running_apps.called
                assert get_hdmi_input.called

    @awaiter
    async def test_get_properties_dict(self):
        """Check that ``get_properties_dict()`` works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            with patch_calls(self.atv, self.atv.get_properties) as get_properties:
                await self.atv.get_properties_dict()
                assert get_properties.called

    @awaiter
    async def test_update(self):
        """Check that the ``update`` method works correctly."""
        with async_patchers.patch_shell(None)[self.PATCH_KEY]:
            with patch_calls(self.atv, self.atv._update) as patched:
                await self.atv.update()
                assert patched.called


class TestAndroidTVAsyncServer(TestAndroidTVAsyncPython):
    PATCH_KEY = "server"
    ADB_ATTR = "_adb_device"

    @awaiter
    async def setUp(self):
        with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell("")[self.PATCH_KEY]:
            self.atv = AndroidTVAsync("HOST", 5555, adb_server_ip="ADB_SERVER_IP")
            await self.atv.adb_connect()


if __name__ == "__main__":
    unittest.main()