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 errno 17 import mock 18 from socket import timeout 19 from socket import error as socket_error 20 import unittest 21 from mock import patch 22 23 from acts.controllers.sl4a_lib import sl4a_ports 24 from acts.controllers.sl4a_lib import sl4a_session 25 from acts.controllers.sl4a_lib import rpc_client 26 27 28 class Sl4aSessionTest(unittest.TestCase): 29 """Tests the sl4a_session.Sl4aSession class.""" 30 31 def test_is_alive_true_on_not_terminated(self): 32 """Tests sl4a_session.Sl4aSession.is_alive. 33 34 Tests that the session is_alive when it has not been terminated. 35 """ 36 session = mock.Mock() 37 session._terminated = False 38 session.is_alive = sl4a_session.Sl4aSession.is_alive 39 self.assertNotEqual(session._terminated, session.is_alive) 40 41 def test_is_alive_false_on_terminated(self): 42 """Tests sl4a_session.Sl4aSession.is_alive. 43 44 Tests that the session is_alive when it has not been terminated. 45 """ 46 session = mock.Mock() 47 session._terminated = True 48 session.is_alive = sl4a_session.Sl4aSession.is_alive 49 self.assertNotEqual(session._terminated, session.is_alive) 50 51 @patch('acts.controllers.sl4a_lib.event_dispatcher.EventDispatcher') 52 def test_get_event_dispatcher_create_on_none(self, _): 53 """Tests sl4a_session.Sl4aSession.get_event_dispatcher. 54 55 Tests that a new event_dispatcher is created if one does not exist. 56 """ 57 session = mock.Mock() 58 session._event_dispatcher = None 59 ed = sl4a_session.Sl4aSession.get_event_dispatcher(session) 60 self.assertTrue(session._event_dispatcher is not None) 61 self.assertEqual(session._event_dispatcher, ed) 62 63 def test_get_event_dispatcher_returns_existing_event_dispatcher(self): 64 """Tests sl4a_session.Sl4aSession.get_event_dispatcher. 65 66 Tests that the existing event_dispatcher is returned. 67 """ 68 session = mock.Mock() 69 session._event_dispatcher = 'Something that is not None' 70 ed = sl4a_session.Sl4aSession.get_event_dispatcher(session) 71 self.assertEqual(session._event_dispatcher, ed) 72 73 def test_create_client_side_connection_hint_already_in_use(self): 74 """Tests sl4a_session.Sl4aSession._create_client_side_connection(). 75 76 Tests that if the hinted port is already in use, the function will 77 call itself with a hinted port of 0 (random). 78 """ 79 session = mock.Mock() 80 session._create_client_side_connection = mock.Mock() 81 with mock.patch('socket.socket') as socket: 82 # Throw an error when trying to bind to the hinted port. 83 error = OSError() 84 error.errno = errno.EADDRINUSE 85 socket_instance = mock.Mock() 86 socket_instance.bind = mock.Mock() 87 socket_instance.bind.side_effect = error 88 socket.return_value = socket_instance 89 90 sl4a_session.Sl4aSession._create_client_side_connection( 91 session, sl4a_ports.Sl4aPorts(1, 2, 3)) 92 93 fn = session._create_client_side_connection 94 self.assertEqual(fn.call_count, 1) 95 # Asserts that the 1st argument (Sl4aPorts) sent to the function 96 # has a client port of 0. 97 self.assertEqual(fn.call_args_list[0][0][0].client_port, 0) 98 99 def test_create_client_side_connection_catches_timeout(self): 100 """Tests sl4a_session.Sl4aSession._create_client_side_connection(). 101 102 Tests that the function will raise an Sl4aConnectionError upon timeout. 103 """ 104 session = mock.Mock() 105 session._create_client_side_connection = mock.Mock() 106 error = timeout() 107 with mock.patch('socket.socket') as socket: 108 # Throw an error when trying to bind to the hinted port. 109 socket_instance = mock.Mock() 110 socket_instance.connect = mock.Mock() 111 socket_instance.connect.side_effect = error 112 socket.return_value = socket_instance 113 114 with self.assertRaises(rpc_client.Sl4aConnectionError): 115 sl4a_session.Sl4aSession._create_client_side_connection( 116 session, sl4a_ports.Sl4aPorts(0, 2, 3)) 117 118 def test_create_client_side_connection_hint_taken_during_fn(self): 119 """Tests sl4a_session.Sl4aSession._create_client_side_connection(). 120 121 Tests that the function will call catch an EADDRNOTAVAIL OSError and 122 call itself again, this time with a hinted port of 0 (random). 123 """ 124 session = mock.Mock() 125 session._create_client_side_connection = mock.Mock() 126 error = socket_error() 127 error.errno = errno.EADDRNOTAVAIL 128 with mock.patch('socket.socket') as socket: 129 # Throw an error when trying to bind to the hinted port. 130 socket_instance = mock.Mock() 131 socket_instance.connect = mock.Mock() 132 socket_instance.connect.side_effect = error 133 socket.return_value = socket_instance 134 135 sl4a_session.Sl4aSession._create_client_side_connection( 136 session, sl4a_ports.Sl4aPorts(0, 2, 3)) 137 138 fn = session._create_client_side_connection 139 self.assertEqual(fn.call_count, 1) 140 # Asserts that the 1st argument (Sl4aPorts) sent to the function 141 # has a client port of 0. 142 self.assertEqual(fn.call_args_list[0][0][0].client_port, 0) 143 144 def test_create_client_side_connection_re_raises_uncaught_errors(self): 145 """Tests sl4a_session.Sl4aSession._create_client_side_connection(). 146 147 Tests that the function will re-raise any socket error that does not 148 have errno.EADDRNOTAVAIL. 149 """ 150 session = mock.Mock() 151 session._create_client_side_connection = mock.Mock() 152 error = socket_error() 153 # Some error that isn't EADDRNOTAVAIL 154 error.errno = errno.ESOCKTNOSUPPORT 155 with mock.patch('socket.socket') as socket: 156 # Throw an error when trying to bind to the hinted port. 157 socket_instance = mock.Mock() 158 socket_instance.connect = mock.Mock() 159 socket_instance.connect.side_effect = error 160 socket.return_value = socket_instance 161 162 with self.assertRaises(socket_error): 163 sl4a_session.Sl4aSession._create_client_side_connection( 164 session, sl4a_ports.Sl4aPorts(0, 2, 3)) 165 166 def test_terminate_only_closes_if_not_terminated(self): 167 """Tests sl4a_session.Sl4aSession.terminate() 168 169 Tests that terminate only runs termination steps if the session has not 170 already been terminated. 171 """ 172 session = mock.Mock() 173 session._terminate_lock = mock.MagicMock() 174 session._terminated = True 175 sl4a_session.Sl4aSession.terminate(session) 176 177 self.assertFalse(session._event_dispatcher.close.called) 178 self.assertFalse(session.rpc_client.terminate.called) 179 180 def test_terminate_closes_session_first(self): 181 """Tests sl4a_session.Sl4aSession.terminate() 182 183 Tests that terminate only runs termination steps if the session has not 184 already been terminated. 185 """ 186 session = mock.Mock() 187 session._terminate_lock = mock.MagicMock() 188 session._terminated = True 189 sl4a_session.Sl4aSession.terminate(session) 190 191 self.assertFalse(session._event_dispatcher.close.called) 192 self.assertFalse(session.rpc_client.terminate.called) 193 194 195 if __name__ == '__main__': 196 unittest.main() 197
