1 #!/usr/bin/python 2 # -*- coding: utf-8 -*- 3 # Copyright 2018 The Chromium OS Authors. All rights reserved. 4 # Use of this source code is governed by a BSD-style license that can be 5 # found in the LICENSE file. 6 7 """Tests for gs_cache_client.""" 8 9 from __future__ import absolute_import 10 from __future__ import division 11 from __future__ import print_function 12 13 import itertools 14 import json 15 import requests 16 import unittest 17 18 import mock 19 20 import common 21 from autotest_lib.client.common_lib import error 22 from autotest_lib.client.common_lib.cros import gs_cache_client 23 24 25 # pylint: disable=unused-argument 26 class ApiTest(unittest.TestCase): 27 """Test class for _GsCacheAPI.""" 28 def setUp(self): 29 self.api = gs_cache_client._GsCacheAPI('localhost') 30 31 def test_extract_via_ssh(self): 32 """Test extracting via ssh.""" 33 gs_cache_client._USE_SSH_CONNECTION = True 34 self.api._is_in_restricted_subnet = True 35 with mock.patch('autotest_lib.client.common_lib.utils.run') as m: 36 m.return_value.stdout = '{}' 37 result = self.api.extract( 38 gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 'archive', 39 'file') 40 self.assertEqual(result, {}) 41 42 def test_extract_via_http(self): 43 """Test extracting via http.""" 44 with mock.patch('requests.get') as m: 45 m.return_value = mock.MagicMock(ok=True, content='{}') 46 result = self.api.extract( 47 gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 'archive', 48 'file') 49 self.assertEqual(result, {}) 50 51 def test_extract_many_files_via_http(self): 52 """Test extracting many files via http.""" 53 with mock.patch('requests.get') as m: 54 m.return_value = mock.MagicMock(ok=True, content='{}') 55 result = self.api.extract( 56 gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 'archive', 57 ['the_file'] * 1000) 58 self.assertEqual(result, {}) 59 self.assertTrue(m.call_count > 1) 60 61 @mock.patch('time.sleep') 62 @mock.patch('time.time', side_effect=itertools.cycle([0, 400])) 63 def test_extract_via_ssh_has_error(self, *args): 64 """Test extracting via ssh when has errors.""" 65 gs_cache_client._USE_SSH_CONNECTION = True 66 self.api._is_in_restricted_subnet = True 67 with mock.patch('autotest_lib.client.common_lib.utils.run') as m: 68 m.side_effect = error.CmdError('cmd', mock.MagicMock(exit_status=1)) 69 with self.assertRaises(gs_cache_client.CommunicationError): 70 self.api.extract(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 71 'archive', 'file') 72 73 m.side_effect = error.CmdError('cmd', mock.MagicMock( 74 exit_status=gs_cache_client._CURL_RC_CANNOT_CONNECT_TO_HOST) 75 ) 76 with self.assertRaises(gs_cache_client.NoGsCacheServerError): 77 self.api.extract(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 78 'archive', 'file') 79 80 m.side_effect = None 81 m.return_value.stdout = '...' 82 with self.assertRaises(gs_cache_client.ResponseContentError): 83 self.api.extract(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 84 'archive', 'file') 85 86 @mock.patch('time.sleep') 87 @mock.patch('time.time', side_effect=itertools.cycle([0, 400])) 88 def test_extract_via_http_has_error(self, *args): 89 """Test extracting via http when has errors.""" 90 with mock.patch('requests.get') as m: 91 m.return_value = mock.MagicMock(ok=False) 92 with self.assertRaises(gs_cache_client.CommunicationError): 93 self.api.extract(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 94 'archive', 'file') 95 96 m.return_value = mock.MagicMock(ok=True, content='...') 97 with self.assertRaises(gs_cache_client.ResponseContentError): 98 self.api.extract(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 99 'archive', 'file') 100 101 m.side_effect = requests.ConnectionError('Gs Cache is not running.') 102 with self.assertRaises(gs_cache_client.NoGsCacheServerError): 103 self.api.extract(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 104 'archive', 'file') 105 106 107 class ClientTest(unittest.TestCase): 108 """Test class for GsCacheClient.""" 109 def setUp(self): 110 self.api = mock.MagicMock(server_netloc='api_netloc') 111 self.dev_server = mock.MagicMock() 112 self.client = gs_cache_client.GsCacheClient(self.dev_server, self.api) 113 114 def test_list_suite_controls_in_map(self): 115 """Test list controls of a suite in map file.""" 116 build = 'release/build' 117 suite = 'suite' 118 control_files = ['suite_control', 'control.1'] 119 120 map_file_name = 'autotest/test_suites/suite_to_control_file_map' 121 self.api.extract.return_value = { 122 map_file_name: json.dumps({suite: control_files})} 123 self.client.list_suite_controls(build, suite) 124 125 expected_calls = [ 126 mock.call(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 127 build + '/test_suites.tar.bz2', [map_file_name]), 128 mock.call(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 129 build + '/control_files.tar', 130 ['autotest/' + c for c in control_files]) 131 ] 132 self.assertListEqual(self.api.extract.call_args_list, expected_calls) 133 134 def test_list_suite_controls_not_in_map(self): 135 """Test list controls of a suite not in map file.""" 136 build = 'release/build' 137 suite = 'suite' 138 map_file_name = 'autotest/test_suites/suite_to_control_file_map' 139 140 self.api.extract.return_value = {map_file_name: json.dumps({})} 141 self.client.list_suite_controls(build, suite) 142 143 expected_calls = [ 144 mock.call(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 145 build + '/test_suites.tar.bz2', [map_file_name]), 146 mock.call(gs_cache_client._CROS_IMAGE_ARCHIVE_BUCKET, 147 build + '/control_files.tar', 148 ['*/control', '*/control.*']) 149 ] 150 self.assertListEqual(self.api.extract.call_args_list, expected_calls) 151 152 def test_fall_back_to_dev_server(self): 153 """Test falling back to calls of dev_server.""" 154 self.client._list_suite_controls = mock.MagicMock( 155 side_effect=gs_cache_client.CommunicationError() 156 ) 157 self.client.list_suite_controls('build', 'suite') 158 self.dev_server.list_suite_controls.assert_called() 159 160 def test_fall_back_to_dev_server_on_content_error(self): 161 """Test falling back to calls of dev_server on wrong content.""" 162 self.api.extract.return_value = {k: 'xx' for k in range(999)} 163 self.client.list_suite_controls('build', 'suite') 164 self.dev_server.list_suite_controls.assert_called() 165 166 167 if __name__ == '__main__': 168 unittest.main() 169
