Mock挡板测试

工具一：python3 mock http请求

1 执行:nohup python httpserver.py > test.log  2>&1 &

2 mock脚本

httpserver.py

import sys
import importlib

importlib.reload(sys)
#sys.setdefaultencoding("utf-8")

import socketserver
import logging
import os
import platform
import time
# from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
from http.server import BaseHTTPRequestHandler, HTTPServer

import mock_utils

logger=logging.getLogger("MockServer")
logger.setLevel(level=logging.INFO)

# 日志
sysstr=platform.system()
BASE_DIR=os.path.dirname(os.path.join(os.path.dirname(os.path.dirname(__file__)), os.path.pardir))

if sysstr == "Windows":
    logFile="%slogsmockserver.log" % BASE_DIR
elif sysstr == "Linux":
    logFile="/home/iotest/smartpotatologs/mockserver.log"

# 建立一个filehandler来把日志记录在文件里，级别为debug以上
handler=logging.FileHandler(logFile)

handler.setLevel(logging.INFO)
formatter=logging.Formatter(
    '%(asctime)s [%(threadName)s:%(thread)d] [%(name)s:%(lineno)d] [%(module)s:%(funcName)s] [%(levelname)s]- %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

addr=('0.0.0.0', 8888)


class RequestHandler(BaseHTTPRequestHandler):
    def _getmockdata(self):
        self.mock_method=None
        # 1、获取mockurl，查询mock信息
        logger.info(str(self.client_address) + " " + str(self.command) + ' >>> 收到Mock请求,访问地址:' + str(self.path))
        mockurl=self.path[1:33]
        print(mockurl)

        try:
            db=mock_utils.create_engine('172.20.201.98', 3306, 'root', 'as123456', 'test')
            sql='''SELECT mock_code,mock_header,mock_data,time_out,request_type FROM smartpotato_mock_config WHERE status=1 AND mock_url= "%s"''' % mockurl;

            mockconfig=mock_utils.select_sql(db, sql)
            logger.info("查询mockconfig：" + str(mockconfig))
            return mockconfig
        except Exception as e:
            logging.error("查询mockconfig失败：" + str(e))
            return None

    def _doresponse(self, mockconfig):
        data_type=self.headers['Content-type']  # python3
        print(data_type)
        if mockconfig:
            try:
                time.sleep(mockconfig[3])
                self.send_response(int(mockconfig[0]))
                if mockconfig[1]:
                    hd=eval(mock_utils.pre_deal_testdata(mockconfig[1])) # eval() 函数用来执行一个字符串表达式，并返回表达式的值。
                    for h in hd:
                        self.send_header(h, hd[h])
                self.send_header('Content-type', data_type)
                self.end_headers()
                logger.info("设置Mock返回。")
                try:
                    res_data=mock_utils.pre_deal_testdata(mockconfig[2])
                except Exception as e:
                    print(e)
                self.wfile.write(res_data.encode('utf-8'))

            except Exception as e:
                logger.error("设置mock返回数据失败：" + str(e))
                self.send_response(500)
                self.send_header('Content-type', data_type)
                self.end_headers()
                self.wfile.write(mock_utils.result_error(data_type, {'result': '00,设置mock返回数据失败。'}).encode('utf-8'))
        else:
            logger.error("查询无mockconfig，或状态不可用：" + str(mockurl))
            self.send_response(404)
            self.send_header('Content-type', data_type)
            self.end_headers()
            self.wfile.write(mock_utils.result_error(data_type, {'result': '01,查询无mockconfig，或状态不可用。'}).encode('utf-8'))

    def do_Head(self):
        self._doresponse(self._getmockdata())

    def do_GET(self):
        mockconfig=self._getmockdata()
        if mockconfig[4] == 'GET':
            self._doresponse(mockconfig)
        else:
            self.send_error(404)

    def do_POST(self):

        mockconfig=self._getmockdata()
        #mockconfig ="("returnCode": "1","returnMsg": "查询成功","details": {"dataFileInfo": {"dataFileUrl": "http://10.11.4.124:80/b.csv","fileName": "b.csv"}})"
        datas=self.rfile.read(int(self.headers['content-length'])) 
        datas.decode()
        logger.info("data:"+str(datas))
        if mockconfig[4] == 'POST':
            self._doresponse(mockconfig)
        else:
            self.send_error(404)
            
        '''
        print(self.headers)
        print(self.command)
        req_datas = self.rfile.read(int(self.headers['content-length'])) #重点在此步!
        print(req_datas.decode())
        data = {
            'result_code': '2',
            'result_desc': 'Success',
            'timestamp': '',
            'data': {'message_id': '25d55ad283aa400af464c76d713c07ad'}
        }
        self.send_response(200)
        self.send_header('Content-type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(data).encode('utf-8'))'''

# 支持多线程
class ThreadingHttpServer(socketserver.ThreadingMixIn, HTTPServer):
    pass


try:
    server=ThreadingHttpServer(addr, RequestHandler)
    print("MockServer已经启动。")
    server.serve_forever()
except Exception as e:
    logger.error("MockServer异常：", e)

mock_utils.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import sys
import importlib
import logging

importlib.reload(sys)
# sys.setdefaultencoding("utf-8")
import json
from xml.etree.ElementTree import Element, SubElement, tostring
import random_utils


def dict_to_xml(mydict):
    objects=Element('xml')  # 根节点
    for isbn, info in mydict.items():  # 迭代每一个属性，属性和值（这里的每个值都是另一个对象）
        object=SubElement(objects, isbn)  # 创建子节点
        try:
            for key, val in info.items():
                SubElement(object, key).text=', '.join(str(val).split(':'))
        except:
            object.text=(', '.join(str(info).split(':')))
    return tostring(objects)


def dict_to_json(mydict):
    try:
        return json.dumps(mydict)
    except:
        return str(mydict)


def result_error(str_type, mydict):
    if (str(str_type).lower()).find('json') > 0:
        return dict_to_json(mydict)
    elif str(str_type).lower().find('xml') > 0:
        return dict_to_xml(mydict)
    else:
        return mydict


def pre_deal_testdata(data_str):
    """处理请求报文，替换请求报文中被参数化掉的参数"""
    data=str(data_str)
    if '$RANDOM' in data:
        data=data.replace("$RANDOM_UUID", random_utils.create_uuid()).replace("$RANDOM_MD5_UUID",
                                                                              random_utils.create_uuid_md5()).replace(
            "$RANDOM_EMAIL", random_utils.randomEmail()).replace("$RANDOM_STR8",
                                                                 random_utils.randomStr(8)).replace(
            "$RANDOM_STR16", random_utils.randomStr(16)).replace("$RANDOM_STR32", random_utils.randomStr(32))
    if '$TIME' in data:
        data=data.replace("$TIME_NOW", random_utils.time_now()).replace("$TIME_TODAY",
                                                                        random_utils.time_today()).replace(
            "$TIME_FUTURE", random_utils.time_future()).replace("$TIME_NEXT_MONTH",
                                                                random_utils.time_next_month())
    if '$DATE' in data:
        data=data.replace("$DATE_TODAY", random_utils.date_today()).replace("$DATE_NEXT_MONTH",
                                                                            random_utils.date_next_month()).replace(
            "$DATE_MONTH_LATER", random_utils.date_month_later())
    return data


# -*- coding: utf-8 -*-
import pymysql


# 打开数据库连接
def create_engine(host, port, user, password, dbname):
    db=pymysql.connect(host=host, port=port, user=user, password=password, db=dbname)
    return db


# 使用cursor()方法获取操作游标
def exc_sql(db, sql):
    if db:
        cursor=db.cursor()
    try:
        # 执行sql语句
        cursor.execute(sql)
        # 提交到数据库执行
        db.commit()
        return cursor.rowcount
    except:
        # Rollback in case there is any error
        db.rollback()
        return 0


# 使用cursor()方法获取操作游标
def select_sql(db, sql):
    if db:
        cursor=db.cursor()
    try:
        # 执行sql语句
        count=cursor.execute(sql)
        result=cursor.fetchone()
        return result
    except Exception as e:
        print(e)
        # Rollback in case there is any error
        db.rollback()
        return None


def db_close(db):
    db.close()

random_utils.py

# coding:utf-8

import random
import time
import datetime
import uuid,string
import hashlib

"""
产生随机参数，待需求确认需要哪些随机数据
"""


def create_uuid():
    """通过UUID的方式创建"""
    return str(uuid.uuid1())


def create_uuid_md5():
    """通过MD5的方式创建UUID"""
    m=hashlib.md5()
    m.update(bytes(str(time.time()), encoding = 'utf-8'))
    # m.update("小马过河".encode("utf-8"))  update对指定字符串进行加密
    return m.hexdigest()

def create_pwd_md5_complex(arg):
    """加盐"""
    m = hashlib.md5(bytes('jiayan', encoding= 'utf-8'))
    m.update(bytes(arg, encoding = 'utf-8'))
    return m.hexdigest()

def randomStr(i):
    """返回随机字符串, 从a-zA-Z0-9生成指定数量的随机字符："""
    return ''.join(random.sample(string.ascii_letters + string.digits, i))


def randomEmail():
    """返回随机邮箱"""
    name=''.join(random.sample(string.ascii_letters + string.digits, 4))+''.join(random.sample(string.ascii_letters + string.digits, 5))
    return name+"@iottest.com"


def time_now():
    """返回当前时间"""
    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())


