tools/httptester/server.py

   1 """Regression testing HTTP 'server'
   2
   3 The intent of this is to provide a (scriptable) framework for regression
   4 testing mozilla stuff. See the docs for details.
   5 """
   6
   7 __version__ = "0.1"
   8
   9 import BaseHTTPServer
  10 import string
  11 import time
  12 import os
  13 from stat import *
  14 import results
  15
  16 class RegressionHTTPRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
  17     server_version = "Regression/" + __version__
  18
  19     protocol_version = "HTTP/1.1"
  20
  21     def do_GET(self):
  22         if self.path == '/':
  23             return self.initial_redirect()
  24         if self.path[:1] != '/':
  25             return self.send_error(400,
  26                                    "Path %s does not begin with /" % `self.path`)
  27
  28         try:
  29             id, req = string.split(self.path[1:], '/', 1)
  30         except ValueError:
  31             return self.send_error(404, "Missing id and path")
  32
  33         if not req:
  34             # Initial request. Need to get a file list
  35             return self.list_tests(id)
  36         elif req == 'report':
  37             self.send_response(200)
  38             self.send_header('Content-Type', "text/plain")
  39             self.end_headers()
  40             res = results.results(id)
  41             res.write_report(self.wfile)
  42             del res
  43             return
  44         else:
  45             return self.handle_request(id, req)
  46
  47     def initial_redirect(self):
  48         """Redirect the initial query.
  49
  50         I don't want to use cookies, because that would bring
  51         wallet into all the tests. So the url will be:
  52         http://foo/123456/path/and/filename.html"""
  53
  54         self.send_response(302)
  55         self.send_header("Location","/%s/" % (long(time.time()*1000)))
  56         self.end_headers()
  57
  58     def list_tests(self, id):
  59         """List all test cases."""
  60
  61         try:
  62             os.stat("tests")
  63         except IOError:
  64             return self.send_error(500, "Tests were not found")
  65
  66         self.send_response(200)
  67         self.send_header('Content-Type', "text/plain")
  68         self.end_headers()
  69         return self.recurse_dir(id,"tests")
  70
  71     def recurse_dir(self, id, path):
  72         hasDir = 0
  73
  74         dir = os.listdir(path)
  75         dir.sort()
  76
  77         for i in dir:
  78             if i == 'CVS':
  79               continue
  80             mode = os.stat(path+'/'+i)[ST_MODE]
  81             if S_ISDIR(mode):
  82                 hasDir = 1
  83                 self.recurse_dir(id,path+"/"+i)
  84             elif hasDir:
  85                 print "%s: Warning! dir and non dir are mixed." % (path)
  86
  87         if not hasDir:
  88             self.wfile.write("http://localhost:8000/%s/%s/\n" % (id, path))
  89
  90     def copyfileobj(self, src, dst):
  91         """See shutil.copyfileobj from 2.x
  92
  93         I want this to be usable with 1.5 though"""
  94
  95         while 1:
  96             data = src.read(4096)
  97             if not data: break
  98             dst.write(data)
  99
 100     default_reply = "Testcase %s for %s loaded\n"
 101
 102     def handle_request(self, id, req):
 103         """Answer a request
 104
 105         We first look for a file with the name of the request.
 106         If that exists, then we spit that out, otherwise we
 107         open req.headers and req.body (if available) separately.
 108
 109         Why would you want to split it out?
 110         a) binary files
 111         b) Separating it out will send the 'standard' headers,
 112         and handle the Connection: details for you, if you're
 113         not testing that.
 114         c) You don't need to come up with your own body"""
 115
 116         res = results.results(id)
 117
 118         path = string.join(string.split(req, '/')[:-1], '/')
 119
 120         path = path + '/'
 121
 122         tester = res.get_tester(path)
 123
 124         self.fname = string.split(req,'/')[-1]
 125
 126         if not self.fname:
 127             self.fname = tester.baseName
 128
 129         if not tester.verify_request(self):
 130             res.set_tester(req, tester)
 131             return self.send_error(400, tester.reason)
 132
 133         ### perhaps this isn't the best design model...
 134         res.set_tester(path, tester)
 135
 136         del res
 137
 138         if req[-1:] == '/':
 139             req = req + tester.baseName
 140
 141         try:
 142             f = open(req, 'rb')
 143             self.log_message('"%s" sent successfully for %s',
 144                              self.requestline,
 145                              id)
 146             self.copyfileobj(f,self.wfile)
 147             return f.close()
 148         except IOError:
 149             try:
 150                 f = open(req+".headers", 'rb')
 151             except IOError:
 152                 return self.send_error(404, "File %s not found" % (req))
 153
 154         self.send_response(f.readline())
 155         # XXX - I should parse these out, and use send_header instead
 156         # so that I can change behaviour (like keep-alive...)
 157         # But then I couldn't test 'incorrect' header formats
 158         self.copyfileobj(f,self.wfile)
 159         f.close()
 160
 161         try:
 162             f = open(req+".body", 'rb')
 163             ## XXXXX - Need to configify this
 164             ## and then send content-length, etc
 165             self.end_headers()
 166             self.copyfileobj(f, self.wfile)
 167             return f.close()
 168         except IOError:
 169             self.send_header('Content-Type', "text/plain")
 170             body = self.default_reply % (req, id)
 171             self.send_header('Content-Length', len(body))
 172             self.end_headers()
 173             self.wfile.write(body)
 174
 175     def send_response(self, line, msg=None):
 176         if msg:
 177             return BaseHTTPServer.BaseHTTPRequestHandler.send_response(self, line,msg)
 178         try:
 179             x = int(line)
 180             BaseHTTPServer.BaseHTTPRequestHandler.send_response(self, x)
 181         except ValueError:
 182             tuple = string.split(line, ' ',2)
 183             ## XXXX
 184             old = self.protocol_version
 185             self.protocol_version = tuple[0]
 186             BaseHTTPServer.BaseHTTPRequestHandler.send_response(self, int(tuple[1]), tuple[2][:-1])
 187             self.protocol_version = old
 188
 189 import socket
 190
 191 # I need to thread this, with the mixin class
 192 class RegressionHTTPServer(BaseHTTPServer.HTTPServer):
 193     # The 1.5.2 version doesn't do this:
 194     def server_bind(self):
 195         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 196         BaseHTTPServer.HTTPServer.server_bind(self)
 197
 198 def run(HandlerClass = RegressionHTTPRequestHandler,
 199         ServerClass = RegressionHTTPServer):
 200     BaseHTTPServer.test(HandlerClass, ServerClass)
 201
 202 if __name__ == '__main__':
 203     run()