1 #!/usr/bin/env python3 2 # 3 # Copyright 2018 - The Android Open Source Project 4 # 5 # Licensed under the Apache License, Version 2.0 (the "License"); 6 # you may not use this file except in compliance with the License. 7 # You may obtain a copy of the License at 8 # 9 # http://www.apache.org/licenses/LICENSE-2.0 10 # 11 # Unless required by applicable law or agreed to in writing, software 12 # distributed under the License is distributed on an "AS IS" BASIS, 13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 # See the License for the specific language governing permissions and 15 # limitations under the License. 16 import mock 17 import unittest 18 19 from acts.controllers.sl4a_lib import sl4a_manager 20 from acts.controllers.sl4a_lib import rpc_client 21 22 23 class Sl4aManagerFactoryTest(unittest.TestCase): 24 """Tests the sl4a_manager module-level functions.""" 25 26 def setUp(self): 27 """Clears the Sl4aManager cache.""" 28 sl4a_manager._all_sl4a_managers = {} 29 30 def test_create_manager(self): 31 """Tests sl4a_manager.create_sl4a_manager(). 32 33 Tests that a new Sl4aManager is returned without an error. 34 """ 35 adb = mock.Mock() 36 adb.serial = 'SERIAL' 37 sl4a_man = sl4a_manager.create_sl4a_manager(adb) 38 self.assertEqual(sl4a_man.adb, adb) 39 40 def test_create_sl4a_manager_return_already_created_manager(self): 41 """Tests sl4a_manager.create_sl4a_manager(). 42 43 Tests that a second call to create_sl4a_manager() does not create a 44 new Sl4aManager, and returns the first created Sl4aManager instead. 45 """ 46 adb = mock.Mock() 47 adb.serial = 'SERIAL' 48 first_manager = sl4a_manager.create_sl4a_manager(adb) 49 50 adb_same_serial = mock.Mock() 51 adb_same_serial.serial = 'SERIAL' 52 second_manager = sl4a_manager.create_sl4a_manager(adb) 53 54 self.assertEqual(first_manager, second_manager) 55 56 def test_create_sl4a_manager_multiple_devices_with_one_manager_each(self): 57 """Tests sl4a_manager.create_sl4a_manager(). 58 59 Tests that when create_s4l4a_manager() is called for different devices, 60 each device gets its own Sl4aManager object. 61 """ 62 adb_1 = mock.Mock() 63 adb_1.serial = 'SERIAL' 64 first_manager = sl4a_manager.create_sl4a_manager(adb_1) 65 66 adb_2 = mock.Mock() 67 adb_2.serial = 'DIFFERENT_SERIAL_NUMBER' 68 second_manager = sl4a_manager.create_sl4a_manager(adb_2) 69 70 self.assertNotEqual(first_manager, second_manager) 71 72 73 class Sl4aManagerTest(unittest.TestCase): 74 """Tests the sl4a_manager.Sl4aManager class.""" 75 ATTEMPT_INTERVAL = .25 76 MAX_WAIT_ON_SERVER_SECONDS = 1 77 _SL4A_LAUNCH_SERVER_CMD = '' 78 _SL4A_CLOSE_SERVER_CMD = '' 79 _SL4A_ROOT_FIND_PORT_CMD = '' 80 _SL4A_USER_FIND_PORT_CMD = '' 81 _SL4A_START_SERVICE_CMD = '' 82 83 @classmethod 84 def setUpClass(cls): 85 # Copy all module constants before testing begins. 86 Sl4aManagerTest.ATTEMPT_INTERVAL = \ 87 sl4a_manager.ATTEMPT_INTERVAL 88 Sl4aManagerTest.MAX_WAIT_ON_SERVER_SECONDS = \ 89 sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS 90 Sl4aManagerTest._SL4A_LAUNCH_SERVER_CMD = \ 91 sl4a_manager._SL4A_LAUNCH_SERVER_CMD 92 Sl4aManagerTest._SL4A_CLOSE_SERVER_CMD = \ 93 sl4a_manager._SL4A_CLOSE_SERVER_CMD 94 Sl4aManagerTest._SL4A_ROOT_FIND_PORT_CMD = \ 95 sl4a_manager._SL4A_ROOT_FIND_PORT_CMD 96 Sl4aManagerTest._SL4A_USER_FIND_PORT_CMD = \ 97 sl4a_manager._SL4A_USER_FIND_PORT_CMD 98 Sl4aManagerTest._SL4A_START_SERVICE_CMD = \ 99 sl4a_manager._SL4A_START_SERVICE_CMD 100 101 def setUp(self): 102 # Restore all module constants at the beginning of each test case. 103 sl4a_manager.ATTEMPT_INTERVAL = \ 104 Sl4aManagerTest.ATTEMPT_INTERVAL 105 sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS = \ 106 Sl4aManagerTest.MAX_WAIT_ON_SERVER_SECONDS 107 sl4a_manager._SL4A_LAUNCH_SERVER_CMD = \ 108 Sl4aManagerTest._SL4A_LAUNCH_SERVER_CMD 109 sl4a_manager._SL4A_CLOSE_SERVER_CMD = \ 110 Sl4aManagerTest._SL4A_CLOSE_SERVER_CMD 111 sl4a_manager._SL4A_ROOT_FIND_PORT_CMD = \ 112 Sl4aManagerTest._SL4A_ROOT_FIND_PORT_CMD 113 sl4a_manager._SL4A_USER_FIND_PORT_CMD = \ 114 Sl4aManagerTest._SL4A_USER_FIND_PORT_CMD 115 sl4a_manager._SL4A_START_SERVICE_CMD = \ 116 Sl4aManagerTest._SL4A_START_SERVICE_CMD 117 118 # Reset module data at the beginning of each test. 119 sl4a_manager._all_sl4a_managers = {} 120 121 def test_sl4a_ports_in_use(self): 122 """Tests sl4a_manager.Sl4aManager.sl4a_ports_in_use 123 124 Tests to make sure all server ports are returned with no duplicates. 125 """ 126 adb = mock.Mock() 127 manager = sl4a_manager.Sl4aManager(adb) 128 session_1 = mock.Mock() 129 session_1.server_port = 12345 130 manager.sessions[1] = session_1 131 session_2 = mock.Mock() 132 session_2.server_port = 15973 133 manager.sessions[2] = session_2 134 session_3 = mock.Mock() 135 session_3.server_port = 12345 136 manager.sessions[3] = session_3 137 session_4 = mock.Mock() 138 session_4.server_port = 67890 139 manager.sessions[4] = session_4 140 session_5 = mock.Mock() 141 session_5.server_port = 75638 142 manager.sessions[5] = session_5 143 144 returned_ports = manager.sl4a_ports_in_use 145 146 # No duplicated ports. 147 self.assertEqual(len(returned_ports), len(set(returned_ports))) 148 # One call for each session 149 self.assertSetEqual(set(returned_ports), {12345, 15973, 67890, 75638}) 150 151 @mock.patch('time.sleep', return_value=None) 152 def test_start_sl4a_server_uses_all_retries(self, _): 153 """Tests sl4a_manager.Sl4aManager.start_sl4a_server(). 154 155 Tests to ensure that _start_sl4a_server retries and successfully returns 156 a port. 157 """ 158 adb = mock.Mock() 159 adb.shell = lambda _, **kwargs: '' 160 161 side_effects = [] 162 expected_port = 12345 163 for _ in range(int(sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS / 164 sl4a_manager.ATTEMPT_INTERVAL) - 1): 165 side_effects.append(None) 166 side_effects.append(expected_port) 167 168 manager = sl4a_manager.create_sl4a_manager(adb) 169 manager._get_open_listening_port = mock.Mock(side_effect=side_effects) 170 try: 171 found_port = manager.start_sl4a_server(0) 172 self.assertTrue(found_port) 173 except rpc_client.Sl4aConnectionError: 174 self.fail('start_sl4a_server failed to respect FIND_PORT_RETRIES.') 175 176 @mock.patch('time.sleep', return_value=None) 177 def test_start_sl4a_server_fails_all_retries(self, _): 178 """Tests sl4a_manager.Sl4aManager.start_sl4a_server(). 179 180 Tests to ensure that start_sl4a_server throws an error if all retries 181 fail. 182 """ 183 adb = mock.Mock() 184 adb.shell = lambda _, **kwargs: '' 185 186 side_effects = [] 187 for _ in range(int(sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS / 188 sl4a_manager.ATTEMPT_INTERVAL)): 189 side_effects.append(None) 190 191 manager = sl4a_manager.create_sl4a_manager(adb) 192 manager._get_open_listening_port = mock.Mock(side_effect=side_effects) 193 try: 194 manager.start_sl4a_server(0) 195 self.fail('Sl4aConnectionError was not thrown.') 196 except rpc_client.Sl4aConnectionError: 197 pass 198 199 def test_get_all_ports_command_uses_root_cmd(self): 200 """Tests sl4a_manager.Sl4aManager._get_all_ports_command(). 201 202 Tests that _get_all_ports_command calls the root command when root is 203 available. 204 """ 205 adb = mock.Mock() 206 adb.is_root = lambda: True 207 command = 'ngo45hke3b4vie3mv5ni93,vfu3j' 208 sl4a_manager._SL4A_ROOT_FIND_PORT_CMD = command 209 210 manager = sl4a_manager.create_sl4a_manager(adb) 211 self.assertEqual(manager._get_all_ports_command(), command) 212 213 def test_get_all_ports_command_escalates_to_root(self): 214 """Tests sl4a_manager.Sl4aManager._call_get_ports_command(). 215 216 Tests that _call_get_ports_command calls the root command when adb is 217 user but can escalate to root. 218 """ 219 adb = mock.Mock() 220 adb.is_root = lambda: False 221 adb.ensure_root = lambda: True 222 command = 'ngo45hke3b4vie3mv5ni93,vfu3j' 223 sl4a_manager._SL4A_ROOT_FIND_PORT_CMD = command 224 225 manager = sl4a_manager.create_sl4a_manager(adb) 226 self.assertEqual(manager._get_all_ports_command(), command) 227 228 def test_get_all_ports_command_uses_user_cmd(self): 229 """Tests sl4a_manager.Sl4aManager._call_get_ports_command(). 230 231 Tests that _call_get_ports_command calls the user command when root is 232 unavailable. 233 """ 234 adb = mock.Mock() 235 adb.is_root = lambda: False 236 adb.ensure_root = lambda: False 237 command = 'ngo45hke3b4vie3mv5ni93,vfu3j' 238 sl4a_manager._SL4A_USER_FIND_PORT_CMD = command 239 240 manager = sl4a_manager.create_sl4a_manager(adb) 241 self.assertEqual(manager._get_all_ports_command(), command) 242 243 def test_get_open_listening_port_no_port_found(self): 244 """Tests sl4a_manager.Sl4aManager._get_open_listening_port(). 245 246 Tests to ensure None is returned if no open port is found. 247 """ 248 adb = mock.Mock() 249 adb.shell = lambda _: '' 250 251 manager = sl4a_manager.create_sl4a_manager(adb) 252 self.assertIsNone(manager._get_open_listening_port()) 253 254 def test_get_open_listening_port_no_new_port_found(self): 255 """Tests sl4a_manager.Sl4aManager._get_open_listening_port(). 256 257 Tests to ensure None is returned if the ports returned have all been 258 marked as in used. 259 """ 260 adb = mock.Mock() 261 adb.shell = lambda _: '12345 67890' 262 263 manager = sl4a_manager.create_sl4a_manager(adb) 264 manager._sl4a_ports = {'12345', '67890'} 265 self.assertIsNone(manager._get_open_listening_port()) 266 267 def test_get_open_listening_port_port_is_avaiable(self): 268 """Tests sl4a_manager.Sl4aManager._get_open_listening_port(). 269 270 Tests to ensure a port is returned if a port is found and has not been 271 marked as used. 272 """ 273 adb = mock.Mock() 274 adb.shell = lambda _: '12345 67890' 275 276 manager = sl4a_manager.create_sl4a_manager(adb) 277 manager._sl4a_ports = {'12345'} 278 self.assertEqual(manager._get_open_listening_port(), 67890) 279 280 def test_is_sl4a_installed_is_true(self): 281 """Tests sl4a_manager.Sl4aManager.is_sl4a_installed(). 282 283 Tests is_sl4a_installed() returns true when pm returns data 284 """ 285 adb = mock.Mock() 286 adb.shell = lambda _, **kwargs: 'asdf' 287 manager = sl4a_manager.create_sl4a_manager(adb) 288 self.assertTrue(manager.is_sl4a_installed()) 289 290 def test_is_sl4a_installed_is_false(self): 291 """Tests sl4a_manager.Sl4aManager.is_sl4a_installed(). 292 293 Tests is_sl4a_installed() returns true when pm returns data 294 """ 295 adb = mock.Mock() 296 adb.shell = lambda _, **kwargs: '' 297 manager = sl4a_manager.create_sl4a_manager(adb) 298 self.assertFalse(manager.is_sl4a_installed()) 299 300 def test_start_sl4a_throws_error_on_sl4a_not_installed(self): 301 """Tests sl4a_manager.Sl4aManager.start_sl4a_service(). 302 303 Tests that a MissingSl4aError is thrown when SL4A is not installed. 304 """ 305 adb = mock.Mock() 306 307 manager = sl4a_manager.create_sl4a_manager(adb) 308 manager.is_sl4a_installed = lambda: False 309 try: 310 manager.start_sl4a_service() 311 self.fail('An error should have been thrown.') 312 except rpc_client.MissingSl4AError: 313 pass 314 315 def test_start_sl4a_starts_sl4a_if_not_running(self): 316 """Tests sl4a_manager.Sl4aManager.start_sl4a_service(). 317 318 Tests that SL4A is started if it was not already running. 319 """ 320 adb = mock.Mock() 321 adb.shell = mock.Mock(side_effect=['', '', '']) 322 323 manager = sl4a_manager.create_sl4a_manager(adb) 324 manager.is_sl4a_installed = lambda: True 325 try: 326 manager.start_sl4a_service() 327 except rpc_client.MissingSl4AError: 328 self.fail('An error should not have been thrown.') 329 adb.shell.assert_called_with(sl4a_manager._SL4A_START_SERVICE_CMD) 330 331 def test_create_session_uses_oldest_server_port(self): 332 """Tests sl4a_manager.Sl4aManager.create_session(). 333 334 Tests that when no port is given, the oldest server port opened is used 335 as the server port for a new session. The oldest server port can be 336 found by getting the oldest session's server port. 337 """ 338 adb = mock.Mock() 339 340 manager = sl4a_manager.create_sl4a_manager(adb) 341 # Ignore starting SL4A. 342 manager.start_sl4a_service = lambda: None 343 344 session_1 = mock.Mock() 345 session_1.server_port = 12345 346 session_2 = mock.Mock() 347 session_2.server_port = 67890 348 session_3 = mock.Mock() 349 session_3.server_port = 67890 350 351 manager.sessions[3] = session_3 352 manager.sessions[1] = session_1 353 manager.sessions[2] = session_2 354 355 with mock.patch.object( 356 rpc_client.RpcClient, '__init__', return_value=None): 357 created_session = manager.create_session() 358 359 self.assertEqual(created_session.server_port, session_1.server_port) 360 361 def test_create_session_uses_random_port_when_no_session_exists(self): 362 """Tests sl4a_manager.Sl4aManager.create_session(). 363 364 Tests that when no port is given, and no SL4A server exists, the server 365 port for the session is set to 0. 366 """ 367 adb = mock.Mock() 368 369 manager = sl4a_manager.create_sl4a_manager(adb) 370 # Ignore starting SL4A. 371 manager.start_sl4a_service = lambda: None 372 373 with mock.patch.object( 374 rpc_client.RpcClient, '__init__', return_value=None): 375 created_session = manager.create_session() 376 377 self.assertEqual(created_session.server_port, 0) 378 379 def test_terminate_all_session_call_terminate_on_all_sessions(self): 380 """Tests sl4a_manager.Sl4aManager.terminate_all_sessions(). 381 382 Tests to see that the manager has called terminate on all sessions. 383 """ 384 called_terminate_on = list() 385 386 def called_on(session): 387 called_terminate_on.append(session) 388 389 adb = mock.Mock() 390 manager = sl4a_manager.Sl4aManager(adb) 391 392 session_1 = mock.Mock() 393 session_1.terminate = lambda *args, **kwargs: called_on(session_1) 394 manager.sessions[1] = session_1 395 session_4 = mock.Mock() 396 session_4.terminate = lambda *args, **kwargs: called_on(session_4) 397 manager.sessions[4] = session_4 398 session_5 = mock.Mock() 399 session_5.terminate = lambda *args, **kwargs: called_on(session_5) 400 manager.sessions[5] = session_5 401 402 manager._get_all_ports = lambda: [] 403 manager.terminate_all_sessions() 404 # No duplicates calls to terminate. 405 self.assertEqual( 406 len(called_terminate_on), len(set(called_terminate_on))) 407 # One call for each session 408 self.assertSetEqual( 409 set(called_terminate_on), {session_1, session_4, session_5}) 410 411 def test_terminate_all_session_close_each_server(self): 412 """Tests sl4a_manager.Sl4aManager.terminate_all_sessions(). 413 414 Tests to see that the manager has called terminate on all sessions. 415 """ 416 closed_ports = list() 417 418 def close(command): 419 if str.isdigit(command): 420 closed_ports.append(command) 421 return '' 422 423 adb = mock.Mock() 424 adb.shell = close 425 sl4a_manager._SL4A_CLOSE_SERVER_CMD = '%s' 426 ports_to_close = {'12345', '67890', '24680', '13579'} 427 428 manager = sl4a_manager.Sl4aManager(adb) 429 manager._sl4a_ports = set(ports_to_close) 430 manager._get_all_ports = lambda: [] 431 manager.terminate_all_sessions() 432 433 # No duplicate calls to close port 434 self.assertEqual(len(closed_ports), len(set(closed_ports))) 435 # One call for each port 436 self.assertSetEqual(ports_to_close, set(closed_ports)) 437 438 def test_obtain_sl4a_server_starts_new_server(self): 439 """Tests sl4a_manager.Sl4aManager.obtain_sl4a_server(). 440 441 Tests that a new server can be returned if the server does not exist. 442 """ 443 adb = mock.Mock() 444 manager = sl4a_manager.Sl4aManager(adb) 445 manager.start_sl4a_server = mock.Mock() 446 447 manager.obtain_sl4a_server(0) 448 449 self.assertTrue(manager.start_sl4a_server.called) 450 451 @mock.patch( 452 'acts.controllers.sl4a_lib.sl4a_manager.Sl4aManager.sl4a_ports_in_use', 453 new_callable=mock.PropertyMock) 454 def test_obtain_sl4a_server_returns_existing_server( 455 self, sl4a_ports_in_use): 456 """Tests sl4a_manager.Sl4aManager.obtain_sl4a_server(). 457 458 Tests that an existing server is returned if it is already opened. 459 """ 460 adb = mock.Mock() 461 manager = sl4a_manager.Sl4aManager(adb) 462 manager.start_sl4a_server = mock.Mock() 463 sl4a_ports_in_use.return_value = [12345] 464 465 ret = manager.obtain_sl4a_server(12345) 466 467 self.assertFalse(manager.start_sl4a_server.called) 468 self.assertEqual(12345, ret) 469 470 471 if __name__ == '__main__': 472 unittest.main() 473
