1 #!/usr/bin/env python3 2 # 3 # Copyright 2016 - 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 rpc_client, rpc_connection 20 21 MOCK_RESP = b'{"id": 0, "result": 123, "error": null, "status": 1, "uid": 1}' 22 MOCK_RESP_UNKNOWN_UID = b'{"id": 0, "result": 123, "error": null, "status": 0}' 23 MOCK_RESP_WITH_ERROR = b'{"id": 0, "error": 1, "status": 1, "uid": 1}' 24 25 26 class MockSocketFile(object): 27 def __init__(self, resp): 28 self.resp = resp 29 self.last_write = None 30 31 def write(self, msg): 32 self.last_write = msg 33 34 def readline(self): 35 return self.resp 36 37 def flush(self): 38 pass 39 40 41 class RpcConnectionTest(unittest.TestCase): 42 """This test class has unit tests for the implementation of everything 43 under acts.controllers.android, which is the RPC client module for sl4a. 44 """ 45 46 @staticmethod 47 def mock_rpc_connection(response=MOCK_RESP, 48 uid=rpc_connection.UNKNOWN_UID): 49 """Sets up a faked socket file from the mock connection.""" 50 fake_file = MockSocketFile(response) 51 fake_conn = mock.MagicMock() 52 fake_conn.makefile.return_value = fake_file 53 adb = mock.Mock() 54 ports = mock.Mock() 55 56 return rpc_connection.RpcConnection( 57 adb, ports, fake_conn, fake_file, uid=uid) 58 59 def test_open_chooses_init_on_unknown_uid(self): 60 """Tests rpc_connection.RpcConnection.open(). 61 62 Tests that open uses the init start command when the uid is unknown. 63 """ 64 65 def pass_on_init(start_command): 66 if not start_command == rpc_connection.Sl4aConnectionCommand.INIT: 67 self.fail( 68 'Must call "init". Called "%s" instead.' % start_command) 69 70 connection = self.mock_rpc_connection() 71 connection._initiate_handshake = pass_on_init 72 connection.open() 73 74 def test_open_chooses_continue_on_known_uid(self): 75 """Tests rpc_connection.RpcConnection.open(). 76 77 Tests that open uses the continue start command when the uid is known. 78 """ 79 80 def pass_on_continue(start_command): 81 if start_command != rpc_connection.Sl4aConnectionCommand.CONTINUE: 82 self.fail('Must call "continue". Called "%s" instead.' % 83 start_command) 84 85 connection = self.mock_rpc_connection(uid=1) 86 connection._initiate_handshake = pass_on_continue 87 connection.open() 88 89 def test_initiate_handshake_returns_uid(self): 90 """Tests rpc_connection.RpcConnection._initiate_handshake(). 91 92 Test that at the end of a handshake with no errors the client object 93 has the correct parameters. 94 """ 95 connection = self.mock_rpc_connection() 96 connection._initiate_handshake( 97 rpc_connection.Sl4aConnectionCommand.INIT) 98 99 self.assertEqual(connection.uid, 1) 100 101 def test_initiate_handshake_returns_unknown_status(self): 102 """Tests rpc_connection.RpcConnection._initiate_handshake(). 103 104 Test that when the handshake is given an unknown uid then the client 105 will not be given a uid. 106 """ 107 connection = self.mock_rpc_connection(MOCK_RESP_UNKNOWN_UID) 108 connection._initiate_handshake( 109 rpc_connection.Sl4aConnectionCommand.INIT) 110 111 self.assertEqual(connection.uid, rpc_client.UNKNOWN_UID) 112 113 def test_initiate_handshake_no_response(self): 114 """Tests rpc_connection.RpcConnection._initiate_handshake(). 115 116 Test that if a handshake receives no response then it will give a 117 protocol error. 118 """ 119 connection = self.mock_rpc_connection(b'') 120 121 with self.assertRaises( 122 rpc_client.Sl4aProtocolError, 123 msg=rpc_client.Sl4aProtocolError.NO_RESPONSE_FROM_HANDSHAKE): 124 connection._initiate_handshake( 125 rpc_connection.Sl4aConnectionCommand.INIT) 126 127 def test_cmd_properly_formatted(self): 128 """Tests rpc_connection.RpcConnection._cmd(). 129 130 Tests that the command sent is properly formatted. 131 """ 132 connection = self.mock_rpc_connection(MOCK_RESP) 133 connection._cmd('test') 134 self.assertIn( 135 connection._socket_file.last_write, 136 [b'{"cmd": "test", "uid": -1}\n', b'{"uid": -1, "cmd": "test"}\n']) 137 138 def test_get_new_ticket(self): 139 """Tests rpc_connection.RpcConnection.get_new_ticket(). 140 141 Tests that a new number is always given for get_new_ticket(). 142 """ 143 connection = self.mock_rpc_connection(MOCK_RESP) 144 self.assertEqual(connection.get_new_ticket() + 1, 145 connection.get_new_ticket()) 146 147 148 if __name__ == "__main__": 149 unittest.main() 150
