在django中如何从零开始搭建一个mock服务

mock概念

mock 就是模拟接口返回的一系列数据，用自定义的数据替换接口实际需要返回的数据，通过自定义的数据来实现对下级接口模块的测试。
这里分为两类测试：一类是前端对接口的mock，一类是后端单元测试中涉及的mock

mock服务的产生

在软件测试中经常会出现一些特殊的接口，如银行支付结果获取接口，这个接口不可能实际去支付，那么就需要一个服务来承担这个接口的任务，所谓服务就是针对大多数人而不是单纯的针对自己，同时是针对大多数这种模拟操作，而不单单只是接口，也可以模拟服务，这个时候单独的mock已经不是那么适用了。

如何搭建一个自定义的mock服务

mock服务需要承载用户的一些操作行为，这种行为包括查询，查看，修改，删除，新增。在我们实际开发的过程中应该考虑到前端的一些简洁和灵活性，设计前端基本遵循以下原则：
1、界面简洁，内容平易近人
2、功能齐全，操作流程简单
3、反应速度快，性能优秀
在django中设计前端时，本项目采用的是前后端分离的思想，将前端静态页面直接返回，后端数据接口返回的方式，实现分离思想。

 return render(request, 'page/mock/edit-mock-data.html', {"mock_data_id": mock_data_id})
 return render(request, "page/mock/mock-config.html")

mock服务中所有的操作数据都存放在数据库中，django的models类创建代码如下

class MockModel(models.Model):
    """
    mock model
    """
    relate_interface = models.CharField(max_length=255, unique=True,  default='', verbose_name='关联接口')
    mock_data_id = models.IntegerField(default=-1, verbose_name="对应mock数据id")
    service = models.CharField(max_length=255, default="", verbose_name="服务方")
    origin_url = models.CharField(max_length=500, default="", verbose_name="正常url地址")
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    remark = models.CharField(max_length=500, verbose_name="备注")
    status = models.CharField(max_length=10, default='notDefault', verbose_name="是否默认状态")
    update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间")

    class Meta:
        db_table = 't_mock'
        verbose_name ='mock数据表'
        verbose_name_plural = verbose_name


class MockDataModel(models.Model):
    """
    mock data model
    """
    # id = models.IntegerField(auto_created=True,unique=True, primary_key=True, verbose_name="id")
    mockName = models.CharField(max_length=255, null=False, default='',unique=True, verbose_name="mock名")
    # interface = models.ForeignKey(to=InterfaceModel, to_field='interface', related_name='related_mock', on_delete=models.SET(''), verbose_name='接口名' )
    interface_name_id = models.CharField(max_length=255, default='', verbose_name="关联接口")
    data = models.TextField(verbose_name="mock数据")
    author = models.CharField(max_length=50, null=False, default='', verbose_name='创建者')
    thirdpart = models.CharField(max_length=255, default='', verbose_name="三方接口")
    status = models.CharField(max_length=255, default='undefault', verbose_name="mock状态")
    status_code = models.IntegerField(default=200,verbose_name='状态码')
    timeout = models.IntegerField(default=0, verbose_name="超时时长")
    remarks= models.CharField(max_length=500, verbose_name="备注")
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间")

    class Meta:
        db_table = 't_mock_data'
        verbose_name ='mock表'
        verbose_name_plural= verbose_name

    def get_object(self):
        """
        :return:
        """
        return self.mockName
# mock服务表
class MockServiceModel(models.Model):
    service = models.CharField(max_length=255, null=False, default='',unique=True, verbose_name="服务名")
    status = models.CharField(max_length=255, default='1' ,verbose_name="状态")
    author = models.CharField(max_length=50, null=False, default='', verbose_name='创建者')
    remarks = models.CharField(max_length=500, default='', verbose_name="备注")
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间")

    class Meta:
        db_table = 't_mock_service'
        verbose_name ='mock服务名表'
        verbose_name_plural= verbose_name

    def __unicode__(self):
        return self.service


# interface_mock_map表
class MockInterfaceMapModel(models.Model):
    interface = models.ForeignKey(to=InterfaceModel, to_field='interface', on_delete=models.SET(""), verbose_name="接口名")
    mockIds = models.CharField(max_length=500, default='',  verbose_name="关联mockId")
    remarks = models.CharField(max_length=500, default='', verbose_name="备注")
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间")

    class Meta:
        db_table = 't_mock_interface_map'
        verbose_name ='mock与接口对应关系表'
        verbose_name_plural= verbose_name

项目设计了基本的mock表、mock服务表、mock数据表、以及后续需要使用到的mock与接口关联表。表的生产参考django官网。

后台接口中主要有以下核心接口

1、根据接口名获取mock数据

