tools/telemetry/third_party/webpagereplay/httparchive_test.py

   1 #!/usr/bin/env python
   2 # Copyright 2011 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 calendar
  17 import email.utils
  18 import httparchive
  19 import unittest
  20
  21
  22 def create_request(headers):
  23   return httparchive.ArchivedHttpRequest(
  24       'GET', 'www.test.com', '/', None, headers)
  25
  26 def create_response(headers):
  27   return httparchive.ArchivedHttpResponse(
  28       11, 200, 'OK', headers, '')
  29
  30
  31 class HttpArchiveTest(unittest.TestCase):
  32
  33   REQUEST_HEADERS = {}
  34   REQUEST = create_request(REQUEST_HEADERS)
  35
  36   # Used for if-(un)modified-since checks
  37   DATE_PAST = 'Wed, 13 Jul 2011 03:58:08 GMT'
  38   DATE_PRESENT = 'Wed, 20 Jul 2011 04:58:08 GMT'
  39   DATE_FUTURE = 'Wed, 27 Jul 2011 05:58:08 GMT'
  40   DATE_INVALID = 'This is an invalid date!!'
  41
  42   # etag values
  43   ETAG_VALID = 'etag'
  44   ETAG_INVALID = 'This is an invalid etag value!!'
  45
  46   RESPONSE_HEADERS = [('last-modified', DATE_PRESENT), ('etag', ETAG_VALID)]
  47   RESPONSE = create_response(RESPONSE_HEADERS)
  48
  49   def setUp(self):
  50     self.archive = httparchive.HttpArchive()
  51     self.archive[self.REQUEST] = self.RESPONSE
  52
  53     # Also add an identical POST request for testing
  54     request = httparchive.ArchivedHttpRequest(
  55         'POST', 'www.test.com', '/', None, self.REQUEST_HEADERS)
  56     self.archive[request] = self.RESPONSE
  57
  58   def tearDown(self):
  59     pass
  60
  61   def test_init(self):
  62     archive = httparchive.HttpArchive()
  63     self.assertEqual(len(archive), 0)
  64
  65   def test__TrimHeaders(self):
  66     request = httparchive.ArchivedHttpRequest
  67     header1 = {'accept-encoding': 'gzip,deflate'}
  68     self.assertEqual(request._TrimHeaders(header1),
  69                      [(k, v) for k, v in header1.items()])
  70
  71     header2 = {'referer': 'www.google.com'}
  72     self.assertEqual(request._TrimHeaders(header2), [])
  73
  74     header3 = {'referer': 'www.google.com', 'cookie': 'cookie_monster!',
  75                'hello': 'world'}
  76     self.assertEqual(request._TrimHeaders(header3), [('hello', 'world')])
  77
  78     # Tests that spaces and trailing comma get stripped.
  79     header4 = {'accept-encoding': 'gzip, deflate,, '}
  80     self.assertEqual(request._TrimHeaders(header4),
  81                      [('accept-encoding', 'gzip,deflate')])
  82
  83   def test_matches(self):
  84     headers = {}
  85     request1 = httparchive.ArchivedHttpRequest(
  86         'GET', 'www.test.com', '/index.html?hello=world', None, headers)
  87     request2 = httparchive.ArchivedHttpRequest(
  88         'GET', 'www.test.com', '/index.html?foo=bar', None, headers)
  89
  90     self.assert_(not request1.matches(
  91         request2.command, request2.host, request2.full_path, use_query=True))
  92     self.assert_(request1.matches(
  93         request2.command, request2.host, request2.full_path, use_query=False))
  94
  95     self.assert_(request1.matches(
  96         request2.command, request2.host, None, use_query=True))
  97     self.assert_(request1.matches(
  98         request2.command, None, request2.full_path, use_query=False))
  99
 100     empty_request = httparchive.ArchivedHttpRequest(
 101         None, None, None, None, headers)
 102     self.assert_(not empty_request.matches(
 103         request2.command, request2.host, None, use_query=True))
 104     self.assert_(not empty_request.matches(
 105         request2.command, None, request2.full_path, use_query=False))
 106
 107   def setup_find_closest_request(self):
 108     headers = {}
 109     request1 = httparchive.ArchivedHttpRequest(
 110         'GET', 'www.test.com', '/a?hello=world', None, headers)
 111     request2 = httparchive.ArchivedHttpRequest(
 112         'GET', 'www.test.com', '/a?foo=bar', None, headers)
 113     request3 = httparchive.ArchivedHttpRequest(
 114         'GET', 'www.test.com', '/b?hello=world', None, headers)
 115
 116     archive = httparchive.HttpArchive()
 117     # Add requests 2 and 3 and find closest match with request1
 118     archive[request2] = self.RESPONSE
 119     archive[request3] = self.RESPONSE
 120
 121     return archive, request1, request2, request3
 122
 123   def test_find_closest_request(self):
 124     archive, request1, request2, request3 = self.setup_find_closest_request()
 125
 126     # Request 3 is the closest match to request 1
 127     self.assertEqual(
 128         request3, archive.find_closest_request(request1, use_path=False))
 129     # However, if we match strictly on path, request2 is the only match
 130     self.assertEqual(
 131         request2, archive.find_closest_request(request1, use_path=True))
 132
 133   def test_find_closest_request_delete_simple(self):
 134     archive, request1, request2, request3 = self.setup_find_closest_request()
 135
 136     del archive[request3]
 137     self.assertEqual(
 138         request2, archive.find_closest_request(request1, use_path=False))
 139     self.assertEqual(
 140         request2, archive.find_closest_request(request1, use_path=True))
 141
 142   def test_find_closest_request_delete_complex(self):
 143     archive, request1, request2, request3 = self.setup_find_closest_request()
 144
 145     del archive[request2]
 146     self.assertEqual(
 147         request3, archive.find_closest_request(request1, use_path=False))
 148     self.assertEqual(
 149         None, archive.find_closest_request(request1, use_path=True))
 150
 151   def test_get_simple(self):
 152     request = self.REQUEST
 153     response = self.RESPONSE
 154     archive = self.archive
 155
 156     self.assertEqual(archive.get(request), response)
 157
 158     false_request_headers = {'foo': 'bar'}
 159     false_request = create_request(false_request_headers)
 160     self.assertEqual(archive.get(false_request, default=None), None)
 161
 162   def test_get_modified_headers(self):
 163     request = self.REQUEST
 164     response = self.RESPONSE
 165     archive = self.archive
 166     not_modified_response = httparchive.create_response(304)
 167
 168     # Fail check and return response again
 169     request_headers = {'if-modified-since': self.DATE_PAST}
 170     request = create_request(request_headers)
 171     self.assertEqual(archive.get(request), response)
 172
 173     # Succeed check and return 304 Not Modified
 174     request_headers = {'if-modified-since': self.DATE_FUTURE}
 175     request = create_request(request_headers)
 176     self.assertEqual(archive.get(request), not_modified_response)
 177
 178     # Succeed check and return 304 Not Modified
 179     request_headers = {'if-modified-since': self.DATE_PRESENT}
 180     request = create_request(request_headers)
 181     self.assertEqual(archive.get(request), not_modified_response)
 182
 183     # Invalid date, fail check and return response again
 184     request_headers = {'if-modified-since': self.DATE_INVALID}
 185     request = create_request(request_headers)
 186     self.assertEqual(archive.get(request), response)
 187
 188     # fail check since the request is not a GET or HEAD request (as per RFC)
 189     request_headers = {'if-modified-since': self.DATE_FUTURE}
 190     request = httparchive.ArchivedHttpRequest(
 191         'POST', 'www.test.com', '/', None, request_headers)
 192     self.assertEqual(archive.get(request), response)
 193
 194   def test_get_unmodified_headers(self):
 195     request = self.REQUEST
 196     response = self.RESPONSE
 197     archive = self.archive
 198     not_modified_response = httparchive.create_response(304)
 199
 200     # Succeed check
 201     request_headers = {'if-unmodified-since': self.DATE_PAST}
 202     request = create_request(request_headers)
 203     self.assertEqual(archive.get(request), not_modified_response)
 204
 205     # Fail check
 206     request_headers = {'if-unmodified-since': self.DATE_FUTURE}
 207     request = create_request(request_headers)
 208     self.assertEqual(archive.get(request), response)
 209
 210     # Succeed check
 211     request_headers = {'if-unmodified-since': self.DATE_PRESENT}
 212     request = create_request(request_headers)
 213     self.assertEqual(archive.get(request), not_modified_response)
 214
 215     # Fail check
 216     request_headers = {'if-unmodified-since': self.DATE_INVALID}
 217     request = create_request(request_headers)
 218     self.assertEqual(archive.get(request), response)
 219
 220     # Fail check since the request is not a GET or HEAD request (as per RFC)
 221     request_headers = {'if-modified-since': self.DATE_PAST}
 222     request = httparchive.ArchivedHttpRequest(
 223         'POST', 'www.test.com', '/', None, request_headers)
 224     self.assertEqual(archive.get(request), response)
 225
 226   def test_get_etags(self):
 227     request = self.REQUEST
 228     response = self.RESPONSE
 229     archive = self.archive
 230     not_modified_response = httparchive.create_response(304)
 231     precondition_failed_response = httparchive.create_response(412)
 232
 233     # if-match headers
 234     request_headers = {'if-match': self.ETAG_VALID}
 235     request = create_request(request_headers)
 236     self.assertEqual(archive.get(request), response)
 237
 238     request_headers = {'if-match': self.ETAG_INVALID}
 239     request = create_request(request_headers)
 240     self.assertEqual(archive.get(request), precondition_failed_response)
 241
 242     # if-none-match headers
 243     request_headers = {'if-none-match': self.ETAG_VALID}
 244     request = create_request(request_headers)
 245     self.assertEqual(archive.get(request), not_modified_response)
 246
 247     request_headers = {'if-none-match': self.ETAG_INVALID}
 248     request = create_request(request_headers)
 249     self.assertEqual(archive.get(request), response)
 250
 251   def test_get_multiple_match_headers(self):
 252     request = self.REQUEST
 253     response = self.RESPONSE
 254     archive = self.archive
 255     not_modified_response = httparchive.create_response(304)
 256     precondition_failed_response = httparchive.create_response(412)
 257
 258     # if-match headers
 259     # If the request would, without the If-Match header field,
 260     # result in anything other than a 2xx or 412 status,
 261     # then the If-Match header MUST be ignored.
 262
 263     request_headers = {
 264         'if-match': self.ETAG_VALID,
 265         'if-modified-since': self.DATE_PAST,
 266     }
 267     request = create_request(request_headers)
 268     self.assertEqual(archive.get(request), response)
 269
 270     # Invalid etag, precondition failed
 271     request_headers = {
 272         'if-match': self.ETAG_INVALID,
 273         'if-modified-since': self.DATE_PAST,
 274     }
 275     request = create_request(request_headers)
 276     self.assertEqual(archive.get(request), precondition_failed_response)
 277
 278     # 304 response; ignore if-match header
 279     request_headers = {
 280         'if-match': self.ETAG_VALID,
 281         'if-modified-since': self.DATE_FUTURE,
 282     }
 283     request = create_request(request_headers)
 284     self.assertEqual(archive.get(request), not_modified_response)
 285
 286     # 304 response; ignore if-match header
 287     request_headers = {
 288         'if-match': self.ETAG_INVALID,
 289         'if-modified-since': self.DATE_PRESENT,
 290     }
 291     request = create_request(request_headers)
 292     self.assertEqual(archive.get(request), not_modified_response)
 293
 294     # Invalid etag, precondition failed
 295     request_headers = {
 296         'if-match': self.ETAG_INVALID,
 297         'if-modified-since': self.DATE_INVALID,
 298     }
 299     request = create_request(request_headers)
 300     self.assertEqual(archive.get(request), precondition_failed_response)
 301
 302   def test_get_multiple_none_match_headers(self):
 303     request = self.REQUEST
 304     response = self.RESPONSE
 305     archive = self.archive
 306     not_modified_response = httparchive.create_response(304)
 307     precondition_failed_response = httparchive.create_response(412)
 308
 309     # if-none-match headers
 310     # If the request would, without the If-None-Match header field,
 311     # result in anything other than a 2xx or 304 status,
 312     # then the If-None-Match header MUST be ignored.
 313
 314     request_headers = {
 315         'if-none-match': self.ETAG_VALID,
 316         'if-modified-since': self.DATE_PAST,
 317     }
 318     request = create_request(request_headers)
 319     self.assertEqual(archive.get(request), response)
 320
 321     request_headers = {
 322         'if-none-match': self.ETAG_INVALID,
 323         'if-modified-since': self.DATE_PAST,
 324     }
 325     request = create_request(request_headers)
 326     self.assertEqual(archive.get(request), response)
 327
 328     # etag match, precondition failed
 329     request_headers = {
 330         'if-none-match': self.ETAG_VALID,
 331         'if-modified-since': self.DATE_FUTURE,
 332     }
 333     request = create_request(request_headers)
 334     self.assertEqual(archive.get(request), not_modified_response)
 335
 336     request_headers = {
 337         'if-none-match': self.ETAG_INVALID,
 338         'if-modified-since': self.DATE_PRESENT,
 339     }
 340     request = create_request(request_headers)
 341     self.assertEqual(archive.get(request), not_modified_response)
 342
 343     request_headers = {
 344         'if-none-match': self.ETAG_INVALID,
 345         'if-modified-since': self.DATE_INVALID,
 346     }
 347     request = create_request(request_headers)
 348     self.assertEqual(archive.get(request), response)
 349
 350
 351 class ArchivedHttpResponse(unittest.TestCase):
 352   PAST_DATE_A = 'Tue, 13 Jul 2010 03:47:07 GMT'
 353   PAST_DATE_B = 'Tue, 13 Jul 2010 02:47:07 GMT'  # PAST_DATE_A -1 hour
 354   PAST_DATE_C = 'Tue, 13 Jul 2010 04:47:07 GMT'  # PAST_DATE_A +1 hour
 355   NOW_DATE_A = 'Wed, 20 Jul 2011 04:58:08 GMT'
 356   NOW_DATE_B = 'Wed, 20 Jul 2011 03:58:08 GMT'  # NOW_DATE_A -1 hour
 357   NOW_DATE_C = 'Wed, 20 Jul 2011 05:58:08 GMT'  # NOW_DATE_A +1 hour
 358   NOW_SECONDS = calendar.timegm(email.utils.parsedate(NOW_DATE_A))
 359
 360   def setUp(self):
 361     self.response = create_response([('date', self.PAST_DATE_A)])
 362
 363   def test_update_date_same_date(self):
 364     self.assertEqual(
 365         self.response.update_date(self.PAST_DATE_A, now=self.NOW_SECONDS),
 366         self.NOW_DATE_A)
 367
 368   def test_update_date_before_date(self):
 369     self.assertEqual(
 370         self.response.update_date(self.PAST_DATE_B, now=self.NOW_SECONDS),
 371         self.NOW_DATE_B)
 372
 373   def test_update_date_after_date(self):
 374     self.assertEqual(
 375         self.response.update_date(self.PAST_DATE_C, now=self.NOW_SECONDS),
 376         self.NOW_DATE_C)
 377
 378   def test_update_date_bad_date_param(self):
 379     self.assertEqual(
 380         self.response.update_date('garbage date', now=self.NOW_SECONDS),
 381         'garbage date')
 382
 383   def test_update_date_bad_date_header(self):
 384     self.response.set_header('date', 'garbage date')
 385     self.assertEqual(
 386         self.response.update_date(self.PAST_DATE_B, now=self.NOW_SECONDS),
 387         self.PAST_DATE_B)
 388
 389
 390 if __name__ == '__main__':
 391   unittest.main()