chrome_proxy_utils_test.py (9567B)
1 #!/usr/bin/env vpython3 2 # Copyright 2020 The Chromium Authors 3 # Use of this source code is governed by a BSD-style license that can be 4 # found in the LICENSE file. 5 """Tests for chrome_proxy_utils.""" 6 7 #pylint: disable=protected-access 8 9 import os 10 import unittest 11 12 from pylib.utils import chrome_proxy_utils 13 14 from devil.android import forwarder 15 from devil.android import device_utils 16 from devil.android.sdk import adb_wrapper 17 from py_utils import ts_proxy_server 18 from py_utils import webpagereplay_go_server 19 20 import mock # pylint: disable=import-error 21 22 23 def _DeviceUtilsMock(test_serial, is_ready=True): 24 """Returns a DeviceUtils instance based on given serial.""" 25 adb = mock.Mock(spec=adb_wrapper.AdbWrapper) 26 adb.__str__ = mock.Mock(return_value=test_serial) 27 adb.GetDeviceSerial.return_value = test_serial 28 adb.is_ready = is_ready 29 return device_utils.DeviceUtils(adb) 30 31 32 class ChromeProxySessionTest(unittest.TestCase): 33 """Unittest for ChromeProxySession.""" 34 35 #pylint: disable=no-self-use 36 37 @mock.patch.object(forwarder.Forwarder, 'Map') 38 @mock.patch.object(chrome_proxy_utils.WPRServer, 'StartServer') 39 @mock.patch.object(ts_proxy_server.TsProxyServer, 'StartServer') 40 @mock.patch.object(ts_proxy_server.TsProxyServer, 'UpdateOutboundPorts') 41 @mock.patch.object(ts_proxy_server.TsProxyServer, 'UpdateTrafficSettings') 42 @mock.patch('py_utils.ts_proxy_server.TsProxyServer.port', 43 new_callable=mock.PropertyMock) 44 def test_Start(self, port_mock, traffic_setting_mock, outboundport_mock, 45 start_server_mock, wpr_mock, forwarder_mock): 46 chrome_proxy = chrome_proxy_utils.ChromeProxySession(4) 47 chrome_proxy._wpr_server._host_http_port = 1 48 chrome_proxy._wpr_server._host_https_port = 2 49 port_mock.return_value = 3 50 device = _DeviceUtilsMock('01234') 51 chrome_proxy.Start(device, 'abc') 52 53 forwarder_mock.assert_called_once_with([(4, 3)], device) 54 wpr_mock.assert_called_once_with('abc') 55 start_server_mock.assert_called_once() 56 outboundport_mock.assert_called_once_with(http_port=1, https_port=2) 57 traffic_setting_mock.assert_called_once_with(download_bandwidth_kbps=72000, 58 round_trip_latency_ms=100, 59 upload_bandwidth_kbps=72000) 60 port_mock.assert_called_once() 61 62 @mock.patch.object(forwarder.Forwarder, 'UnmapDevicePort') 63 @mock.patch.object(chrome_proxy_utils.WPRServer, 'StopServer') 64 @mock.patch.object(ts_proxy_server.TsProxyServer, 'StopServer') 65 def test_Stop(self, ts_proxy_mock, wpr_mock, forwarder_mock): 66 chrome_proxy = chrome_proxy_utils.ChromeProxySession(4) 67 device = _DeviceUtilsMock('01234') 68 chrome_proxy.wpr_record_mode = True 69 chrome_proxy._wpr_server._archive_path = 'abc' 70 chrome_proxy.Stop(device) 71 72 forwarder_mock.assert_called_once_with(4, device) 73 wpr_mock.assert_called_once_with() 74 ts_proxy_mock.assert_called_once_with() 75 76 #pylint: enable=no-self-use 77 78 @mock.patch.object(forwarder.Forwarder, 'UnmapDevicePort') 79 @mock.patch.object(webpagereplay_go_server.ReplayServer, 'StopServer') 80 @mock.patch.object(ts_proxy_server.TsProxyServer, 'StopServer') 81 def test_Stop_WithProperties(self, ts_proxy_mock, wpr_mock, forwarder_mock): 82 chrome_proxy = chrome_proxy_utils.ChromeProxySession(4) 83 chrome_proxy._wpr_server._server = webpagereplay_go_server.ReplayServer( 84 os.path.abspath(__file__), chrome_proxy_utils.PROXY_HOST_IP, 0, 0, []) 85 chrome_proxy._wpr_server._archive_path = os.path.abspath(__file__) 86 device = _DeviceUtilsMock('01234') 87 chrome_proxy.wpr_record_mode = True 88 chrome_proxy.Stop(device) 89 90 forwarder_mock.assert_called_once_with(4, device) 91 wpr_mock.assert_called_once_with() 92 ts_proxy_mock.assert_called_once_with() 93 self.assertFalse(chrome_proxy.wpr_replay_mode) 94 self.assertEqual(chrome_proxy.wpr_archive_path, os.path.abspath(__file__)) 95 96 def test_SetWPRRecordMode(self): 97 chrome_proxy = chrome_proxy_utils.ChromeProxySession(4) 98 chrome_proxy.wpr_record_mode = True 99 self.assertTrue(chrome_proxy._wpr_server.record_mode) 100 self.assertTrue(chrome_proxy.wpr_record_mode) 101 self.assertFalse(chrome_proxy.wpr_replay_mode) 102 103 chrome_proxy.wpr_record_mode = False 104 self.assertFalse(chrome_proxy._wpr_server.record_mode) 105 self.assertFalse(chrome_proxy.wpr_record_mode) 106 self.assertTrue(chrome_proxy.wpr_replay_mode) 107 108 def test_SetWPRArchivePath(self): 109 chrome_proxy = chrome_proxy_utils.ChromeProxySession(4) 110 chrome_proxy._wpr_server._archive_path = 'abc' 111 self.assertEqual(chrome_proxy.wpr_archive_path, 'abc') 112 113 def test_UseDefaultDeviceProxyPort(self): 114 chrome_proxy = chrome_proxy_utils.ChromeProxySession() 115 expected_flags = [ 116 '--ignore-certificate-errors-spki-list=' 117 'PhrPvGIaAMmd29hj8BCZOq096yj7uMpRNHpn5PDxI6I=', 118 '--proxy-server=socks5://localhost:1080' 119 ] 120 self.assertEqual(chrome_proxy.device_proxy_port, 1080) 121 self.assertListEqual(chrome_proxy.GetFlags(), expected_flags) 122 123 def test_UseNewDeviceProxyPort(self): 124 chrome_proxy = chrome_proxy_utils.ChromeProxySession(1) 125 expected_flags = [ 126 '--ignore-certificate-errors-spki-list=' 127 'PhrPvGIaAMmd29hj8BCZOq096yj7uMpRNHpn5PDxI6I=', 128 '--proxy-server=socks5://localhost:1' 129 ] 130 self.assertEqual(chrome_proxy.device_proxy_port, 1) 131 self.assertListEqual(chrome_proxy.GetFlags(), expected_flags) 132 133 134 class WPRServerTest(unittest.TestCase): 135 @mock.patch('py_utils.webpagereplay_go_server.ReplayServer') 136 def test_StartSever_fresh_replaymode(self, wpr_mock): 137 wpr_server = chrome_proxy_utils.WPRServer() 138 wpr_archive_file = os.path.abspath(__file__) 139 wpr_server.StartServer(wpr_archive_file) 140 141 wpr_mock.assert_called_once_with(wpr_archive_file, 142 '127.0.0.1', 143 http_port=0, 144 https_port=0, 145 replay_options=[]) 146 147 self.assertEqual(wpr_server._archive_path, wpr_archive_file) 148 self.assertTrue(wpr_server._server) 149 150 @mock.patch('py_utils.webpagereplay_go_server.ReplayServer') 151 def test_StartSever_fresh_recordmode(self, wpr_mock): 152 wpr_server = chrome_proxy_utils.WPRServer() 153 wpr_server.record_mode = True 154 wpr_server.StartServer(os.path.abspath(__file__)) 155 wpr_archive_file = os.path.abspath(__file__) 156 157 wpr_mock.assert_called_once_with(wpr_archive_file, 158 '127.0.0.1', 159 http_port=0, 160 https_port=0, 161 replay_options=['--record']) 162 163 self.assertEqual(wpr_server._archive_path, os.path.abspath(__file__)) 164 self.assertTrue(wpr_server._server) 165 166 #pylint: disable=no-self-use 167 168 @mock.patch.object(webpagereplay_go_server.ReplayServer, 'StartServer') 169 def test_StartSever_recordmode(self, start_server_mock): 170 wpr_server = chrome_proxy_utils.WPRServer() 171 start_server_mock.return_value = {'http': 1, 'https': 2} 172 wpr_server.StartServer(os.path.abspath(__file__)) 173 174 start_server_mock.assert_called_once() 175 self.assertEqual(wpr_server._host_http_port, 1) 176 self.assertEqual(wpr_server._host_https_port, 2) 177 self.assertEqual(wpr_server._archive_path, os.path.abspath(__file__)) 178 self.assertTrue(wpr_server._server) 179 180 @mock.patch.object(webpagereplay_go_server.ReplayServer, 'StartServer') 181 def test_StartSever_reuseServer(self, start_server_mock): 182 wpr_server = chrome_proxy_utils.WPRServer() 183 wpr_server._server = webpagereplay_go_server.ReplayServer( 184 os.path.abspath(__file__), 185 chrome_proxy_utils.PROXY_HOST_IP, 186 http_port=0, 187 https_port=0, 188 replay_options=[]) 189 wpr_server._archive_path = os.path.abspath(__file__) 190 wpr_server.StartServer(os.path.abspath(__file__)) 191 start_server_mock.assert_not_called() 192 193 @mock.patch.object(webpagereplay_go_server.ReplayServer, 'StartServer') 194 @mock.patch.object(webpagereplay_go_server.ReplayServer, 'StopServer') 195 def test_StartSever_notReuseServer(self, stop_server_mock, start_server_mock): 196 wpr_server = chrome_proxy_utils.WPRServer() 197 wpr_server._server = webpagereplay_go_server.ReplayServer( 198 os.path.abspath(__file__), 199 chrome_proxy_utils.PROXY_HOST_IP, 200 http_port=0, 201 https_port=0, 202 replay_options=[]) 203 wpr_server._archive_path = '' 204 wpr_server.StartServer(os.path.abspath(__file__)) 205 start_server_mock.assert_called_once() 206 stop_server_mock.assert_called_once() 207 208 #pylint: enable=no-self-use 209 210 @mock.patch.object(webpagereplay_go_server.ReplayServer, 'StopServer') 211 def test_StopServer(self, stop_server_mock): 212 wpr_server = chrome_proxy_utils.WPRServer() 213 wpr_server._server = webpagereplay_go_server.ReplayServer( 214 os.path.abspath(__file__), 215 chrome_proxy_utils.PROXY_HOST_IP, 216 http_port=0, 217 https_port=0, 218 replay_options=[]) 219 wpr_server.StopServer() 220 stop_server_mock.assert_called_once() 221 self.assertFalse(wpr_server._server) 222 self.assertFalse(wpr_server._archive_path) 223 self.assertFalse(wpr_server.http_port) 224 self.assertFalse(wpr_server.https_port) 225 226 def test_SetWPRRecordMode(self): 227 wpr_server = chrome_proxy_utils.WPRServer() 228 wpr_server.record_mode = True 229 self.assertTrue(wpr_server.record_mode) 230 wpr_server.record_mode = False 231 self.assertFalse(wpr_server.record_mode) 232 233 234 if __name__ == '__main__': 235 unittest.main(verbosity=2)