def time_today():
    """返回今天的00:00:00点"""
    return time.strftime("%Y-%m-%d 00:00:00", time.localtime())


def time_future():
    """返回30天后的00:00:00点"""
    day = datetime.datetime.now()
    date = day + datetime.timedelta(days=30)
    return str(date)[0:10] + " 00:00:00"


def time_next_month():
    """返回次月1日凌晨00:00:00"""
    day_tuple = time.localtime()
    day_list = list(day_tuple)
    if day_list[1] == 12:
        day_list[0] += 1
        day_list[1] = 1
        day_list[2] = 1
    else:
        day_list[1] += 1
        day_list[2] = 1
    return time.strftime("%Y-%m-%d 00:00:00",tuple(day_list))


def date_today():
    """返回当前日期"""
    return time.strftime("%Y-%m-%d", time.localtime())


def date_next_month():
    """返回次月1号的日期"""
    day_tuple = time.localtime()
    day_list = list(day_tuple)
    if day_list[1] == 12:
        day_list[0] += 1
        day_list[1] = 1
        day_list[2] = 1
    else:
        day_list[1] += 1
        day_list[2] = 1
    return time.strftime("%Y-%m-%d", tuple(day_list))


def date_month_later():
    """返回30天后的日期"""
    day = datetime.datetime.now()
    date = day + datetime.timedelta(days=30)
    return str(date)[0:10]


