# # Copyright (C) 2016 The Android Open Source Project # # 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 socket import unittest import logging import errno from socket import error as socket_error from vts.runners.host import errors from vts.proto import AndroidSystemControlMessage_pb2 as SysMsg_pb2 from vts.runners.host.tcp_server import callback_server HOST, PORT = "localhost", 0 ERROR_PORT = 380 # port at which we test the error case. class TestMethods(unittest.TestCase): """This class defines unit test methods. The common scenarios are when we wish to test the whether we are able to receive the expected data from the server; and whether we receive the correct error when we try to connect to server from a wrong port. Attributes: _callback_server: an instance of CallbackServer that is used to start and stop the TCP server. _counter: This is used to keep track of number of calls made to the callback function. """ _callback_server = None _counter = 0 def setUp(self): """This function initiates starting the server in CallbackServer.""" self._callback_server = callback_server.CallbackServer() self._callback_server.Start() def tearDown(self): """To initiate shutdown of the server. This function calls the callback_server.CallbackServer.Stop which shutdowns the server. """ self._callback_server.Stop() def DoErrorCase(self): """Unit test for Error case. This function tests the cases that throw exception. e.g sending requests to port 25. Raises: ConnectionRefusedError: ConnectionRefusedError occurred in test_ErrorCase(). """ host = self._callback_server.ip # Create a socket (SOCK_STREAM means a TCP socket) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: # Connect to server; this should result in Connection refused error sock.connect((host, ERROR_PORT)) except socket_error as e: # We are comparing the error number of the error we expect and # the error that we get. # Test fails if ConnectionRefusedError is not raised at this step. if e.errno == errno.ECONNREFUSED: raise errors.ConnectionRefusedError # Test is a success here else: raise e # Test fails, since ConnectionRefusedError was expected finally: sock.close() def ConnectToServer(self, func_id): """This function creates a connection to TCP server and sends/receives message. Args: func_id: This is the unique key corresponding to a function and also the id field of the request_message that we send to the server. Returns: response_message: The object that the TCP host returns. Raises: TcpServerConnectionError: Exception occurred while stopping server. """ # This object is sent to the TCP host request_message = SysMsg_pb2.AndroidSystemCallbackRequestMessage() request_message.id = func_id # The response in string format that we receive from host received_message = "" # The final object that this function returns response_message = SysMsg_pb2.AndroidSystemCallbackResponseMessage() # Create a socket (SOCK_STREAM means a TCP socket) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) host = self._callback_server.ip port = self._callback_server.port logging.debug('Sending Request to host %s using port %s', host, port) try: # Connect to server and send request_message sock.connect((host, port)) message = request_message.SerializeToString() sock.sendall(str(len(message)) + "\n" + message) logging.debug("Sent: %s", message) # Receive request_message from the server and shut down received_message = sock.recv(1024) response_message.ParseFromString(received_message) logging.debug('Received: %s', received_message) except socket_error as e: logging.error(e) raise errors.TcpServerConnectionError('Exception occurred.') finally: sock.close() return response_message def testDoErrorCase(self): """Unit test for error cases.""" with self.assertRaises(errors.ConnectionRefusedError): self.DoErrorCase() def testCallback(self): """Tests two callback use cases.""" self.TestNormalCase() self.TestDoRegisterCallback() def TestNormalCase(self): """Tests the normal request to TCPServer. This function sends the request to the Tcp server where the request should be a success. This function also checks the register callback feature by ensuring that callback_func() is called and the value of the global counter is increased by one. """ def callback_func(): self._counter += 1 # Function should be registered with RegisterCallback func_id = self._callback_server.RegisterCallback(callback_func) self.assertEqual(func_id, '0') # Capture the previous value of global counter prev_value = self._counter # Connect to server response_message = self.ConnectToServer(func_id) # Confirm whether the callback_func() was called thereby increasing # value of global counter by 1 self.assertEqual(self._counter, prev_value + 1) # Also confirm if query resulted in a success self.assertEqual(response_message.response_code, SysMsg_pb2.SUCCESS) def TestDoRegisterCallback(self): """Checks the register callback functionality of the Server. This function checks whether the value of global counter remains same if function is not registered. It also checks whether it's incremented by 1 when the function is registered. """ def callback_func(): self._counter += 1 # Capture the previous value of global counter prev_value = self._counter # Function should be registered with RegisterCallback func_id = self._callback_server.RegisterCallback(callback_func) found_func_id = self._callback_server.GetCallbackId(callback_func) self.assertEqual(func_id, found_func_id) # Connect to server response_message = self.ConnectToServer(func_id) # Confirm whether the callback_func() was not called. self.assertEqual(self._counter, prev_value + 1) # also confirm the error message self.assertEqual(response_message.response_code, SysMsg_pb2.SUCCESS) # Now unregister the function and check again # Function should be unregistered with UnegisterCallback # and the key should also be present self._callback_server.UnregisterCallback(func_id) # Capture the previous value of global counter prev_value = self._counter # Connect to server response_message = self.ConnectToServer(func_id) # Confirm whether the callback_func() was not called. self.assertEqual(self._counter, prev_value) # also confirm the error message self.assertEqual(response_message.response_code, SysMsg_pb2.FAIL) if __name__ == '__main__': unittest.main()