@csrf_exempt
def get_mock_data_by_interface(self, request, interface_name, others=''):
    """
    根据接口名获取mock数据
    :return:
    """
    try:
        mock_data_id = MockModel.objects.filter(relate_interface=interface_name).values_list("mock_data_id")[0][0]
        mock_data, status_code, timeout= MockDataModel.objects.filter(id=mock_data_id).values_list("data", "status_code", "timeout")[0]
        mock_data = json.loads(mock_data)
        if(timeout!=0):
            time.sleep(int(timeout))
        logging.debug(mock_data)
    except Exception as e:
        logging.error("根据接口名获取mock数据失败,测试数据格式错误")
        logging.error(e)
        status_code = 500
        mock_data = {"msg":"系统错误","code":status_code}
    return JsonResponse(mock_data, status=int(status_code), safe=False, json_dumps_params={"ensure_ascii":False})

@csrf_exempt装饰器来标识一个视图可以被跨域访问，类中的装饰写法可以参考上面代码，当然也可以在url中实现如

from django.conf.urls import url
from django.views.decorators.csrf import csrf_exempt
import views
urlpatterns = [
    url(r'^myview/$', csrf_exempt(views.MyView.as_view()), name='myview'),
]

搜索mock数据, 模糊匹配

@csrf_exempt
def search(self, request):
    """
    搜索mock数据, 模糊匹配
    :return:
    """
    query_form = request.POST.dict()
    logging.debug(query_form)
    new_query_form = {}
    for key, value in query_form.items():
        # 如果值不为空，将该键值对保存到新字典中
        if key == "relate_interface" and value != '':
            new_query_form["relate_interface"] = value
        elif key == 'service' and value != '':
            new_query_form["service"] = value
        elif key == 'search_input_value' and value != '':
            new_query_form["search_input_value"] = value
        else:
            continue
    # 获取符合条件的mock数据
    logging.debug(new_query_form)
    # 如果搜索条件为空，则返回全部数据,否则按条件过滤
    if new_query_form == {}:
        objs = MockModel.objects.all().order_by("create_time")
    else:
        try:
            logging.debug("search_input_value搜索字段不为空")
            mockFilterStr = new_query_form["search_input_value"]
            del new_query_form["search_input_value"]
            objs = MockModel.objects.filter(relate_interface__contains=mockFilterStr).filter(
                **new_query_form).order_by('create_time')
        except KeyError:
            logging.debug("mockFilterStr搜索字段不存在")
            objs = MockModel.objects.filter(**new_query_form).order_by('create_time')
    # total总数据，page:可分页数, limit:每页限制数(0)，objs：每页展示数(20)
    total = len(objs)
    if objs == None:
        response = {
            'code': 201,
            'msg': ' return None',
            'total': 0,
            'data': []
        }
    else:
        # 数据序列化
        objs = serializers.serialize("json", objs)
        obj_list = []
        for obj in eval(objs):
            obj_new = obj["fields"]
            try:
                obj_new['id'] = obj["pk"]
            except KeyError:
                continue
            obj_list.append(obj_new)
        response = {
            'code': 200,
            'msg': 'success',
            'total': total,
            'data': obj_list
        }
    return JsonResponse(response)

查询出来的数据是需要一定的数据处理的，接口内容中主要通过django.core的序列化方法objs = serializers.serialize("json", objs)进行处理。

mock数据删除接口

@staticmethod
def delete(request):
    """
    删除
    :return:
    """
    mock_data_id_list = request.POST.getlist("idList")
    logging.debug(mock_data_id_list)
    ids_string = ','.join(mock_data_id_list)if len(mock_data_id_list)>1 else mock_data_id_list[0]
    try:
        MockDataModel.objects.extra(where=['id IN ('+ ids_string +')']).delete()
        response = {
            'code': 200,
            'msg': '删除成功'
        }
    except KeyError:
        logging.warn("没有这个键")
        response = {
            'code': 201,
            'msg': '删除失败'
        }
    return JsonResponse(response)

数据删除这一块还是使用的自带orm操作，获取到批量的id列表，统一进行删除操作。

新增mock接口信息

@HD.decorate_create_model(MockModel)
def m_add_interface(self, request):
    '''
    新增mock接口信息
    :param request:
    :return:
    '''
    interfaceName = request.POST.get("t-addInterface")
    serverName = request.POST.get("s-mockServiceName")
    origin_url = request.POST.get("origin_url")
    logging.debug("接口名：{}".format(interfaceName))
    logging.debug("服务名：{}".format(serverName))
    if serverName and interfaceName:
        interfaceName = interfaceName.strip()
        serverName = serverName.strip()
        origin_url = origin_url.strip()
        return {"relate_interface":interfaceName, "service":serverName, "origin_url":origin_url}
    else:
        return {"msg": "接口名称或服务名为空，请重新输入", "code": 0}

以上为一些核心接口的代码设计

mock规则

后端接口的规划

