| OLD | NEW |
| (Empty) |
| 1 #!/usr/bin/env python | |
| 2 # Copyright 2012 Google Inc. All Rights Reserved. | |
| 3 # | |
| 4 # Licensed under the Apache License, Version 2.0 (the "License"); | |
| 5 # you may not use this file except in compliance with the License. | |
| 6 # You may obtain a copy of the License at | |
| 7 # | |
| 8 # http://www.apache.org/licenses/LICENSE-2.0 | |
| 9 # | |
| 10 # Unless required by applicable law or agreed to in writing, software | |
| 11 # distributed under the License is distributed on an "AS IS" BASIS, | |
| 12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| 13 # See the License for the specific language governing permissions and | |
| 14 # limitations under the License. | |
| 15 | |
| 16 import unittest | |
| 17 | |
| 18 import dnsproxy | |
| 19 import httparchive | |
| 20 import httpclient | |
| 21 import platformsettings | |
| 22 import test_utils | |
| 23 | |
| 24 | |
| 25 class RealHttpFetchTest(unittest.TestCase): | |
| 26 | |
| 27 # Initialize test data | |
| 28 CONTENT_TYPE = 'content-type: image/x-icon' | |
| 29 COOKIE_1 = ('Set-Cookie: GMAIL_IMP=EXPIRED; ' | |
| 30 'Expires=Thu, 12-Jul-2012 22:41:22 GMT; ' | |
| 31 'Path=/mail; Secure') | |
| 32 COOKIE_2 = ('Set-Cookie: GMAIL_STAT_205a=EXPIRED; ' | |
| 33 'Expires=Thu, 12-Jul-2012 22:42:24 GMT; ' | |
| 34 'Path=/mail; Secure') | |
| 35 FIRST_LINE = 'fake-header: first line' | |
| 36 SECOND_LINE = ' second line' | |
| 37 THIRD_LINE = '\tthird line' | |
| 38 BAD_HEADER = 'this is a bad header' | |
| 39 | |
| 40 def test__GetHeaderNameValueBasic(self): | |
| 41 """Test _GetHeaderNameValue with normal header.""" | |
| 42 | |
| 43 real_http_fetch = httpclient.RealHttpFetch | |
| 44 name_value = real_http_fetch._GetHeaderNameValue(self.CONTENT_TYPE) | |
| 45 self.assertEqual(name_value, ('content-type', 'image/x-icon')) | |
| 46 | |
| 47 def test__GetHeaderNameValueLowercasesName(self): | |
| 48 """_GetHeaderNameValue lowercases header name.""" | |
| 49 | |
| 50 real_http_fetch = httpclient.RealHttpFetch | |
| 51 header = 'X-Google-Gfe-Backend-Request-Info: eid=1KMAUMeiK4eMiAL52YyMBg' | |
| 52 expected = ('x-google-gfe-backend-request-info', | |
| 53 'eid=1KMAUMeiK4eMiAL52YyMBg') | |
| 54 name_value = real_http_fetch._GetHeaderNameValue(header) | |
| 55 self.assertEqual(name_value, expected) | |
| 56 | |
| 57 def test__GetHeaderNameValueBadLineGivesNone(self): | |
| 58 """_GetHeaderNameValue returns None for a header in wrong format.""" | |
| 59 | |
| 60 real_http_fetch = httpclient.RealHttpFetch | |
| 61 name_value = real_http_fetch._GetHeaderNameValue(self.BAD_HEADER) | |
| 62 self.assertIsNone(name_value) | |
| 63 | |
| 64 def test__ToTuplesBasic(self): | |
| 65 """Test _ToTuples with normal input.""" | |
| 66 | |
| 67 real_http_fetch = httpclient.RealHttpFetch | |
| 68 headers = [self.CONTENT_TYPE, self.COOKIE_1, self.FIRST_LINE] | |
| 69 result = real_http_fetch._ToTuples(headers) | |
| 70 expected = [('content-type', 'image/x-icon'), | |
| 71 ('set-cookie', self.COOKIE_1[12:]), | |
| 72 ('fake-header', 'first line')] | |
| 73 self.assertEqual(result, expected) | |
| 74 | |
| 75 def test__ToTuplesMultipleHeadersWithSameName(self): | |
| 76 """Test mulitple headers with the same name.""" | |
| 77 | |
| 78 real_http_fetch = httpclient.RealHttpFetch | |
| 79 headers = [self.CONTENT_TYPE, self.COOKIE_1, self.COOKIE_2, self.FIRST_LINE] | |
| 80 result = real_http_fetch._ToTuples(headers) | |
| 81 expected = [('content-type', 'image/x-icon'), | |
| 82 ('set-cookie', self.COOKIE_1[12:]), | |
| 83 ('set-cookie', self.COOKIE_2[12:]), | |
| 84 ('fake-header', 'first line')] | |
| 85 self.assertEqual(result, expected) | |
| 86 | |
| 87 def test__ToTuplesAppendsContinuationLine(self): | |
| 88 """Test continuation line is handled.""" | |
| 89 | |
| 90 real_http_fetch = httpclient.RealHttpFetch | |
| 91 headers = [self.CONTENT_TYPE, self.COOKIE_1, self.FIRST_LINE, | |
| 92 self.SECOND_LINE, self.THIRD_LINE] | |
| 93 result = real_http_fetch._ToTuples(headers) | |
| 94 expected = [('content-type', 'image/x-icon'), | |
| 95 ('set-cookie', self.COOKIE_1[12:]), | |
| 96 ('fake-header', 'first line\n second line\n third line')] | |
| 97 self.assertEqual(result, expected) | |
| 98 | |
| 99 def test__ToTuplesIgnoresBadHeader(self): | |
| 100 """Test bad header is ignored.""" | |
| 101 | |
| 102 real_http_fetch = httpclient.RealHttpFetch | |
| 103 bad_headers = [self.CONTENT_TYPE, self.BAD_HEADER, self.COOKIE_1] | |
| 104 expected = [('content-type', 'image/x-icon'), | |
| 105 ('set-cookie', self.COOKIE_1[12:])] | |
| 106 result = real_http_fetch._ToTuples(bad_headers) | |
| 107 self.assertEqual(result, expected) | |
| 108 | |
| 109 def test__ToTuplesIgnoresMisplacedContinuationLine(self): | |
| 110 """Test misplaced continuation line is ignored.""" | |
| 111 | |
| 112 real_http_fetch = httpclient.RealHttpFetch | |
| 113 misplaced_headers = [self.THIRD_LINE, self.CONTENT_TYPE, | |
| 114 self.COOKIE_1, self.FIRST_LINE, self.SECOND_LINE] | |
| 115 result = real_http_fetch._ToTuples(misplaced_headers) | |
| 116 expected = [('content-type', 'image/x-icon'), | |
| 117 ('set-cookie', self.COOKIE_1[12:]), | |
| 118 ('fake-header', 'first line\n second line')] | |
| 119 self.assertEqual(result, expected) | |
| 120 | |
| 121 | |
| 122 class RealHttpFetchGetConnectionTest(unittest.TestCase): | |
| 123 """Test that a connection is made with request IP/port or proxy IP/port.""" | |
| 124 | |
| 125 def setUp(self): | |
| 126 def real_dns_lookup(host): | |
| 127 return { | |
| 128 'example.com': '127.127.127.127', | |
| 129 'proxy.com': '2.2.2.2', | |
| 130 }[host] | |
| 131 self.fetch = httpclient.RealHttpFetch(real_dns_lookup) | |
| 132 self.https_proxy = None | |
| 133 self.http_proxy = None | |
| 134 def get_proxy(is_ssl): | |
| 135 return self.https_proxy if is_ssl else self.http_proxy | |
| 136 self.fetch._get_system_proxy = get_proxy | |
| 137 | |
| 138 def set_http_proxy(self, host, port): | |
| 139 self.http_proxy = platformsettings.SystemProxy(host, port) | |
| 140 | |
| 141 def set_https_proxy(self, host, port): | |
| 142 self.https_proxy = platformsettings.SystemProxy(host, port) | |
| 143 | |
| 144 def test_get_connection_without_proxy_connects_to_host_ip(self): | |
| 145 """HTTP connection with no proxy connects to host IP.""" | |
| 146 self.set_http_proxy(host=None, port=None) | |
| 147 connection = self.fetch._get_connection('example.com', None, is_ssl=False) | |
| 148 self.assertEqual('127.127.127.127', connection.host) | |
| 149 self.assertEqual(80, connection.port) # default HTTP port | |
| 150 | |
| 151 def test_get_connection_without_proxy_uses_nondefault_request_port(self): | |
| 152 """HTTP connection with no proxy connects with request port.""" | |
| 153 self.set_https_proxy(host=None, port=None) | |
| 154 connection = self.fetch._get_connection('example.com', 8888, is_ssl=False) | |
| 155 self.assertEqual('127.127.127.127', connection.host) | |
| 156 self.assertEqual(8888, connection.port) # request HTTP port | |
| 157 | |
| 158 def test_get_connection_with_proxy_uses_proxy_port(self): | |
| 159 """HTTP connection with proxy connects used proxy port.""" | |
| 160 self.set_http_proxy(host='proxy.com', port=None) | |
| 161 connection = self.fetch._get_connection('example.com', 8888, is_ssl=False) | |
| 162 self.assertEqual('2.2.2.2', connection.host) # proxy IP | |
| 163 self.assertEqual(80, connection.port) # proxy port (default HTTP) | |
| 164 | |
| 165 def test_ssl_get_connection_without_proxy_connects_to_host_ip(self): | |
| 166 """HTTPS (SSL) connection with no proxy connects to host IP.""" | |
| 167 self.set_https_proxy(host=None, port=None) | |
| 168 connection = self.fetch._get_connection('example.com', None, is_ssl=True) | |
| 169 self.assertEqual('127.127.127.127', connection.host) | |
| 170 self.assertEqual(443, connection.port) # default SSL port | |
| 171 | |
| 172 def test_ssl_get_connection_with_proxy_connects_to_proxy_ip(self): | |
| 173 """HTTPS (SSL) connection with proxy connects to proxy IP.""" | |
| 174 self.set_https_proxy(host='proxy.com', port=8443) | |
| 175 connection = self.fetch._get_connection('example.com', None, is_ssl=True) | |
| 176 self.assertEqual('2.2.2.2', connection.host) # proxy IP | |
| 177 self.assertEqual(8443, connection.port) # SSL proxy port | |
| 178 | |
| 179 def test_ssl_get_connection_with_proxy_tunnels_to_host(self): | |
| 180 """HTTPS (SSL) connection with proxy tunnels to target host.""" | |
| 181 self.set_https_proxy(host='proxy.com', port=8443) | |
| 182 connection = self.fetch._get_connection('example.com', None, is_ssl=True) | |
| 183 self.assertEqual('example.com', connection._tunnel_host) # host name | |
| 184 self.assertEqual(None, connection._tunnel_port) # host port | |
| 185 | |
| 186 | |
| 187 class ActualNetworkFetchTest(test_utils.RealNetworkFetchTest): | |
| 188 | |
| 189 def testFetchNonSSLRequest(self): | |
| 190 real_dns_lookup = dnsproxy.RealDnsLookup( | |
| 191 name_servers=[platformsettings.get_original_primary_nameserver()]) | |
| 192 fetch = httpclient.RealHttpFetch(real_dns_lookup) | |
| 193 request = httparchive.ArchivedHttpRequest( | |
| 194 command='GET', host='google.com', full_path='/search?q=dogs', | |
| 195 request_body=None, headers={}, is_ssl=False) | |
| 196 response = fetch(request) | |
| 197 self.assertIsNotNone(response) | |
| 198 | |
| 199 def testFetchSSLRequest(self): | |
| 200 real_dns_lookup = dnsproxy.RealDnsLookup( | |
| 201 name_servers=[platformsettings.get_original_primary_nameserver()]) | |
| 202 fetch = httpclient.RealHttpFetch(real_dns_lookup) | |
| 203 request = httparchive.ArchivedHttpRequest( | |
| 204 command='GET', host='google.com', full_path='/search?q=dogs', | |
| 205 request_body=None, headers={}, is_ssl=True) | |
| 206 response = fetch(request) | |
| 207 self.assertIsNotNone(response) | |
| 208 | |
| 209 | |
| 210 if __name__ == '__main__': | |
| 211 unittest.main() | |
| OLD | NEW |