mock server —— 用于自动化测试

我们在测试某个模块的时候，该模块会依赖很多外部http server，真实搭建这样的server有两个缺点：1、部署维护成本较大，太重不适合自动化测试，2、网络异常、超时、返回部分结果等特殊情况很难测试到，这时候如果用到mock就很方便了，这里来介绍一个简单的mock server。

功能描述

模拟http服务, 同时支持get及post访问方式

提供两种set方法供用户自定义http服务response_code及response_content

接口说明

set_code_handler_func(self, func) #设置code处理方法，func接受参数为path，返回状态码，如：path:/index.jsp?galaxy=test return:200

set_content_handler_func(self, func) #设置content处理方法，func接受参数为path,返回page content，如：path:/index.html return:"hello world!"

start(self) # 启动服务

stop(self) # 停止服务

测试样例

from galaxy.mock_server import http

## init
server = http.MockHTTPServer(port=8000) # 默认使用8000端口

## set handler func
'''
模拟http服务response_code及response_content，对应提供如下两种set方法，eg：
func_code = lambda path: 504
func_content = lambda path: "hello world = " + path

以上func接受参数为path，当以post方式请求时path=path + “?” + data
'''
server.set_code_handler_func(func_code)
server.set_content_handler_func(func_content)

## start server
server.start()

## request code eg:
url = "http://10.250.11.57:8000/"
query = "url=http://www.taobao.com"
try:
req = urllib2.Request(url + urllib.quote(query))
response = urllib2.urlopen(req)
print response.read()
response.close()
except Exception, err:
print str(err)

## stop server
server.stop()

mock_server_http.py

"""
Mock HTTP Server.

This module builds on BaseHTTPServer by implementing the standard GET and POST
requests in a fairly straightforward manner.
"""

__version__ = "1.0"

__all__ = ["MockHTTPRequestHandler", "MockHTTPServer"]

import os
import cgi
import time
import BaseHTTPServer
import SocketServer
import urllib
import shutil
try:
from cStringIO import StringIO
except ImportError:
from StringIO import StringIO

import sys
sys.path.append("/usr/local/atest/atest-1.1.5/lib/")
import atest.log as log

class MockHTTPRequestHandler(SocketServer.ThreadingMixIn, BaseHTTPServer.BaseHTTPRequestHandler):

server_version = "MockHTTP/" + __version__

code_handler_func = None
content_handler_func = None
read_timeout = 0

def log_message(self, format, *args):

"""override for ignore any log."""

pass

def do_GET(self):

path = urllib.unquote(self.path)

log.debug("path: %s" % path)

self.__response(path)

def do_POST(self):

# Parse the form data posted

form = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
environ={'REQUEST_METHOD':'POST',
'CONTENT_TYPE':self.headers['Content-Type'],
})

data = [field + "=" + form[field].value for field in form.keys()]
path = "%s?%s" % (self.path, "&".join(data))

log.debug("path: %s" % path)

self.__response(path)

def __response(self, path):

response_code = self.code_handler_func.im_func(path) if self.code_handler_func else 200
response_content = self.content_handler_func.im_func(path) if self.content_handler_func else ""

log.debug("code: %s, content: %s" % (response_code, response_content))

f = StringIO()
f.write(response_content)
length = f.tell()
f.seek(0)
self.send_response(response_code)
self.send_header("Content-type", "text/html")
self.send_header("Content-Length", str(length))
self.end_headers()
if f:
sleep_time = float(self.read_timeout)/1000
log.debug("Sleep %s seconds" % sleep_time)
time.sleep(sleep_time)
shutil.copyfileobj(f, self.wfile)
f.close()

class MockHTTPServer:

def __init__(self, port=8000):

self._port = port
self._code_handler_func = None
self._content_handler_func = None
self._read_timeout = 0

self._pid = 0

def set_code_handler_func(self, func):

"""Code handler func, eg:
>>>def func(path): # /index.jsp?galaxy=test
...    return 200
@param func: lambda x: 200
"""

self._code_handler_func = func

def set_content_handler_func(self, func):

"""The same as set_code_handler_func"""

self._content_handler_func = func

def set_read_timeout(self, milliseconds):

"""Set read timeout"""

self._read_timeout = milliseconds

def _server(self, protocol="HTTP/1.0",
HandlerClass = MockHTTPRequestHandler,
ServerClass = BaseHTTPServer.HTTPServer):

server_address = ('', self._port)

HandlerClass.protocol_version = protocol
HandlerClass.code_handler_func = self._code_handler_func
HandlerClass.content_handler_func = self._content_handler_func
HandlerClass.read_timeout = self._read_timeout
HandlerClass.daemon_threads = True

httpd = ServerClass(server_address, HandlerClass)

sa = httpd.socket.getsockname()
log.info("Serving HTTP on %s port %s ..." % sa)

httpd.serve_forever()

def start(self):

log.info("Start mock http server ...")

child = os.fork()
if not child:
self._server()

self._pid = child

def stop(self):

log.info("Stop mock http server ...")

if self._pid:
os.kill(self._pid, 9)