在数据库的存储上，每一个接口信息可能对应多条返回数据，那么在实际请求一个接口时如何返回我们指定的数据呢？
1、我们将众多数据中的状态做一个标识，只保存一个有效状态，当我们需要返回某个数据时只需要修改为有效状态，而我们正是这样做的。

@csrf_exempt
def set_default_mock(self, request):
    """
    设置默认值
    :param mock_data_id:
    :return:
    """
    mock_data_id = request.POST.get("mock_data_id")
    # logging.debug(dict(MockModel.objects.filter(mock_data_id=mock_data_id)))
    try:
        self.set_mock_data_status(mock_data_id)
        response = {
            "code": 200,
            'msg': "设置成功",
            'data': [],
        }
    except Exception as e:
        logging.debug("没有对应的数据")
        response = {
            "code": 302,
            'msg': "设置失败",
            'data': e,
        }
    logging.debug(response)
    return JsonResponse(response)

2、在我们需要对某个接口进行请求获取mock数据时增加一个数据id的入参，返回数据时根据入参的数据id进行返回。
每一个接口对应多条mock数据，那么如果很多接口势必会导致在加载数据时前端获取开发的接口返回时间过长，渲染后给用户的感觉会有所延迟，那么如何提高给前端接口数据返回的速度呢？
1、前端增加分页处理，每次获取数据时返回当前页的数据，本项目就是这样处理的

@staticmethod
def decorate_http_table_response(ModelName):
    """
    :return:
    """
    def wrapper(func):
        def inner(*args, **kwargs):
            request_dict = func(*args, **kwargs)
            id = request_dict.get("id")
            #判断是根据id获取单条数据还是根据page和limit获取数据
            if id:
                objs = ModelName.objects.filter(id=request_dict['id'])
                total = 1
            else:
                #如果没有id，则根据传入的page和limit获取对应数据,page 和 limit 均默认为1
                page = request_dict.get("page") if (request_dict.get("page")) else 1
                limit = request_dict.get("limit") if (request_dict.get("limit")) else 20
                objs_all = ModelName.objects.get_queryset().order_by('-id')
                #total总数据，page:可分页数, limit:每页限制数，objs：每页展示数
                logging.debug(objs_all)
                total = len(objs_all)
                p = Paginator(objs_all, limit, 0)
                objs = p.page(page).object_list
            if objs == None:
                response = {
                    'code': 201,
                    'msg': ' return None',
                    'total': 0,
                    'data': []
                }
            else:
                #数据序列化
                objs = serializers.serialize("json", objs)
                obj_list = []
                for obj in eval(objs):
                    obj_new = obj["fields"]
                    try:
                        obj_new['id']=obj["pk"]
                    except KeyError:
                        continue
                    obj_list.append(obj_new)
                response = {
                    'code': 200,
                    'msg': 'success',
                    'total': total,
                    'data': obj_list
                }
                logging.debug(response)
            return JsonResponse(response)
        return inner
    return wrapper

通过特定的前端传参来标识当前是第几页，当前也需要返回几条数据，最后数据以二维数组的形式返回。
2、通过增加redis缓存，同时按方法1获取部分数据量来达到接口的快速返回。

异常处理

延时设置

在软件测试的实际应用中，经常会出现这么种情况，就是B接口依赖于A接口的返回，那么在测试的时候假设要模拟A接口出现异常返回很慢的情况时，又要如何开发mock呢？
这种场景的实现相对来说比较简单基本就是增加一个接口等待，延时返回数据，本项目中通过获取用户保存的延时时间，使用sheep方法执行等待。

状态码设置

关于返回的状态码，是直接通过修改返回的状态码来实现的，这一个相对较为简单不加说明。

return JsonResponse(mock_data, status=int(status_code), safe=False, json_dumps_params={"ensure_ascii":False})

不同类型数据返回

软件测试中对于mock的接口返回数据类型经常会出现特殊的需求，比如返回图片、xml、json等，那么又要如何开发呢？
首先数据库我们需要新增字段，字段类型分别为image、clob类型，将图片数据以二进制数据流的方式存放在image类型字段下，将xml、html数据存放在clob字段类型下。
在数据返回时对返回类型进行判断，然后对数据进行返回。

def index(request):  
    if request.GET["type"] == "img":  
        return HttpResponse(open("test.png","rb"),content_type="image/png")  
        ## 这里 返回图片  
    elif request.GET["type"] == "html":  
        return HttpResponse(open("1.html","rb"),content_type="text/html")  
        ## 返回 html文本  
    elif request.GET["type"] == "xml":  
        return HttpResponse(open("1.html","rb"),content_type="text/xml")  
        ##返回 xml文本  
    elif request.GET["type"] == "json":  
        return HttpResponse({"code":"ok"},content_type="application/json")  
        ##返回 json文本 

更多内容关注微信公众号 软件测试微课堂