tests/mobly/controllers/android_device_lib/jsonrpc_client_base_test.py - third_party/github.com/google/mobly - Git at Google

 # Copyright 2016 Google Inc.
 #
 # 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 json
 import socket
 import unittest
 from unittest import mock

 from mobly.controllers.android_device_lib import jsonrpc_client_base
 from tests.lib import jsonrpc_client_test_base


 class FakeRpcClient(jsonrpc_client_base.JsonRpcClientBase):

   def __init__(self):
     super().__init__(app_name='FakeRpcClient', ad=mock.Mock())


 class JsonRpcClientBaseTest(jsonrpc_client_test_base.JsonRpcClientTestBase):
   """Unit tests for mobly.controllers.android_device_lib.jsonrpc_client_base.
   """

   @mock.patch('socket.create_connection')
   def test_open_timeout_io_error(self, mock_create_connection):
     """Test socket timeout with io error

     Test that if the net socket gives an io error, then the client
     will eventually exit with an IOError.
     """
     mock_create_connection.side_effect = IOError()
     with self.assertRaises(IOError):
       client = FakeRpcClient()
       client.connect()

   @mock.patch('socket.create_connection')
   def test_connect_timeout(self, mock_create_connection):
     """Test socket timeout

     Test that a timeout exception will be raised if the socket gives a
     timeout.
     """
     mock_create_connection.side_effect = socket.timeout
     with self.assertRaises(socket.timeout):
       client = FakeRpcClient()
       client.connect()

   @mock.patch('socket.create_connection')
   def test_handshake_error(self, mock_create_connection):
     """Test error in jsonrpc handshake

     Test that if there is an error in the jsonrpc handshake then a protocol
     error will be raised.
     """
     self.setup_mock_socket_file(mock_create_connection, resp=None)
     client = FakeRpcClient()
     with self.assertRaisesRegex(
         jsonrpc_client_base.ProtocolError,
         jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_HANDSHAKE):
       client.connect()

   def test_disconnect(self):
     client = FakeRpcClient()
     mock_conn = mock.MagicMock()
     client.clear_host_port = mock.MagicMock()
     client._conn = mock_conn
     client.disconnect()
     self.assertIsNone(client._conn)
     mock_conn.close.assert_called_once_with()
     client.clear_host_port.assert_called_once_with()

   def test_disconnect_raises(self):
     client = FakeRpcClient()
     mock_conn = mock.MagicMock()
     client.clear_host_port = mock.MagicMock()
     client._conn = mock_conn
     # Explicitly making the second side_effect noop to avoid uncaught exception
     # when `__del__` is called after the test is done, which triggers
     # `disconnect`.
     mock_conn.close.side_effect = [Exception('ha'), None]
     with self.assertRaisesRegex(Exception, 'ha'):
       client.disconnect()
     client.clear_host_port.assert_called_once_with()

   def test_clear_host_port_positive(self):
     client = FakeRpcClient()
     client.host_port = 1
     client.clear_host_port()
     client._ad.adb.forward.assert_called_once_with(['--remove', 'tcp:1'])
     self.assertIsNone(client.host_port)

   def test_clear_host_port_negative(self):
     client = FakeRpcClient()
     client.host_port = None
     client.clear_host_port()
     client._ad.adb.forward.assert_not_called()

   @mock.patch('socket.create_connection')
   def test_connect_handshake(self, mock_create_connection):
     """Test client handshake

     Test that at the end of a handshake with no errors the client object
     has the correct parameters.
     """
     self.setup_mock_socket_file(mock_create_connection)
     client = FakeRpcClient()
     client.connect()
     self.assertEqual(client.uid, 1)

   @mock.patch('socket.create_connection')
   def test_connect_handshake_unknown_status(self, mock_create_connection):
     """Test handshake with unknown status response

     Test that when the handshake is given an unknown status then the client
     will not be given a uid.
     """
     self.setup_mock_socket_file(mock_create_connection,
                                 resp=self.MOCK_RESP_UNKNOWN_STATUS)
     client = FakeRpcClient()
     client.connect()
     self.assertEqual(client.uid, jsonrpc_client_base.UNKNOWN_UID)

   @mock.patch('socket.create_connection')
   def test_rpc_error_response(self, mock_create_connection):
     """Test rpc that is given an error response

     Test that when an rpc receives a response with an error will raised
     an api error.
     """
     fake_file = self.setup_mock_socket_file(mock_create_connection)

     client = FakeRpcClient()
     client.connect()

     fake_file.resp = self.MOCK_RESP_WITH_ERROR

     with self.assertRaisesRegex(jsonrpc_client_base.ApiError, '1'):
       client.some_rpc(1, 2, 3)

   @mock.patch('socket.create_connection')
   def test_rpc_callback_response(self, mock_create_connection):
     """Test rpc that is given a callback response.

     Test that when an rpc receives a callback response, a callback object is
     created correctly.
     """
     fake_file = self.setup_mock_socket_file(mock_create_connection)

     client = FakeRpcClient()
     client.connect()

     fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
     client._event_client = mock.Mock()

     callback = client.some_rpc(1, 2, 3)
     self.assertEqual(callback.ret_value, 123)
     self.assertEqual(callback._id, '1-0')

   @mock.patch('socket.create_connection')
   def test_rpc_id_mismatch(self, mock_create_connection):
     """Test rpc that returns a different id than expected

     Test that if an rpc returns with an id that is different than what
     is expected will give a protocl error.
     """
     fake_file = self.setup_mock_socket_file(mock_create_connection)

     client = FakeRpcClient()
     client.connect()

     fake_file.resp = (self.MOCK_RESP_TEMPLATE % 52).encode('utf8')

     with self.assertRaisesRegex(
         jsonrpc_client_base.ProtocolError,
         jsonrpc_client_base.ProtocolError.MISMATCHED_API_ID):
       client.some_rpc(1, 2, 3)

   @mock.patch('socket.create_connection')
   def test_rpc_no_response(self, mock_create_connection):
     """Test rpc that does not get a response

     Test that when an rpc does not get a response it throws a protocol
     error.
     """
     fake_file = self.setup_mock_socket_file(mock_create_connection)

     client = FakeRpcClient()
     client.connect()

     fake_file.resp = None

     with self.assertRaisesRegex(
         jsonrpc_client_base.ProtocolError,
         jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_SERVER):
       client.some_rpc(1, 2, 3)

   @mock.patch('socket.create_connection')
   def test_rpc_send_to_socket(self, mock_create_connection):
     """Test rpc sending and recieving

     Tests that when an rpc is sent and received the corrent data
     is used.
     """
     fake_file = self.setup_mock_socket_file(mock_create_connection)

     client = FakeRpcClient()
     client.connect()

     result = client.some_rpc(1, 2, 3)
     self.assertEqual(result, 123)

     expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
     actual = json.loads(fake_file.last_write.decode('utf-8'))

     self.assertEqual(expected, actual)

   @mock.patch('socket.create_connection')
   def test_rpc_send_to_socket_without_callback(self, mock_create_connection):
     """Test rpc sending and recieving with Rpc protocol before callback was
     added to the resp message.

     Logic is the same as test_rpc_send_to_socket.
     """
     fake_file = self.setup_mock_socket_file(
         mock_create_connection, resp=self.MOCK_RESP_WITHOUT_CALLBACK)

     client = FakeRpcClient()
     client.connect()

     result = client.some_rpc(1, 2, 3)
     self.assertEqual(result, 123)

     expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
     actual = json.loads(fake_file.last_write.decode('utf-8'))

     self.assertEqual(expected, actual)

   @mock.patch('socket.create_connection')
   def test_rpc_call_increment_counter(self, mock_create_connection):
     """Test rpc counter

     Test that with each rpc call the counter is incremented by 1.
     """
     fake_file = self.setup_mock_socket_file(mock_create_connection)

     client = FakeRpcClient()
     client.connect()

     for i in range(0, 10):
       fake_file.resp = (self.MOCK_RESP_TEMPLATE % i).encode('utf-8')
       client.some_rpc()

     self.assertEqual(next(client._counter), 10)

   @mock.patch('socket.create_connection')
   def test_rpc_verbose_logging_with_long_string(self, mock_create_connection):
     """Test rpc response fully write into DEBUG level log."""
     fake_file = self.setup_mock_socket_file(mock_create_connection)
     testing_rpc_response = self.generate_rpc_response(4000)
     fake_file.resp = testing_rpc_response

     client = FakeRpcClient()
     client.connect()

     response = client._client_receive()
     self.assertEqual(response, testing_rpc_response)

     client.log.debug.assert_called_with('Snippet received: %s',
                                         testing_rpc_response)

   @mock.patch('socket.create_connection')
   def test_rpc_truncated_logging_short_response(self, mock_create_connection):
     """Test rpc response will full logged when length is short."""
     fake_file = self.setup_mock_socket_file(mock_create_connection)
     testing_rpc_response = self.generate_rpc_response(
         int(jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH / 2))
     fake_file.resp = testing_rpc_response

     client = FakeRpcClient()
     client.connect()

     client.set_snippet_client_verbose_logging(False)
     response = client._client_receive()

     self.assertEqual(response, testing_rpc_response)
     client.log.debug.assert_called_with('Snippet received: %s',
                                         testing_rpc_response)

   @mock.patch('socket.create_connection')
   def test_rpc_truncated_logging_fit_size_response(self,
                                                    mock_create_connection):
     """Test rpc response will full logged when length is equal to threshold.
     """
     fake_file = self.setup_mock_socket_file(mock_create_connection)
     testing_rpc_response = self.generate_rpc_response(
         jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)
     fake_file.resp = testing_rpc_response

     client = FakeRpcClient()
     client.connect()

     client.set_snippet_client_verbose_logging(False)
     response = client._client_receive()

     self.assertEqual(response, testing_rpc_response)
     client.log.debug.assert_called_with('Snippet received: %s',
                                         testing_rpc_response)

   @mock.patch('socket.create_connection')
   def test_rpc_truncated_logging_long_response(self, mock_create_connection):
     """Test rpc response truncated with given length in DEBUG level log."""
     fake_file = self.setup_mock_socket_file(mock_create_connection)
     resp_len = jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH * 40
     testing_rpc_response = self.generate_rpc_response(resp_len)
     fake_file.resp = testing_rpc_response

     client = FakeRpcClient()
     client.connect()

     client.set_snippet_client_verbose_logging(False)
     response = client._client_receive()

     self.assertEqual(response, testing_rpc_response)
     # DEBUG level log should truncated by given length.
     client.log.debug.assert_called_with(
         'Snippet received: %s... %d chars are truncated',
         testing_rpc_response[:jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH],
         resp_len - jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)

   def test_close_scoket_connection(self):
     client = FakeRpcClient()
     mock_conn = mock.Mock()
     client._conn = mock_conn

     client.close_socket_connection()
     mock_conn.close.assert_called_once()
     self.assertIsNone(client._conn)

   def test_close_scoket_connection_without_connection(self):
     client = FakeRpcClient()
     client._conn = None

     client.close_socket_connection()
     self.assertIsNone(client._conn)


 if __name__ == '__main__':
   unittest.main()
	# Copyright 2016 Google Inc.
	#
	# 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 json
	import socket
	import unittest
	from unittest import mock

	from mobly.controllers.android_device_lib import jsonrpc_client_base
	from tests.lib import jsonrpc_client_test_base


	class FakeRpcClient(jsonrpc_client_base.JsonRpcClientBase):

	def __init__(self):
	super().__init__(app_name='FakeRpcClient', ad=mock.Mock())


	class JsonRpcClientBaseTest(jsonrpc_client_test_base.JsonRpcClientTestBase):
	"""Unit tests for mobly.controllers.android_device_lib.jsonrpc_client_base.
	"""

	@mock.patch('socket.create_connection')
	def test_open_timeout_io_error(self, mock_create_connection):
	"""Test socket timeout with io error

	Test that if the net socket gives an io error, then the client
	will eventually exit with an IOError.
	"""
	mock_create_connection.side_effect = IOError()
	with self.assertRaises(IOError):
	client = FakeRpcClient()
	client.connect()

	@mock.patch('socket.create_connection')
	def test_connect_timeout(self, mock_create_connection):
	"""Test socket timeout

	Test that a timeout exception will be raised if the socket gives a
	timeout.
	"""
	mock_create_connection.side_effect = socket.timeout
	with self.assertRaises(socket.timeout):
	client = FakeRpcClient()
	client.connect()

	@mock.patch('socket.create_connection')
	def test_handshake_error(self, mock_create_connection):
	"""Test error in jsonrpc handshake

	Test that if there is an error in the jsonrpc handshake then a protocol
	error will be raised.
	"""
	self.setup_mock_socket_file(mock_create_connection, resp=None)
	client = FakeRpcClient()
	with self.assertRaisesRegex(
	jsonrpc_client_base.ProtocolError,
	jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_HANDSHAKE):
	client.connect()

	def test_disconnect(self):
	client = FakeRpcClient()
	mock_conn = mock.MagicMock()
	client.clear_host_port = mock.MagicMock()
	client._conn = mock_conn
	client.disconnect()
	self.assertIsNone(client._conn)
	mock_conn.close.assert_called_once_with()
	client.clear_host_port.assert_called_once_with()

	def test_disconnect_raises(self):
	client = FakeRpcClient()
	mock_conn = mock.MagicMock()
	client.clear_host_port = mock.MagicMock()
	client._conn = mock_conn
	# Explicitly making the second side_effect noop to avoid uncaught exception
	# when `__del__` is called after the test is done, which triggers
	# `disconnect`.
	mock_conn.close.side_effect = [Exception('ha'), None]
	with self.assertRaisesRegex(Exception, 'ha'):
	client.disconnect()
	client.clear_host_port.assert_called_once_with()

	def test_clear_host_port_positive(self):
	client = FakeRpcClient()
	client.host_port = 1
	client.clear_host_port()
	client._ad.adb.forward.assert_called_once_with(['--remove', 'tcp:1'])
	self.assertIsNone(client.host_port)

	def test_clear_host_port_negative(self):
	client = FakeRpcClient()
	client.host_port = None
	client.clear_host_port()
	client._ad.adb.forward.assert_not_called()

	@mock.patch('socket.create_connection')
	def test_connect_handshake(self, mock_create_connection):
	"""Test client handshake

	Test that at the end of a handshake with no errors the client object
	has the correct parameters.
	"""
	self.setup_mock_socket_file(mock_create_connection)
	client = FakeRpcClient()
	client.connect()
	self.assertEqual(client.uid, 1)

	@mock.patch('socket.create_connection')
	def test_connect_handshake_unknown_status(self, mock_create_connection):
	"""Test handshake with unknown status response

	Test that when the handshake is given an unknown status then the client
	will not be given a uid.
	"""
	self.setup_mock_socket_file(mock_create_connection,
	resp=self.MOCK_RESP_UNKNOWN_STATUS)
	client = FakeRpcClient()
	client.connect()
	self.assertEqual(client.uid, jsonrpc_client_base.UNKNOWN_UID)

	@mock.patch('socket.create_connection')
	def test_rpc_error_response(self, mock_create_connection):
	"""Test rpc that is given an error response

	Test that when an rpc receives a response with an error will raised
	an api error.
	"""
	fake_file = self.setup_mock_socket_file(mock_create_connection)

	client = FakeRpcClient()
	client.connect()

	fake_file.resp = self.MOCK_RESP_WITH_ERROR

	with self.assertRaisesRegex(jsonrpc_client_base.ApiError, '1'):
	client.some_rpc(1, 2, 3)

	@mock.patch('socket.create_connection')
	def test_rpc_callback_response(self, mock_create_connection):
	"""Test rpc that is given a callback response.

	Test that when an rpc receives a callback response, a callback object is
	created correctly.
	"""
	fake_file = self.setup_mock_socket_file(mock_create_connection)

	client = FakeRpcClient()
	client.connect()

	fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
	client._event_client = mock.Mock()

	callback = client.some_rpc(1, 2, 3)
	self.assertEqual(callback.ret_value, 123)
	self.assertEqual(callback._id, '1-0')

	@mock.patch('socket.create_connection')
	def test_rpc_id_mismatch(self, mock_create_connection):
	"""Test rpc that returns a different id than expected

	Test that if an rpc returns with an id that is different than what
	is expected will give a protocl error.
	"""
	fake_file = self.setup_mock_socket_file(mock_create_connection)

	client = FakeRpcClient()
	client.connect()

	fake_file.resp = (self.MOCK_RESP_TEMPLATE % 52).encode('utf8')

	with self.assertRaisesRegex(
	jsonrpc_client_base.ProtocolError,
	jsonrpc_client_base.ProtocolError.MISMATCHED_API_ID):
	client.some_rpc(1, 2, 3)

	@mock.patch('socket.create_connection')
	def test_rpc_no_response(self, mock_create_connection):
	"""Test rpc that does not get a response

	Test that when an rpc does not get a response it throws a protocol
	error.
	"""
	fake_file = self.setup_mock_socket_file(mock_create_connection)

	client = FakeRpcClient()
	client.connect()

	fake_file.resp = None

	with self.assertRaisesRegex(
	jsonrpc_client_base.ProtocolError,
	jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_SERVER):
	client.some_rpc(1, 2, 3)

	@mock.patch('socket.create_connection')
	def test_rpc_send_to_socket(self, mock_create_connection):
	"""Test rpc sending and recieving

	Tests that when an rpc is sent and received the corrent data
	is used.
	"""
	fake_file = self.setup_mock_socket_file(mock_create_connection)

	client = FakeRpcClient()
	client.connect()

	result = client.some_rpc(1, 2, 3)
	self.assertEqual(result, 123)

	expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
	actual = json.loads(fake_file.last_write.decode('utf-8'))

	self.assertEqual(expected, actual)

	@mock.patch('socket.create_connection')
	def test_rpc_send_to_socket_without_callback(self, mock_create_connection):
	"""Test rpc sending and recieving with Rpc protocol before callback was
	added to the resp message.

	Logic is the same as test_rpc_send_to_socket.
	"""
	fake_file = self.setup_mock_socket_file(
	mock_create_connection, resp=self.MOCK_RESP_WITHOUT_CALLBACK)

	client = FakeRpcClient()
	client.connect()

	result = client.some_rpc(1, 2, 3)
	self.assertEqual(result, 123)

	expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
	actual = json.loads(fake_file.last_write.decode('utf-8'))

	self.assertEqual(expected, actual)

	@mock.patch('socket.create_connection')
	def test_rpc_call_increment_counter(self, mock_create_connection):
	"""Test rpc counter

	Test that with each rpc call the counter is incremented by 1.
	"""
	fake_file = self.setup_mock_socket_file(mock_create_connection)

	client = FakeRpcClient()
	client.connect()

	for i in range(0, 10):
	fake_file.resp = (self.MOCK_RESP_TEMPLATE % i).encode('utf-8')
	client.some_rpc()

	self.assertEqual(next(client._counter), 10)

	@mock.patch('socket.create_connection')
	def test_rpc_verbose_logging_with_long_string(self, mock_create_connection):
	"""Test rpc response fully write into DEBUG level log."""
	fake_file = self.setup_mock_socket_file(mock_create_connection)
	testing_rpc_response = self.generate_rpc_response(4000)
	fake_file.resp = testing_rpc_response

	client = FakeRpcClient()
	client.connect()

	response = client._client_receive()
	self.assertEqual(response, testing_rpc_response)

	client.log.debug.assert_called_with('Snippet received: %s',
	testing_rpc_response)

	@mock.patch('socket.create_connection')
	def test_rpc_truncated_logging_short_response(self, mock_create_connection):
	"""Test rpc response will full logged when length is short."""
	fake_file = self.setup_mock_socket_file(mock_create_connection)
	testing_rpc_response = self.generate_rpc_response(
	int(jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH / 2))
	fake_file.resp = testing_rpc_response

	client = FakeRpcClient()
	client.connect()

	client.set_snippet_client_verbose_logging(False)
	response = client._client_receive()

	self.assertEqual(response, testing_rpc_response)
	client.log.debug.assert_called_with('Snippet received: %s',
	testing_rpc_response)

	@mock.patch('socket.create_connection')
	def test_rpc_truncated_logging_fit_size_response(self,
	mock_create_connection):
	"""Test rpc response will full logged when length is equal to threshold.
	"""
	fake_file = self.setup_mock_socket_file(mock_create_connection)
	testing_rpc_response = self.generate_rpc_response(
	jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)
	fake_file.resp = testing_rpc_response

	client = FakeRpcClient()
	client.connect()

	client.set_snippet_client_verbose_logging(False)
	response = client._client_receive()

	self.assertEqual(response, testing_rpc_response)
	client.log.debug.assert_called_with('Snippet received: %s',
	testing_rpc_response)

	@mock.patch('socket.create_connection')
	def test_rpc_truncated_logging_long_response(self, mock_create_connection):
	"""Test rpc response truncated with given length in DEBUG level log."""
	fake_file = self.setup_mock_socket_file(mock_create_connection)
	resp_len = jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH * 40
	testing_rpc_response = self.generate_rpc_response(resp_len)
	fake_file.resp = testing_rpc_response

	client = FakeRpcClient()
	client.connect()

	client.set_snippet_client_verbose_logging(False)
	response = client._client_receive()

	self.assertEqual(response, testing_rpc_response)
	# DEBUG level log should truncated by given length.
	client.log.debug.assert_called_with(
	'Snippet received: %s... %d chars are truncated',
	testing_rpc_response[:jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH],
	resp_len - jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)

	def test_close_scoket_connection(self):
	client = FakeRpcClient()
	mock_conn = mock.Mock()
	client._conn = mock_conn

	client.close_socket_connection()
	mock_conn.close.assert_called_once()
	self.assertIsNone(client._conn)

	def test_close_scoket_connection_without_connection(self):
	client = FakeRpcClient()
	client._conn = None

	client.close_socket_connection()
	self.assertIsNone(client._conn)


	if __name__ == '__main__':
	unittest.main()