device_target_test.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236
  1. #!/usr/bin/env vpython3
  2. # Copyright 2021 The Chromium Authors. All rights reserved.
  3. # Use of this source code is governed by a BSD-style license that can be
  4. # found in the LICENSE file.
  5. """Tests scenarios with number of devices and invalid devices"""
  6. import common
  7. import os
  8. import subprocess
  9. import time
  10. import unittest
  11. import unittest.mock as mock
  12. from argparse import Namespace
  13. from device_target import DeviceTarget
  14. from ermine_ctl import ErmineCtl
  15. from ffx_session import FfxRunner, FfxTarget
  16. from target import Target, FuchsiaTargetException
  17. @mock.patch.object(FfxRunner, 'daemon_stop')
  18. class TestDiscoverDeviceTarget(unittest.TestCase):
  19. def setUp(self):
  20. self.args = Namespace(out_dir='out/fuchsia',
  21. target_cpu='x64',
  22. host=None,
  23. node_name=None,
  24. port=None,
  25. ssh_config='mock_config',
  26. fuchsia_out_dir=None,
  27. os_check='ignore',
  28. logs_dir=None,
  29. system_image_dir=None)
  30. def testUnspecifiedNodeNameOneDeviceReturnNoneCheckNameAndAddress(
  31. self, mock_daemon_stop):
  32. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  33. mock.patch.object(FfxRunner, 'list_targets') as mock_list_targets, \
  34. mock.patch.object(
  35. FfxTarget, 'get_ssh_address') as mock_get_ssh_address, \
  36. mock.patch.object(
  37. DeviceTarget, '_ConnectToTarget') as mock_connecttotarget, \
  38. mock.patch.object(
  39. DeviceTarget, '_Login') as mock_login:
  40. mock_list_targets.return_value = [{
  41. "nodename": "device_name",
  42. "rcs_state": "Y",
  43. "serial": "<unknown>",
  44. "target_type": "terminal.qemu-x64",
  45. "target_state": "Product",
  46. }]
  47. mock_get_ssh_address.return_value = ('address', 12345)
  48. mock_connecttotarget.return_value = True
  49. self.assertIsNone(device_target_instance.Start())
  50. self.assertEqual(device_target_instance._host, 'address')
  51. self.assertEqual(device_target_instance._port, 12345)
  52. mock_daemon_stop.assert_called_once()
  53. def testUnspecifiedNodeNameOneUnknownDeviceReturnNoneCheckAddressAndPort(
  54. self, mock_daemon_stop):
  55. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  56. mock.patch.object(FfxRunner, 'list_targets') as mock_list_targets, \
  57. mock.patch.object(
  58. FfxTarget, 'get_ssh_address') as mock_get_ssh_address, \
  59. mock.patch.object(
  60. DeviceTarget, '_ConnectToTarget') as mock_connecttotarget, \
  61. mock.patch.object(
  62. DeviceTarget, '_Login') as mock_login:
  63. mock_list_targets.return_value = [{
  64. "nodename": "<unknown>",
  65. "rcs_state": "Y",
  66. "serial": "<unknown>",
  67. "target_type": "terminal.qemu-x64",
  68. "target_state": "Product",
  69. "addresses": ["address"]
  70. }]
  71. mock_get_ssh_address.return_value = ('address', 12345)
  72. mock_connecttotarget.return_value = True
  73. self.assertIsNone(device_target_instance.Start())
  74. self.assertEqual(device_target_instance._host, 'address')
  75. self.assertEqual(device_target_instance._port, 12345)
  76. mock_login.assert_called_once()
  77. mock_daemon_stop.assert_called_once()
  78. def testUnspecifiedNodeNameTwoDevicesRaiseExceptionAmbiguousTarget(
  79. self, mock_daemon_stop):
  80. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  81. mock.patch.object(FfxRunner, 'list_targets') as mock_list_targets, \
  82. mock.patch.object(
  83. FfxTarget, 'get_ssh_address') as mock_get_ssh_address, \
  84. self.assertRaisesRegex(Exception, \
  85. 'More than one device was discovered'):
  86. mock_get_ssh_address.return_value = ('address', 12345)
  87. mock_list_targets.return_value = [{
  88. "nodename": "device_name1",
  89. "rcs_state": "Y",
  90. "serial": "<unknown>",
  91. "target_type": "terminal.qemu-x64",
  92. "target_state": "Product",
  93. "addresses": ["address1"]
  94. }, {
  95. "nodename": "device_name2",
  96. "rcs_state": "Y",
  97. "serial": "<unknown>",
  98. "target_type": "terminal.qemu-x64",
  99. "target_state": "Product",
  100. "addresses": ["address2"]
  101. }]
  102. device_target_instance.Start()
  103. self.assertIsNone(device_target_instance._node_name)
  104. self.assertIsNone(device_target_instance._host)
  105. mock_daemon_stop.assert_called_once()
  106. def testNodeNameDefinedDeviceFoundReturnNoneCheckNameAndHost(
  107. self, mock_daemon_stop):
  108. self.args.node_name = 'device_name'
  109. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  110. mock.patch.object(
  111. FfxTarget, 'get_ssh_address') as mock_get_ssh_address, \
  112. mock.patch.object(
  113. DeviceTarget, '_ConnectToTarget') as mock_connecttotarget, \
  114. mock.patch.object(
  115. DeviceTarget, '_Login') as mock_login:
  116. mock_get_ssh_address.return_value = ('address', 12345)
  117. mock_connecttotarget.return_value = True
  118. self.assertIsNone(device_target_instance.Start())
  119. self.assertEqual(device_target_instance._node_name, 'device_name')
  120. self.assertEqual(device_target_instance._host, 'address')
  121. self.assertEqual(device_target_instance._port, 12345)
  122. mock_login.assert_called_once()
  123. mock_daemon_stop.assert_called_once()
  124. def testNodeNameDefinedDeviceNotFoundRaiseExceptionCouldNotFind(
  125. self, mock_daemon_stop):
  126. self.args.node_name = 'wrong_device_name'
  127. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  128. mock.patch.object(
  129. FfxTarget, 'get_ssh_address') as mock_get_ssh_address, \
  130. self.assertRaisesRegex(Exception, 'Could not find device.'):
  131. mock_get_ssh_address.return_value = None
  132. device_target_instance.Start()
  133. self.assertIsNone(device_target_instance._node_name)
  134. self.assertIsNone(device_target_instance._host)
  135. mock_daemon_stop.assert_called_once()
  136. def testNoDevicesFoundRaiseExceptionCouldNotFind(self, mock_daemon_stop):
  137. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  138. mock.patch.object(FfxRunner, 'list_targets') as mock_list_targets, \
  139. self.assertRaisesRegex(Exception, 'Could not find device.'):
  140. mock_list_targets.return_value = []
  141. device_target_instance.Start()
  142. self.assertIsNone(device_target_instance._node_name)
  143. self.assertIsNone(device_target_instance._host)
  144. mock_daemon_stop.assert_called_once()
  145. def testNoProvisionDeviceIfVersionsMatch(self, mock_daemon_stop):
  146. self.args.os_check = 'update'
  147. self.args.system_image_dir = 'mockdir'
  148. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  149. mock.patch.object(DeviceTarget, '_Discover') as mock_discover, \
  150. mock.patch.object(DeviceTarget, '_ConnectToTarget') as mock_connect, \
  151. mock.patch('device_target.get_sdk_hash') as mock_hash, \
  152. mock.patch.object(
  153. DeviceTarget, '_GetInstalledSdkVersion') as mock_version, \
  154. mock.patch.object(
  155. DeviceTarget, '_ProvisionDevice') as mock_provision, \
  156. mock.patch.object(
  157. DeviceTarget, '_Login') as mock_login:
  158. mock_discover.return_value = True
  159. mock_hash.return_value = '1.0'
  160. mock_version.return_value = '1.0'
  161. device_target_instance.Start()
  162. self.assertEqual(mock_provision.call_count, 0)
  163. mock_login.assert_called_once()
  164. mock_daemon_stop.assert_called_once()
  165. def testRaiseExceptionIfCheckVersionsNoMatch(self, mock_daemon_stop):
  166. self.args.os_check = 'check'
  167. self.args.system_image_dir = 'mockdir'
  168. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  169. mock.patch.object(DeviceTarget, '_Discover') as mock_discover, \
  170. mock.patch.object(DeviceTarget, '_ConnectToTarget') as mock_ready, \
  171. mock.patch('device_target.get_sdk_hash') as mock_hash, \
  172. mock.patch.object(
  173. DeviceTarget, '_GetInstalledSdkVersion') as mock_version, \
  174. mock.patch.object(
  175. DeviceTarget, '_ProvisionDevice') as mock_provision, \
  176. self.assertRaisesRegex(Exception, 'Image and Fuchsia version'):
  177. mock_discover.return_value = True
  178. mock_hash.return_value = '2.0'
  179. mock_version.return_value = '1.0'
  180. device_target_instance.Start()
  181. mock_daemon_stop.assert_called_once()
  182. def testLoginCallsOnlyIfErmineExists(self, mock_daemon_stop):
  183. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  184. mock.patch.object(
  185. ErmineCtl, 'exists',
  186. new_callable=mock.PropertyMock) as mock_exists, \
  187. mock.patch.object(ErmineCtl, 'TakeToShell') as mock_shell:
  188. mock_exists.return_value = True
  189. device_target_instance._Login()
  190. mock_exists.assert_called_once()
  191. mock_shell.assert_called_once()
  192. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  193. mock.patch.object(
  194. ErmineCtl, 'exists',
  195. new_callable=mock.PropertyMock) as mock_exists, \
  196. mock.patch.object(ErmineCtl, 'TakeToShell') as mock_shell:
  197. mock_exists.return_value = False
  198. device_target_instance._Login()
  199. mock_exists.assert_called_once()
  200. self.assertEqual(mock_shell.call_count, 0)
  201. def testProvisionIfOneNonDetectableDevice(self, mock_daemon_stop):
  202. self.args.os_check = 'update'
  203. self.args.node_name = 'mocknode'
  204. self.args.system_image_dir = 'mockdir'
  205. with DeviceTarget.CreateFromArgs(self.args) as device_target_instance, \
  206. mock.patch.object(
  207. FfxTarget, 'get_ssh_address') as mock_get_ssh_address, \
  208. mock.patch.object(DeviceTarget,
  209. '_ProvisionDevice') as mock_provision, \
  210. mock.patch.object(DeviceTarget, '_Login') as mock_bypass:
  211. mock_get_ssh_address.return_value = None
  212. device_target_instance.Start()
  213. self.assertEqual(mock_provision.call_count, 1)
  214. mock_daemon_stop.assert_called_once()
  215. if __name__ == '__main__':
  216. unittest.main()