if __name__ == '__main__':
    print(create_uuid())
    print(create_uuid_md5())
    print(create_pwd_md5_complex('test'))
    print(randomStr(8))
    print(randomStr(16))
    print(randomStr(32))
    print(randomEmail())
    print(time_now())
    print(time_today())
    print(time_future())
    print(time_next_month())
    print(date_today())
    print(date_next_month())
    print(date_month_later())

 3 请求示例

工具二：moco-runner

1 jar包地址：https://github.com/dreamhead/moco

2 配置参考：

   https://www.cnblogs.com/wzl0916/p/12678493.html

   https://www.cnblogs.com/bingoTest/p/11353575.html

 

json配置文件： test.json

 
[
  {
    "description":"moco test03",
    "request":{
        "uri":"/moco/test01"
    },
    "response":{
        "headers":{
            "Content-Type":"text/html;charset=utf8"
        },
        "text":"The first moco script!moco 第一个脚本。"
    }
  },
  {
    "description":"moco test03",
    "request":{
        "uri":"/moco/test02"
    },
    "response":{
        "headers":{
            "Content-Type":"text/html;charset=utf8"
        },
        "text":"The first moco script!moco第2个脚本。"
    }
  }
]

-------------------------------------------------------------------------------------------------

带参数请求

[  
{
    "description":"moco test03",
    "request":{
        "method":"get",
        "uri":"/moco/test01",
        "queries":{
            "UserName":"wzl",
            "Password":"123456"
        }
    },
    "response":{
        "headers":{
            "Content-Type":"text/html;charset=utf8"
        },
        "text":"Login success"
    }
  }
]

-------------------------------------------------------------------------------------------------------

响应读取外部文件

[
  {
    "description":"moco test03",
    "request":{
        "method":"get",
        "uri":"/moco/test01",
        "queries":{
            "UserName":"wzl",
            "Password":"123456"
        }
    },
    "response":{
        "headers":{
            "Content-Type":"text/html;charset=utf8"
        },
        "file":"success_result.json"
    }
  }
]
 
----success_result.json
{
    "status":"200",
    "msg":"Login Success",
    "data":{
        "UserName":"wzl",
        "Password":"123456"
    }
}

----------------------------------------------------------------------------------------------------------------------

 

3 启动 nohup java  -jar  moco-runner-1.1.0-standalone.jar http -p  3001  -c test.json > /dev/null 2>&1 &

 

4 实例-响应读取外部文件：

配置

[
  {
    "description":"moco big.data.list",
    "request":{
        "uri":"/data_subcribe_query"
    },
 
    "response":{
        "headers":{
            "Content-Type":"text/html;charset=utf8"
        },
        "file":"data_subcribe_query.json"
    }
  },
  {
    "description":"moco tag.info",
    "request":{
        "uri":"/data_file_get"
    },
 
    "response":{
        "headers":{
            "Content-Type":"text/html;charset=utf8"
        },
        "file":"data_file_get.json"
    }
  }
]

启动

nohup java -jar  ./moco-runner-1.1.0-standalone.jar http -p 8899 -c uts_mock.json  > /dev/null 2>&1 &