httparchive_test.py

#!/usr/bin/env python
# Copyright 2011 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import calendar
import email.utils
import httparchive
import unittest


def create_request(headers):
  return httparchive.ArchivedHttpRequest(
      'GET', 'www.test.com', '/', None, headers)

def create_response(headers):
  return httparchive.ArchivedHttpResponse(
      11, 200, 'OK', headers, '')


class HttpArchiveTest(unittest.TestCase):

  REQUEST_HEADERS = {}
  REQUEST = create_request(REQUEST_HEADERS)

  # Used for if-(un)modified-since checks
  DATE_PAST = 'Wed, 13 Jul 2011 03:58:08 GMT'
  DATE_PRESENT = 'Wed, 20 Jul 2011 04:58:08 GMT'
  DATE_FUTURE = 'Wed, 27 Jul 2011 05:58:08 GMT'
  DATE_INVALID = 'This is an invalid date!!'

  # etag values
  ETAG_VALID = 'etag'
  ETAG_INVALID = 'This is an invalid etag value!!'

  RESPONSE_HEADERS = [('last-modified', DATE_PRESENT), ('etag', ETAG_VALID)]
  RESPONSE = create_response(RESPONSE_HEADERS)

  def setUp(self):
    self.archive = httparchive.HttpArchive()
    self.archive[self.REQUEST] = self.RESPONSE

    # Also add an identical POST request for testing
    request = httparchive.ArchivedHttpRequest(
        'POST', 'www.test.com', '/', None, self.REQUEST_HEADERS)
    self.archive[request] = self.RESPONSE

  def tearDown(self):
    pass

  def test_init(self):
    archive = httparchive.HttpArchive()
    self.assertEqual(len(archive), 0)

  def test_request__TrimHeaders(self):
    request = httparchive.ArchivedHttpRequest
    header1 = {'accept-encoding': 'gzip,deflate'}
    self.assertEqual(request._TrimHeaders(header1),
                     [(k, v) for k, v in header1.items()])

    header2 = {'referer': 'www.google.com'}
    self.assertEqual(request._TrimHeaders(header2), [])

    header3 = {'referer': 'www.google.com', 'cookie': 'cookie_monster!',
               'hello': 'world'}
    self.assertEqual(request._TrimHeaders(header3), [('hello', 'world')])

    # Tests that spaces and trailing comma get stripped.
    header4 = {'accept-encoding': 'gzip, deflate,, '}
    self.assertEqual(request._TrimHeaders(header4),
                     [('accept-encoding', 'gzip,deflate')])

    # Tests that 'lzma' gets stripped.
    header5 = {'accept-encoding': 'gzip, deflate, lzma'}
    self.assertEqual(request._TrimHeaders(header5),
                     [('accept-encoding', 'gzip,deflate')])

    # Tests that x-client-data gets stripped.
    header6 = {'x-client-data': 'testdata'}
    self.assertEqual(request._TrimHeaders(header6), [])

  def test_matches(self):
    headers = {}
    request1 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/index.html?hello=world', None, headers)
    request2 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/index.html?foo=bar', None, headers)

    self.assert_(not request1.matches(
        request2.command, request2.host, request2.full_path, use_query=True))
    self.assert_(request1.matches(
        request2.command, request2.host, request2.full_path, use_query=False))

    self.assert_(request1.matches(
        request2.command, request2.host, None, use_query=True))
    self.assert_(request1.matches(
        request2.command, None, request2.full_path, use_query=False))

    empty_request = httparchive.ArchivedHttpRequest(
        None, None, None, None, headers)
    self.assert_(not empty_request.matches(
        request2.command, request2.host, None, use_query=True))
    self.assert_(not empty_request.matches(
        request2.command, None, request2.full_path, use_query=False))

  def setup_find_closest_request(self):
    headers = {}
    request1 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/a?hello=world', None, headers)
    request2 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/a?foo=bar', None, headers)
    request3 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/b?hello=world', None, headers)
    request4 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/c?hello=world', None, headers)

    archive = httparchive.HttpArchive()
    # Add requests 2 and 3 and find closest match with request1
    archive[request2] = self.RESPONSE
    archive[request3] = self.RESPONSE

    return archive, request1, request2, request3, request4

  def test_find_closest_request(self):
    archive, request1, request2, request3, request4 = (
      self.setup_find_closest_request())

    # Always favor requests with same paths, even if use_path=False.
    self.assertEqual(
        request2, archive.find_closest_request(request1, use_path=False))
    # If we match strictly on path, request2 is the only match
    self.assertEqual(
        request2, archive.find_closest_request(request1, use_path=True))
    # request4 can be matched with request3, if use_path=False
    self.assertEqual(
        request3, archive.find_closest_request(request4, use_path=False))
    # ...but None, if use_path=True
    self.assertEqual(
        None, archive.find_closest_request(request4, use_path=True))

  def test_find_closest_request_delete_simple(self):
    archive, request1, request2, request3, request4 = (
      self.setup_find_closest_request())

    del archive[request3]
    self.assertEqual(
        request2, archive.find_closest_request(request1, use_path=False))
    self.assertEqual(
        request2, archive.find_closest_request(request1, use_path=True))

  def test_find_closest_request_delete_complex(self):
    archive, request1, request2, request3, request4 = (
      self.setup_find_closest_request())

    del archive[request2]
    self.assertEqual(
        request3, archive.find_closest_request(request1, use_path=False))
    self.assertEqual(
        None, archive.find_closest_request(request1, use_path=True))

  def test_find_closest_request_timestamp(self):
    headers = {}
    request1 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/index.html?time=100000000&important=true',
        None, headers)
    request2 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/index.html?time=99999999&important=true',
        None, headers)
    request3 = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/index.html?time=10000000&important=false',
        None, headers)
    archive = httparchive.HttpArchive()
    # Add requests 2 and 3 and find closest match with request1
    archive[request2] = self.RESPONSE
    archive[request3] = self.RESPONSE

    # Although request3 is lexicographically closer, request2 is semantically
    # more similar.
    self.assertEqual(
        request2, archive.find_closest_request(request1, use_path=True))

  def test_get_cmp_seq(self):
    # The order of key-value pairs in query and header respectively should not
    # matter.
    headers = {'k2': 'v2', 'k1': 'v1'}
    request = httparchive.ArchivedHttpRequest(
        'GET', 'www.test.com', '/a?c=d&a=b;e=f', None, headers)
    self.assertEqual([('a', 'b'), ('c', 'd'), ('e', 'f'),
                      ('k1', 'v1'), ('k2', 'v2')],
                     request._GetCmpSeq('c=d&a=b;e=f'))

  def test_get_simple(self):
    request = self.REQUEST
    response = self.RESPONSE
    archive = self.archive

    self.assertEqual(archive.get(request), response)

    false_request_headers = {'foo': 'bar'}
    false_request = create_request(false_request_headers)
    self.assertEqual(archive.get(false_request, default=None), None)

  def test_get_modified_headers(self):
    request = self.REQUEST
    response = self.RESPONSE
    archive = self.archive
    not_modified_response = httparchive.create_response(304)

    # Fail check and return response again
    request_headers = {'if-modified-since': self.DATE_PAST}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    # Succeed check and return 304 Not Modified
    request_headers = {'if-modified-since': self.DATE_FUTURE}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    # Succeed check and return 304 Not Modified
    request_headers = {'if-modified-since': self.DATE_PRESENT}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    # Invalid date, fail check and return response again
    request_headers = {'if-modified-since': self.DATE_INVALID}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    # fail check since the request is not a GET or HEAD request (as per RFC)
    request_headers = {'if-modified-since': self.DATE_FUTURE}
    request = httparchive.ArchivedHttpRequest(
        'POST', 'www.test.com', '/', None, request_headers)
    self.assertEqual(archive.get(request), response)

  def test_get_unmodified_headers(self):
    request = self.REQUEST
    response = self.RESPONSE
    archive = self.archive
    not_modified_response = httparchive.create_response(304)

    # Succeed check
    request_headers = {'if-unmodified-since': self.DATE_PAST}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    # Fail check
    request_headers = {'if-unmodified-since': self.DATE_FUTURE}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    # Succeed check
    request_headers = {'if-unmodified-since': self.DATE_PRESENT}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    # Fail check
    request_headers = {'if-unmodified-since': self.DATE_INVALID}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    # Fail check since the request is not a GET or HEAD request (as per RFC)
    request_headers = {'if-modified-since': self.DATE_PAST}
    request = httparchive.ArchivedHttpRequest(
        'POST', 'www.test.com', '/', None, request_headers)
    self.assertEqual(archive.get(request), response)

  def test_get_etags(self):
    request = self.REQUEST
    response = self.RESPONSE
    archive = self.archive
    not_modified_response = httparchive.create_response(304)
    precondition_failed_response = httparchive.create_response(412)

    # if-match headers
    request_headers = {'if-match': self.ETAG_VALID}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    request_headers = {'if-match': self.ETAG_INVALID}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), precondition_failed_response)

    # if-none-match headers
    request_headers = {'if-none-match': self.ETAG_VALID}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    request_headers = {'if-none-match': self.ETAG_INVALID}
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

  def test_get_multiple_match_headers(self):
    request = self.REQUEST
    response = self.RESPONSE
    archive = self.archive
    not_modified_response = httparchive.create_response(304)
    precondition_failed_response = httparchive.create_response(412)

    # if-match headers
    # If the request would, without the If-Match header field,
    # result in anything other than a 2xx or 412 status,
    # then the If-Match header MUST be ignored.

    request_headers = {
        'if-match': self.ETAG_VALID,
        'if-modified-since': self.DATE_PAST,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    # Invalid etag, precondition failed
    request_headers = {
        'if-match': self.ETAG_INVALID,
        'if-modified-since': self.DATE_PAST,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), precondition_failed_response)

    # 304 response; ignore if-match header
    request_headers = {
        'if-match': self.ETAG_VALID,
        'if-modified-since': self.DATE_FUTURE,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    # 304 response; ignore if-match header
    request_headers = {
        'if-match': self.ETAG_INVALID,
        'if-modified-since': self.DATE_PRESENT,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    # Invalid etag, precondition failed
    request_headers = {
        'if-match': self.ETAG_INVALID,
        'if-modified-since': self.DATE_INVALID,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), precondition_failed_response)

  def test_get_multiple_none_match_headers(self):
    request = self.REQUEST
    response = self.RESPONSE
    archive = self.archive
    not_modified_response = httparchive.create_response(304)
    precondition_failed_response = httparchive.create_response(412)

    # if-none-match headers
    # If the request would, without the If-None-Match header field,
    # result in anything other than a 2xx or 304 status,
    # then the If-None-Match header MUST be ignored.

    request_headers = {
        'if-none-match': self.ETAG_VALID,
        'if-modified-since': self.DATE_PAST,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    request_headers = {
        'if-none-match': self.ETAG_INVALID,
        'if-modified-since': self.DATE_PAST,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

    # etag match, precondition failed
    request_headers = {
        'if-none-match': self.ETAG_VALID,
        'if-modified-since': self.DATE_FUTURE,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    request_headers = {
        'if-none-match': self.ETAG_INVALID,
        'if-modified-since': self.DATE_PRESENT,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), not_modified_response)

    request_headers = {
        'if-none-match': self.ETAG_INVALID,
        'if-modified-since': self.DATE_INVALID,
    }
    request = create_request(request_headers)
    self.assertEqual(archive.get(request), response)

  def test_response__TrimHeaders(self):
    response = httparchive.ArchivedHttpResponse
    header1 = [('access-control-allow-origin', '*'),
               ('content-type', 'image/jpeg'),
               ('content-length', 2878)]
    self.assertEqual(response._TrimHeaders(header1), header1)

    header2 = [('content-type', 'text/javascript; charset=utf-8'),
               ('connection', 'keep-alive'),
               ('cache-control', 'private, must-revalidate, max-age=0'),
               ('content-encoding', 'gzip')]
    self.assertEqual(response._TrimHeaders(header2), header2)

    header3 = [('content-security-policy', """\
default-src 'self' http://*.cnn.com:* https://*.cnn.com:* \
*.cnn.net:* *.turner.com:* *.ugdturner.com:* *.vgtf.net:*; \
script-src 'unsafe-inline' 'unsafe-eval' 'self' *; \
style-src 'unsafe-inline' 'self' *; frame-src 'self' *; \
object-src 'self' *; img-src 'self' * data:; media-src 'self' *; \
font-src 'self' *; connect-src 'self' *"""),
               ('access-control-allow-origin', '*'),
               ('content-type', 'text/html; charset=utf-8'),
               ('content-encoding', 'gzip')]
    self.assertEqual(response._TrimHeaders(header3), [
        ('access-control-allow-origin', '*'),
        ('content-type', 'text/html; charset=utf-8'),
        ('content-encoding', 'gzip')
    ])

    header4 = [('content-security-policy', """\
default-src * data: blob:;script-src *.facebook.com *.fbcdn.net \
*.facebook.net *.google-analytics.com *.virtualearth.net *.google.com \
127.0.0.1:* *.spotilocal.com:* 'unsafe-inline' 'unsafe-eval' \
fbstatic-a.akamaihd.net fbcdn-static-b-a.akamaihd.net *.atlassolutions.com \
blob: chrome-extension://lifbcibllhkdhoafpjfnlhfpfgnpldfl \
*.liverail.com;style-src * 'unsafe-inline' data:;connect-src *.facebook.com \
*.fbcdn.net *.facebook.net *.spotilocal.com:* *.akamaihd.net \
wss://*.facebook.com:* https://fb.scanandcleanlocal.com:* \
*.atlassolutions.com attachment.fbsbx.com ws://localhost:* \
blob: 127.0.0.1:* *.liverail.com""")]
    self.assertEqual(response._TrimHeaders(header4), [])


class ArchivedHttpResponse(unittest.TestCase):
  PAST_DATE_A = 'Tue, 13 Jul 2010 03:47:07 GMT'
  PAST_DATE_B = 'Tue, 13 Jul 2010 02:47:07 GMT'  # PAST_DATE_A -1 hour
  PAST_DATE_C = 'Tue, 13 Jul 2010 04:47:07 GMT'  # PAST_DATE_A +1 hour
  NOW_DATE_A = 'Wed, 20 Jul 2011 04:58:08 GMT'
  NOW_DATE_B = 'Wed, 20 Jul 2011 03:58:08 GMT'  # NOW_DATE_A -1 hour
  NOW_DATE_C = 'Wed, 20 Jul 2011 05:58:08 GMT'  # NOW_DATE_A +1 hour
  NOW_SECONDS = calendar.timegm(email.utils.parsedate(NOW_DATE_A))

  def setUp(self):
    self.response = create_response([('date', self.PAST_DATE_A)])

  def test_update_date_same_date(self):
    self.assertEqual(
        self.response.update_date(self.PAST_DATE_A, now=self.NOW_SECONDS),
        self.NOW_DATE_A)

  def test_update_date_before_date(self):
    self.assertEqual(
        self.response.update_date(self.PAST_DATE_B, now=self.NOW_SECONDS),
        self.NOW_DATE_B)

  def test_update_date_after_date(self):
    self.assertEqual(
        self.response.update_date(self.PAST_DATE_C, now=self.NOW_SECONDS),
        self.NOW_DATE_C)

  def test_update_date_bad_date_param(self):
    self.assertEqual(
        self.response.update_date('garbage date', now=self.NOW_SECONDS),
        'garbage date')

  def test_update_date_bad_date_header(self):
    self.response.set_header('date', 'garbage date')
    self.assertEqual(
        self.response.update_date(self.PAST_DATE_B, now=self.NOW_SECONDS),
        self.PAST_DATE_B)


if __name__ == '__main__':
  unittest.main()