| #!/usr/bin/env python3 |
| # |
| # Copyright 2022 The Fuchsia Authors |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| import unittest |
| |
| import mock |
| |
| from antlion.controllers.sl4a_lib import rpc_client, sl4a_manager |
| |
| |
| class Sl4aManagerFactoryTest(unittest.TestCase): |
| """Tests the sl4a_manager module-level functions.""" |
| |
| def setUp(self): |
| """Clears the Sl4aManager cache.""" |
| sl4a_manager._all_sl4a_managers = {} |
| |
| def test_create_manager(self): |
| """Tests sl4a_manager.create_sl4a_manager(). |
| |
| Tests that a new Sl4aManager is returned without an error. |
| """ |
| adb = mock.Mock() |
| adb.serial = "SERIAL" |
| sl4a_man = sl4a_manager.create_sl4a_manager(adb) |
| self.assertEqual(sl4a_man.adb, adb) |
| |
| def test_create_sl4a_manager_return_already_created_manager(self): |
| """Tests sl4a_manager.create_sl4a_manager(). |
| |
| Tests that a second call to create_sl4a_manager() does not create a |
| new Sl4aManager, and returns the first created Sl4aManager instead. |
| """ |
| adb = mock.Mock() |
| adb.serial = "SERIAL" |
| first_manager = sl4a_manager.create_sl4a_manager(adb) |
| |
| adb_same_serial = mock.Mock() |
| adb_same_serial.serial = "SERIAL" |
| second_manager = sl4a_manager.create_sl4a_manager(adb) |
| |
| self.assertEqual(first_manager, second_manager) |
| |
| def test_create_sl4a_manager_multiple_devices_with_one_manager_each(self): |
| """Tests sl4a_manager.create_sl4a_manager(). |
| |
| Tests that when create_s4l4a_manager() is called for different devices, |
| each device gets its own Sl4aManager object. |
| """ |
| adb_1 = mock.Mock() |
| adb_1.serial = "SERIAL" |
| first_manager = sl4a_manager.create_sl4a_manager(adb_1) |
| |
| adb_2 = mock.Mock() |
| adb_2.serial = "DIFFERENT_SERIAL_NUMBER" |
| second_manager = sl4a_manager.create_sl4a_manager(adb_2) |
| |
| self.assertNotEqual(first_manager, second_manager) |
| |
| |
| class Sl4aManagerTest(unittest.TestCase): |
| """Tests the sl4a_manager.Sl4aManager class.""" |
| |
| ATTEMPT_INTERVAL = 0.25 |
| MAX_WAIT_ON_SERVER_SECONDS = 1 |
| _SL4A_LAUNCH_SERVER_CMD = "" |
| _SL4A_CLOSE_SERVER_CMD = "" |
| _SL4A_ROOT_FIND_PORT_CMD = "" |
| _SL4A_USER_FIND_PORT_CMD = "" |
| _SL4A_START_SERVICE_CMD = "" |
| |
| @classmethod |
| def setUpClass(cls): |
| # Copy all module constants before testing begins. |
| Sl4aManagerTest.ATTEMPT_INTERVAL = sl4a_manager.ATTEMPT_INTERVAL |
| Sl4aManagerTest.MAX_WAIT_ON_SERVER_SECONDS = ( |
| sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS |
| ) |
| Sl4aManagerTest._SL4A_LAUNCH_SERVER_CMD = sl4a_manager._SL4A_LAUNCH_SERVER_CMD |
| Sl4aManagerTest._SL4A_CLOSE_SERVER_CMD = sl4a_manager._SL4A_CLOSE_SERVER_CMD |
| Sl4aManagerTest._SL4A_ROOT_FIND_PORT_CMD = sl4a_manager._SL4A_ROOT_FIND_PORT_CMD |
| Sl4aManagerTest._SL4A_USER_FIND_PORT_CMD = sl4a_manager._SL4A_USER_FIND_PORT_CMD |
| Sl4aManagerTest._SL4A_START_SERVICE_CMD = sl4a_manager._SL4A_START_SERVICE_CMD |
| |
| def setUp(self): |
| # Restore all module constants at the beginning of each test case. |
| sl4a_manager.ATTEMPT_INTERVAL = Sl4aManagerTest.ATTEMPT_INTERVAL |
| sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS = ( |
| Sl4aManagerTest.MAX_WAIT_ON_SERVER_SECONDS |
| ) |
| sl4a_manager._SL4A_LAUNCH_SERVER_CMD = Sl4aManagerTest._SL4A_LAUNCH_SERVER_CMD |
| sl4a_manager._SL4A_CLOSE_SERVER_CMD = Sl4aManagerTest._SL4A_CLOSE_SERVER_CMD |
| sl4a_manager._SL4A_ROOT_FIND_PORT_CMD = Sl4aManagerTest._SL4A_ROOT_FIND_PORT_CMD |
| sl4a_manager._SL4A_USER_FIND_PORT_CMD = Sl4aManagerTest._SL4A_USER_FIND_PORT_CMD |
| sl4a_manager._SL4A_START_SERVICE_CMD = Sl4aManagerTest._SL4A_START_SERVICE_CMD |
| |
| # Reset module data at the beginning of each test. |
| sl4a_manager._all_sl4a_managers = {} |
| |
| def test_sl4a_ports_in_use(self): |
| """Tests sl4a_manager.Sl4aManager.sl4a_ports_in_use |
| |
| Tests to make sure all server ports are returned with no duplicates. |
| """ |
| adb = mock.Mock() |
| manager = sl4a_manager.Sl4aManager(adb) |
| session_1 = mock.Mock() |
| session_1.server_port = 12345 |
| manager.sessions[1] = session_1 |
| session_2 = mock.Mock() |
| session_2.server_port = 15973 |
| manager.sessions[2] = session_2 |
| session_3 = mock.Mock() |
| session_3.server_port = 12345 |
| manager.sessions[3] = session_3 |
| session_4 = mock.Mock() |
| session_4.server_port = 67890 |
| manager.sessions[4] = session_4 |
| session_5 = mock.Mock() |
| session_5.server_port = 75638 |
| manager.sessions[5] = session_5 |
| |
| returned_ports = manager.sl4a_ports_in_use |
| |
| # No duplicated ports. |
| self.assertEqual(len(returned_ports), len(set(returned_ports))) |
| # One call for each session |
| self.assertSetEqual(set(returned_ports), {12345, 15973, 67890, 75638}) |
| |
| @mock.patch("time.sleep", return_value=None) |
| def test_start_sl4a_server_uses_all_retries(self, _): |
| """Tests sl4a_manager.Sl4aManager.start_sl4a_server(). |
| |
| Tests to ensure that _start_sl4a_server retries and successfully returns |
| a port. |
| """ |
| adb = mock.Mock() |
| adb.shell = lambda _, **kwargs: "" |
| |
| side_effects = [] |
| expected_port = 12345 |
| for _ in range( |
| int(sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS / sl4a_manager.ATTEMPT_INTERVAL) |
| - 1 |
| ): |
| side_effects.append(None) |
| side_effects.append(expected_port) |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| manager._get_open_listening_port = mock.Mock(side_effect=side_effects) |
| try: |
| found_port = manager.start_sl4a_server(0) |
| self.assertTrue(found_port) |
| except rpc_client.Sl4aConnectionError: |
| self.fail("start_sl4a_server failed to respect FIND_PORT_RETRIES.") |
| |
| @mock.patch("time.sleep", return_value=None) |
| def test_start_sl4a_server_fails_all_retries(self, _): |
| """Tests sl4a_manager.Sl4aManager.start_sl4a_server(). |
| |
| Tests to ensure that start_sl4a_server throws an error if all retries |
| fail. |
| """ |
| adb = mock.Mock() |
| adb.shell = lambda _, **kwargs: "" |
| |
| side_effects = [] |
| for _ in range( |
| int(sl4a_manager.MAX_WAIT_ON_SERVER_SECONDS / sl4a_manager.ATTEMPT_INTERVAL) |
| ): |
| side_effects.append(None) |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| manager._get_open_listening_port = mock.Mock(side_effect=side_effects) |
| try: |
| manager.start_sl4a_server(0) |
| self.fail("Sl4aConnectionError was not thrown.") |
| except rpc_client.Sl4aConnectionError: |
| pass |
| |
| def test_get_all_ports_command_uses_root_cmd(self): |
| """Tests sl4a_manager.Sl4aManager._get_all_ports_command(). |
| |
| Tests that _get_all_ports_command calls the root command when root is |
| available. |
| """ |
| adb = mock.Mock() |
| adb.is_root = lambda: True |
| command = "ngo45hke3b4vie3mv5ni93,vfu3j" |
| sl4a_manager._SL4A_ROOT_FIND_PORT_CMD = command |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| self.assertEqual(manager._get_all_ports_command(), command) |
| |
| def test_get_all_ports_command_escalates_to_root(self): |
| """Tests sl4a_manager.Sl4aManager._call_get_ports_command(). |
| |
| Tests that _call_get_ports_command calls the root command when adb is |
| user but can escalate to root. |
| """ |
| adb = mock.Mock() |
| adb.is_root = lambda: False |
| adb.ensure_root = lambda: True |
| command = "ngo45hke3b4vie3mv5ni93,vfu3j" |
| sl4a_manager._SL4A_ROOT_FIND_PORT_CMD = command |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| self.assertEqual(manager._get_all_ports_command(), command) |
| |
| def test_get_all_ports_command_uses_user_cmd(self): |
| """Tests sl4a_manager.Sl4aManager._call_get_ports_command(). |
| |
| Tests that _call_get_ports_command calls the user command when root is |
| unavailable. |
| """ |
| adb = mock.Mock() |
| adb.is_root = lambda: False |
| adb.ensure_root = lambda: False |
| command = "ngo45hke3b4vie3mv5ni93,vfu3j" |
| sl4a_manager._SL4A_USER_FIND_PORT_CMD = command |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| self.assertEqual(manager._get_all_ports_command(), command) |
| |
| def test_get_open_listening_port_no_port_found(self): |
| """Tests sl4a_manager.Sl4aManager._get_open_listening_port(). |
| |
| Tests to ensure None is returned if no open port is found. |
| """ |
| adb = mock.Mock() |
| adb.shell = lambda _: "" |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| self.assertIsNone(manager._get_open_listening_port()) |
| |
| def test_get_open_listening_port_no_new_port_found(self): |
| """Tests sl4a_manager.Sl4aManager._get_open_listening_port(). |
| |
| Tests to ensure None is returned if the ports returned have all been |
| marked as in used. |
| """ |
| adb = mock.Mock() |
| adb.shell = lambda _: "12345 67890" |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| manager._sl4a_ports = {"12345", "67890"} |
| self.assertIsNone(manager._get_open_listening_port()) |
| |
| def test_get_open_listening_port_port_is_avaiable(self): |
| """Tests sl4a_manager.Sl4aManager._get_open_listening_port(). |
| |
| Tests to ensure a port is returned if a port is found and has not been |
| marked as used. |
| """ |
| adb = mock.Mock() |
| adb.shell = lambda _: "12345 67890" |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| manager._sl4a_ports = {"12345"} |
| self.assertEqual(manager._get_open_listening_port(), 67890) |
| |
| def test_is_sl4a_installed_is_true(self): |
| """Tests sl4a_manager.Sl4aManager.is_sl4a_installed(). |
| |
| Tests is_sl4a_installed() returns true when pm returns data |
| """ |
| adb = mock.Mock() |
| adb.shell = lambda _, **kwargs: "asdf" |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| self.assertTrue(manager.is_sl4a_installed()) |
| |
| def test_is_sl4a_installed_is_false(self): |
| """Tests sl4a_manager.Sl4aManager.is_sl4a_installed(). |
| |
| Tests is_sl4a_installed() returns true when pm returns data |
| """ |
| adb = mock.Mock() |
| adb.shell = lambda _, **kwargs: "" |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| self.assertFalse(manager.is_sl4a_installed()) |
| |
| def test_start_sl4a_throws_error_on_sl4a_not_installed(self): |
| """Tests sl4a_manager.Sl4aManager.start_sl4a_service(). |
| |
| Tests that a MissingSl4aError is thrown when SL4A is not installed. |
| """ |
| adb = mock.Mock() |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| manager.is_sl4a_installed = lambda: False |
| try: |
| manager.start_sl4a_service() |
| self.fail("An error should have been thrown.") |
| except rpc_client.Sl4aNotInstalledError: |
| pass |
| |
| def test_start_sl4a_starts_sl4a_if_not_running(self): |
| """Tests sl4a_manager.Sl4aManager.start_sl4a_service(). |
| |
| Tests that SL4A is started if it was not already running. |
| """ |
| adb = mock.Mock() |
| adb.shell = mock.Mock(side_effect=["", "", ""]) |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| manager.is_sl4a_installed = lambda: True |
| try: |
| manager.start_sl4a_service() |
| except rpc_client.Sl4aNotInstalledError: |
| self.fail("An error should not have been thrown.") |
| adb.shell.assert_called_with(sl4a_manager._SL4A_START_SERVICE_CMD) |
| |
| def test_create_session_uses_oldest_server_port(self): |
| """Tests sl4a_manager.Sl4aManager.create_session(). |
| |
| Tests that when no port is given, the oldest server port opened is used |
| as the server port for a new session. The oldest server port can be |
| found by getting the oldest session's server port. |
| """ |
| adb = mock.Mock() |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| # Ignore starting SL4A. |
| manager.start_sl4a_service = lambda: None |
| |
| session_1 = mock.Mock() |
| session_1.server_port = 12345 |
| session_2 = mock.Mock() |
| session_2.server_port = 67890 |
| session_3 = mock.Mock() |
| session_3.server_port = 67890 |
| |
| manager.sessions[3] = session_3 |
| manager.sessions[1] = session_1 |
| manager.sessions[2] = session_2 |
| |
| with mock.patch.object(rpc_client.RpcClient, "__init__", return_value=None): |
| created_session = manager.create_session() |
| |
| self.assertEqual(created_session.server_port, session_1.server_port) |
| |
| def test_create_session_uses_random_port_when_no_session_exists(self): |
| """Tests sl4a_manager.Sl4aManager.create_session(). |
| |
| Tests that when no port is given, and no SL4A server exists, the server |
| port for the session is set to 0. |
| """ |
| adb = mock.Mock() |
| |
| manager = sl4a_manager.create_sl4a_manager(adb) |
| # Ignore starting SL4A. |
| manager.start_sl4a_service = lambda: None |
| |
| with mock.patch.object(rpc_client.RpcClient, "__init__", return_value=None): |
| created_session = manager.create_session() |
| |
| self.assertEqual(created_session.server_port, 0) |
| |
| def test_terminate_all_session_call_terminate_on_all_sessions(self): |
| """Tests sl4a_manager.Sl4aManager.terminate_all_sessions(). |
| |
| Tests to see that the manager has called terminate on all sessions. |
| """ |
| called_terminate_on = list() |
| |
| def called_on(session): |
| called_terminate_on.append(session) |
| |
| adb = mock.Mock() |
| manager = sl4a_manager.Sl4aManager(adb) |
| |
| session_1 = mock.Mock() |
| session_1.terminate = lambda *args, **kwargs: called_on(session_1) |
| manager.sessions[1] = session_1 |
| session_4 = mock.Mock() |
| session_4.terminate = lambda *args, **kwargs: called_on(session_4) |
| manager.sessions[4] = session_4 |
| session_5 = mock.Mock() |
| session_5.terminate = lambda *args, **kwargs: called_on(session_5) |
| manager.sessions[5] = session_5 |
| |
| manager._get_all_ports = lambda: [] |
| manager.terminate_all_sessions() |
| # No duplicates calls to terminate. |
| self.assertEqual(len(called_terminate_on), len(set(called_terminate_on))) |
| # One call for each session |
| self.assertSetEqual(set(called_terminate_on), {session_1, session_4, session_5}) |
| |
| def test_terminate_all_session_close_each_server(self): |
| """Tests sl4a_manager.Sl4aManager.terminate_all_sessions(). |
| |
| Tests to see that the manager has called terminate on all sessions. |
| """ |
| closed_ports = list() |
| |
| def close(command): |
| if str.isdigit(command): |
| closed_ports.append(command) |
| return "" |
| |
| adb = mock.Mock() |
| adb.shell = close |
| sl4a_manager._SL4A_CLOSE_SERVER_CMD = "%s" |
| ports_to_close = {"12345", "67890", "24680", "13579"} |
| |
| manager = sl4a_manager.Sl4aManager(adb) |
| manager._sl4a_ports = set(ports_to_close) |
| manager._get_all_ports = lambda: [] |
| manager.terminate_all_sessions() |
| |
| # No duplicate calls to close port |
| self.assertEqual(len(closed_ports), len(set(closed_ports))) |
| # One call for each port |
| self.assertSetEqual(ports_to_close, set(closed_ports)) |
| |
| def test_obtain_sl4a_server_starts_new_server(self): |
| """Tests sl4a_manager.Sl4aManager.obtain_sl4a_server(). |
| |
| Tests that a new server can be returned if the server does not exist. |
| """ |
| adb = mock.Mock() |
| manager = sl4a_manager.Sl4aManager(adb) |
| manager.start_sl4a_server = mock.Mock() |
| |
| manager.obtain_sl4a_server(0) |
| |
| self.assertTrue(manager.start_sl4a_server.called) |
| |
| @mock.patch( |
| "antlion.controllers.sl4a_lib.sl4a_manager.Sl4aManager.sl4a_ports_in_use", |
| new_callable=mock.PropertyMock, |
| ) |
| def test_obtain_sl4a_server_returns_existing_server(self, sl4a_ports_in_use): |
| """Tests sl4a_manager.Sl4aManager.obtain_sl4a_server(). |
| |
| Tests that an existing server is returned if it is already opened. |
| """ |
| adb = mock.Mock() |
| manager = sl4a_manager.Sl4aManager(adb) |
| manager.start_sl4a_server = mock.Mock() |
| sl4a_ports_in_use.return_value = [12345] |
| |
| ret = manager.obtain_sl4a_server(12345) |
| |
| self.assertFalse(manager.start_sl4a_server.called) |
| self.assertEqual(12345, ret) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
