1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, "..")
from androidtv import constants
from androidtv.firetv.firetv_async import FireTVAsync
from . import async_patchers
from .async_wrapper import awaiter
from .patchers import patch_calls
class TestFireTVAsyncPython(unittest.TestCase):
ADB_ATTR = "_adb"
PATCH_KEY = "python"
@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.ftv = FireTVAsync("HOST", 5555)
await self.ftv.adb_connect()
@awaiter
async def test_turn_on_off(self):
"""Test that the ``FireTVAsync.turn_on`` and ``FireTVAsync.turn_off`` methods work correctly."""
with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell("")[self.PATCH_KEY]:
await self.ftv.turn_on()
self.assertEqual(
getattr(self.ftv._adb, self.ADB_ATTR).shell_cmd,
constants.CMD_SCREEN_ON
+ " || (input keyevent {0} && input keyevent {1})".format(constants.KEY_POWER, constants.KEY_HOME),
)
await self.ftv.turn_off()
self.assertEqual(
getattr(self.ftv._adb, self.ADB_ATTR).shell_cmd,
constants.CMD_SCREEN_ON + " && input keyevent {0}".format(constants.KEY_SLEEP),
)
@awaiter
async def test_send_intent(self):
"""Test that the ``_send_intent`` method works correctly."""
with async_patchers.patch_shell("output\r\nretcode")[self.PATCH_KEY]:
result = await self.ftv._send_intent("TEST", constants.INTENT_LAUNCH_FIRETV)
self.assertEqual(
getattr(self.ftv._adb, self.ADB_ATTR).shell_cmd,
"monkey -p TEST -c android.intent.category.LAUNCHER 1; echo $?",
)
self.assertDictEqual(result, {"output": "output", "retcode": "retcode"})
with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(None)[self.PATCH_KEY]:
result = await self.ftv._send_intent("TEST", constants.INTENT_LAUNCH_FIRETV)
self.assertEqual(
getattr(self.ftv._adb, self.ADB_ATTR).shell_cmd,
"monkey -p TEST -c android.intent.category.LAUNCHER 1; echo $?",
)
self.assertDictEqual(result, {})
@awaiter
async def test_launch_app_stop_app(self):
"""Test that the ``FireTVAsync.launch_app`` and ``FireTVAsync.stop_app`` methods work correctly."""
with async_patchers.patch_shell("")[self.PATCH_KEY]:
await self.ftv.launch_app("TEST")
self.assertEqual(
getattr(self.ftv._adb, self.ADB_ATTR).shell_cmd, constants.CMD_LAUNCH_APP_FIRETV.format("TEST")
)
await self.ftv.stop_app("TEST")
self.assertEqual(getattr(self.ftv._adb, self.ADB_ATTR).shell_cmd, "am force-stop 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.ftv, self.ftv._running_apps) as patched:
await self.ftv.running_apps()
assert patched.called
@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.ftv, self.ftv.screen_on_awake_wake_lock_size
) as screen_on_awake_wake_lock_size, patch_calls(
self.ftv, self.ftv.current_app_media_session_state
) as current_app_media_session_state, patch_calls(
self.ftv, self.ftv.running_apps
) as running_apps, patch_calls(
self.ftv, self.ftv.get_hdmi_input
) as get_hdmi_input:
await self.ftv.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.ftv, self.ftv.screen_on_awake_wake_lock_size
) as screen_on_awake_wake_lock_size, patch_calls(
self.ftv, self.ftv.current_app_media_session_state
) as current_app_media_session_state, patch_calls(
self.ftv, self.ftv.running_apps
) as running_apps, patch_calls(
self.ftv, self.ftv.get_hdmi_input
) as get_hdmi_input:
await self.ftv.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.ftv, self.ftv.screen_on_awake_wake_lock_size
) as screen_on_awake_wake_lock_size, patch_calls(
self.ftv, self.ftv.current_app_media_session_state
) as current_app_media_session_state, patch_calls(
self.ftv, self.ftv.running_apps
) as running_apps, patch_calls(
self.ftv, self.ftv.get_hdmi_input
) as get_hdmi_input:
await self.ftv.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.ftv, self.ftv.get_properties) as get_properties:
await self.ftv.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.ftv, self.ftv.get_properties) as patched:
await self.ftv.update()
assert patched.called
class TestFireTVAsyncServer(TestFireTVAsyncPython):
ADB_ATTR = "_adb_device"
PATCH_KEY = "server"
@awaiter
async def setUp(self):
with async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell("")[self.PATCH_KEY]:
self.ftv = FireTVAsync("HOST", 5555, adb_server_ip="ADB_SERVER_IP")
await self.ftv.adb_connect()
if __name__ == "__main__":
unittest.main()
|
