tools/httptester/server.py

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