#!/usr/bin/env python # # Copyright 2007 Doug Hellmann. # # # All Rights Reserved # # Permission to use, copy, modify, and distribute this software and # its documentation for any purpose and without fee is hereby # granted, provided that the above copyright notice appear in all # copies and that both that copyright notice and this permission # notice appear in supporting documentation, and that the name of Doug # Hellmann not be used in advertising or publicity pertaining to # distribution of the software without specific, written prior # permission. # # DOUG HELLMANN DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, # INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN # NO EVENT SHALL DOUG HELLMANN BE LIABLE FOR ANY SPECIAL, INDIRECT OR # CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS # OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. # """Simple HTTP server for testing the feed cache. """ __module_id__ = "$Id$" # # Import system modules # import BaseHTTPServer import logging import md5 import threading import time import unittest import urllib # # Import local modules # # # Module # logger = logging.getLogger('feedcache.test_server') def make_etag(data): """Given a string containing data to be returned to the client, compute an ETag value for the data. """ _md5 = md5.new() _md5.update(data) return _md5.hexdigest() class TestHTTPHandler(BaseHTTPServer.BaseHTTPRequestHandler): "HTTP request handler which serves the same feed data every time." FEED_DATA = """ CacheTest test data http://localhost/feedcache/ 2006-10-14T11:00:36Z single test entry 2006-10-14T11:00:36Z author goes here authoremail@example.com http://www.example.com/ description goes here """ # The data does not change, so save the ETag and modified times # as class attributes. ETAG = make_etag(FEED_DATA) # Calculated using email.utils.formatdate(usegmt=True) MODIFIED_TIME = 'Sun, 08 Apr 2012 20:16:48 GMT' def do_GET(self): "Handle GET requests." logger.debug('GET %s', self.path) if self.path == '/shutdown': # Shortcut to handle stopping the server logger.debug('Stopping server') self.server.stop() self.send_response(200) else: # Record the request for tests that count them self.server.requests.append(self.path) # Process the request logger.debug('pre-defined response code: %d', self.server.response) handler_method_name = 'do_GET_%d' % self.server.response handler_method = getattr(self, handler_method_name) handler_method() return def do_GET_3xx(self): "Handle redirects" if self.path.endswith('/redirected'): logger.debug('already redirected') # We have already redirected, so return the data. return self.do_GET_200() new_path = self.server.new_path logger.debug('redirecting to %s', new_path) self.send_response(self.server.response) self.send_header('Location', new_path) return do_GET_301 = do_GET_3xx do_GET_302 = do_GET_3xx do_GET_303 = do_GET_3xx do_GET_307 = do_GET_3xx def do_GET_200(self): logger.debug('Etag: %s' % self.ETAG) logger.debug('Last-Modified: %s' % self.MODIFIED_TIME) incoming_etag = self.headers.get('If-None-Match', None) logger.debug('Incoming ETag: "%s"' % incoming_etag) incoming_modified = self.headers.get('If-Modified-Since', None) logger.debug('Incoming If-Modified-Since: %s' % incoming_modified) send_data = True # Does the client have the same version of the data we have? if self.server.apply_modified_headers: if incoming_etag == self.ETAG: logger.debug('Response 304, etag') self.send_response(304) send_data = False elif incoming_modified == self.MODIFIED_TIME: logger.debug('Response 304, modified time') self.send_response(304) send_data = False # Now optionally send the data, if the client needs it if send_data: logger.debug('Response 200') self.send_response(200) self.send_header('Content-Type', 'application/atom+xml') logger.debug('Outgoing Etag: %s' % self.ETAG) self.send_header('ETag', self.ETAG) logger.debug('Outgoing modified time: %s' % self.MODIFIED_TIME) self.send_header('Last-Modified', self.MODIFIED_TIME) self.end_headers() logger.debug('Sending data') self.wfile.write(self.FEED_DATA) return class TestHTTPServer(BaseHTTPServer.HTTPServer): """HTTP Server which counts the number of requests made and can stop based on client instructions. """ def __init__(self, applyModifiedHeaders=True, handler=TestHTTPHandler): self.apply_modified_headers = applyModifiedHeaders self.keep_serving = True self.requests = [] self.setResponse(200) BaseHTTPServer.HTTPServer.__init__(self, ('', 9999), handler) return def setResponse(self, newResponse, newPath=None): """Sets the response code to use for future requests, and a new path to be used as a redirect target, if necessary. """ self.response = newResponse self.new_path = newPath return def getNumRequests(self): "Return the number of requests which have been made on the server." return len(self.requests) def stop(self): "Stop serving requests, after the next request." self.keep_serving = False return def serve_forever(self): "Main loop for server" while self.keep_serving: self.handle_request() logger.debug('exiting') return class HTTPTestBase(unittest.TestCase): "Base class for tests that use a TestHTTPServer" TEST_URL = 'http://localhost:9999/' CACHE_TTL = 0 def setUp(self): self.server = self.getServer() self.server_thread = threading.Thread(target=self.server.serve_forever) # set daemon flag so the tests don't hang if cleanup fails self.server_thread.setDaemon(True) self.server_thread.start() return def getServer(self): "Return a web server for the test." s = TestHTTPServer() s.setResponse(200) return s def tearDown(self): # Stop the server thread urllib.urlretrieve(self.TEST_URL + 'shutdown') time.sleep(1) self.server.server_close() self.server_thread.join() return