• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
16import unittest
17
18import dnsproxy
19import httparchive
20import httpclient
21import platformsettings
22import test_utils
23
24
25class 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
122class 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
187class 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
210if __name__ == '__main__':
211  unittest.main()
212