michael@0: # This Source Code Form is subject to the terms of the Mozilla Public michael@0: # License, v. 2.0. If a copy of the MPL was not distributed with this michael@0: # file, You can obtain one at http://mozilla.org/MPL/2.0/. michael@0: michael@0: """Regression testing HTTP 'server' michael@0: michael@0: The intent of this is to provide a (scriptable) framework for regression michael@0: testing mozilla stuff. See the docs for details. michael@0: """ michael@0: michael@0: __version__ = "0.1" michael@0: michael@0: import BaseHTTPServer michael@0: import string michael@0: import time michael@0: import os michael@0: from stat import * michael@0: import results michael@0: michael@0: class RegressionHTTPRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): michael@0: server_version = "Regression/" + __version__ michael@0: michael@0: protocol_version = "HTTP/1.1" michael@0: michael@0: def do_GET(self): michael@0: if self.path == '/': michael@0: return self.initial_redirect() michael@0: if self.path[:1] != '/': michael@0: return self.send_error(400, michael@0: "Path %s does not begin with /" % `self.path`) michael@0: michael@0: try: michael@0: id, req = string.split(self.path[1:], '/', 1) michael@0: except ValueError: michael@0: return self.send_error(404, "Missing id and path") michael@0: michael@0: if not req: michael@0: # Initial request. Need to get a file list michael@0: return self.list_tests(id) michael@0: elif req == 'report': michael@0: self.send_response(200) michael@0: self.send_header('Content-Type', "text/plain") michael@0: self.end_headers() michael@0: res = results.results(id) michael@0: res.write_report(self.wfile) michael@0: del res michael@0: return michael@0: else: michael@0: return self.handle_request(id, req) michael@0: michael@0: def initial_redirect(self): michael@0: """Redirect the initial query. michael@0: michael@0: I don't want to use cookies, because that would bring michael@0: wallet into all the tests. So the url will be: michael@0: http://foo/123456/path/and/filename.html""" michael@0: michael@0: self.send_response(302) michael@0: self.send_header("Location","/%s/" % (long(time.time()*1000))) michael@0: self.end_headers() michael@0: michael@0: def list_tests(self, id): michael@0: """List all test cases.""" michael@0: michael@0: try: michael@0: os.stat("tests") michael@0: except IOError: michael@0: return self.send_error(500, "Tests were not found") michael@0: michael@0: self.send_response(200) michael@0: self.send_header('Content-Type', "text/plain") michael@0: self.end_headers() michael@0: return self.recurse_dir(id,"tests") michael@0: michael@0: def recurse_dir(self, id, path): michael@0: hasDir = 0 michael@0: michael@0: dir = os.listdir(path) michael@0: dir.sort() michael@0: michael@0: for i in dir: michael@0: if i == 'CVS': michael@0: continue michael@0: mode = os.stat(path+'/'+i)[ST_MODE] michael@0: if S_ISDIR(mode): michael@0: hasDir = 1 michael@0: self.recurse_dir(id,path+"/"+i) michael@0: elif hasDir: michael@0: print "%s: Warning! dir and non dir are mixed." % (path) michael@0: michael@0: if not hasDir: michael@0: self.wfile.write("http://localhost:8000/%s/%s/\n" % (id, path)) michael@0: michael@0: def copyfileobj(self, src, dst): michael@0: """See shutil.copyfileobj from 2.x michael@0: michael@0: I want this to be usable with 1.5 though""" michael@0: michael@0: while 1: michael@0: data = src.read(4096) michael@0: if not data: break michael@0: dst.write(data) michael@0: michael@0: default_reply = "Testcase %s for %s loaded\n" michael@0: michael@0: def handle_request(self, id, req): michael@0: """Answer a request michael@0: michael@0: We first look for a file with the name of the request. michael@0: If that exists, then we spit that out, otherwise we michael@0: open req.headers and req.body (if available) separately. michael@0: michael@0: Why would you want to split it out? michael@0: a) binary files michael@0: b) Separating it out will send the 'standard' headers, michael@0: and handle the Connection: details for you, if you're michael@0: not testing that. michael@0: c) You don't need to come up with your own body""" michael@0: michael@0: res = results.results(id) michael@0: michael@0: path = string.join(string.split(req, '/')[:-1], '/') michael@0: michael@0: path = path + '/' michael@0: michael@0: tester = res.get_tester(path) michael@0: michael@0: self.fname = string.split(req,'/')[-1] michael@0: michael@0: if not self.fname: michael@0: self.fname = tester.baseName michael@0: michael@0: if not tester.verify_request(self): michael@0: res.set_tester(req, tester) michael@0: return self.send_error(400, tester.reason) michael@0: michael@0: ### perhaps this isn't the best design model... michael@0: res.set_tester(path, tester) michael@0: michael@0: del res michael@0: michael@0: if req[-1:] == '/': michael@0: req = req + tester.baseName michael@0: michael@0: try: michael@0: f = open(req, 'rb') michael@0: self.log_message('"%s" sent successfully for %s', michael@0: self.requestline, michael@0: id) michael@0: self.copyfileobj(f,self.wfile) michael@0: return f.close() michael@0: except IOError: michael@0: try: michael@0: f = open(req+".headers", 'rb') michael@0: except IOError: michael@0: return self.send_error(404, "File %s not found" % (req)) michael@0: michael@0: self.send_response(f.readline()) michael@0: # XXX - I should parse these out, and use send_header instead michael@0: # so that I can change behaviour (like keep-alive...) michael@0: # But then I couldn't test 'incorrect' header formats michael@0: self.copyfileobj(f,self.wfile) michael@0: f.close() michael@0: michael@0: try: michael@0: f = open(req+".body", 'rb') michael@0: ## XXXXX - Need to configify this michael@0: ## and then send content-length, etc michael@0: self.end_headers() michael@0: self.copyfileobj(f, self.wfile) michael@0: return f.close() michael@0: except IOError: michael@0: self.send_header('Content-Type', "text/plain") michael@0: body = self.default_reply % (req, id) michael@0: self.send_header('Content-Length', len(body)) michael@0: self.end_headers() michael@0: self.wfile.write(body) michael@0: michael@0: def send_response(self, line, msg=None): michael@0: if msg: michael@0: return BaseHTTPServer.BaseHTTPRequestHandler.send_response(self, line,msg) michael@0: try: michael@0: x = int(line) michael@0: BaseHTTPServer.BaseHTTPRequestHandler.send_response(self, x) michael@0: except ValueError: michael@0: tuple = string.split(line, ' ',2) michael@0: ## XXXX michael@0: old = self.protocol_version michael@0: self.protocol_version = tuple[0] michael@0: BaseHTTPServer.BaseHTTPRequestHandler.send_response(self, int(tuple[1]), tuple[2][:-1]) michael@0: self.protocol_version = old michael@0: michael@0: import socket michael@0: michael@0: # I need to thread this, with the mixin class michael@0: class RegressionHTTPServer(BaseHTTPServer.HTTPServer): michael@0: # The 1.5.2 version doesn't do this: michael@0: def server_bind(self): michael@0: self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) michael@0: BaseHTTPServer.HTTPServer.server_bind(self) michael@0: michael@0: def run(HandlerClass = RegressionHTTPRequestHandler, michael@0: ServerClass = RegressionHTTPServer): michael@0: BaseHTTPServer.test(HandlerClass, ServerClass) michael@0: michael@0: if __name__ == '__main__': michael@0: run()