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
|
"""Tests for the ClientAsync and AdbDeviceUsbAsync classes."""
import unittest
from unittest.mock import patch
try:
from adb_shell.transport.usb_transport import UsbTransport
except (ImportError, OSError):
UsbTransport = None
from androidtv.adb_manager.adb_manager_async import AdbDeviceUsbAsync, ClientAsync
from .async_wrapper import awaiter
from . import patchers
class TestAsyncClientDevice(unittest.TestCase):
"""Test the ``ClientAsync`` and ``DeviceAsync`` classes defined in ``adb_manager_async.py``.
These tests can be removed once true async support for using an ADB server is available.
"""
@awaiter
async def test_async_client_device(self):
"""Test the ClientAsync class."""
with patch("androidtv.adb_manager.adb_manager_async.Client", patchers.ClientFakeSuccess):
client = ClientAsync("host", "port")
device = await client.device("serial")
with patch("{}.DeviceFake.shell".format(patchers.__name__)):
await device.shell("test")
with patch("{}.DeviceFake.push".format(patchers.__name__)):
await device.push("local_path", "device_path")
with patch("{}.DeviceFake.pull".format(patchers.__name__)):
await device.pull("device_path", "local_path")
with patch("{}.DeviceFake.screencap".format(patchers.__name__)):
await device.screencap()
@awaiter
async def test_async_client_device_fail(self):
"""Test the ClientAsync class when it fails."""
with patch("androidtv.adb_manager.adb_manager_async.Client", patchers.ClientFakeFail):
client = ClientAsync("host", "port")
device = await client.device("serial")
self.assertFalse(device)
@unittest.skipIf(UsbTransport is None, "UsbTransport cannot be imported")
class TestAsyncUsb(unittest.TestCase):
"""Test the ``AdbDeviceUsbAsync`` class defined in ``adb_manager_async.py``.
These tests can be removed once true async support for using a USB connection is available.
"""
@awaiter
async def test_async_usb(self):
"""Test the AdbDeviceUsbAsync class."""
with patch("adb_shell.adb_device.UsbTransport.find_adb", return_value=UsbTransport("device", "setting")):
device = AdbDeviceUsbAsync()
self.assertFalse(device.available)
with patch("androidtv.adb_manager.adb_manager_async.AdbDeviceUsb.connect") as connect:
await device.connect()
assert connect.called
with patch("androidtv.adb_manager.adb_manager_async.AdbDeviceUsb.shell") as shell:
await device.shell("test")
assert shell.called
with patch("androidtv.adb_manager.adb_manager_async.AdbDeviceUsb.push") as push:
await device.push("local_path", "device_path")
assert push.called
with patch("androidtv.adb_manager.adb_manager_async.AdbDeviceUsb.pull") as pull:
await device.pull("device_path", "local_path")
assert pull.called
with patch("androidtv.adb_manager.adb_manager_async.AdbDeviceUsb.close") as close:
await device.close()
assert close.called
|
