利用python构建Mock系统

发布时间：2019-09-22 07:39:35编辑：auto阅读（1838）

    之前的一篇博文提到过利用打算利用python写个mock系统以方便测试，主要是因为所测系统A依赖于其他系统的输入，两个系统属于不同的项目且开发和测试成员都不同，单独针对A系统进行测试时由于A的依赖性导致很多测试点无法覆盖且由于分属不同项目联调成本也较高，于是乎想到通过Mock的方式规避此问题，由于小弟之前只会java，所以第一时间想到的是用java编写一个mock系统，但是被测系统是C++实现的，那么就涉及到网络字节序的转换问题，研究了半天后开写，接收到的请求死活不能正确解析出来，而且发现代码还写了不少，头大，旁边同事在研究python，在他的强烈推荐下我就考虑怎么用python去实现，利用业余时间突击学习了一个月的python，终于大功告成，下面记录一下我的学习历程和收获！

    1.迅速翻阅《python核心编程》第一部分，以及第二部分的“网络编程”章节（因为涉及两个系统的通信），由于有java的经验上手还是比较快的。

    2.根据书中内容用python编写个简单的Socket服务端，能接收到被测系统的请求；这时谷歌上搜到python已经有SocketServer模块，只要引入它及它的子类StreamRequestHandler并重写handler方法（具体请求处理）。于是接下来考虑handler方法的具体实现，主要分两部分：1.请求的接收2.请求结果的返回；第一部分查看python官方文档轻松实现，第二部分先根据两个系统间的请求和反馈格式拼装内容，拼装完后考虑到python没有Byte类型，如何将拼装的内容转化为字节流并且让C++实现的系统能够识别呢？网上搜了搜，发现有struct这个好东西！利用它的pack方法可以将内容转换成指定的格式，pack方法有两个参数，一是转换后的格式定义，二是待转换的内容，下面贴上格式定义符与C++和python中的对应关系：

    > 格式c类型python类型
    > xchar无(表示填充字节)
    > cchar长度为1的字符串
    > bsigned charinteger
    > Bunsigned charinteger
    > hshortinteger
    > Hunsigned shortinteger
    > iintinteger
    > Iunsigned intlong
    > llonginteger
    > Lunsigned longlong
    > qlong longlong
    > Qunsigned long longlong
    > ffloatfloat
    > ddoublefloat
    > schar[]string
    > pchar[]string
    > Pvoid*integer
    默认情况下struct根据本地机器字节顺序转换.不过可以用格式中的第一个字符来改变对齐方式.定义如下：
    > 字符字节顺序长度和对齐方式
    > @nativenative
    > =nativestandard
    > <little-endianstandard
    > >big-endianstandard
    > !network (= big-endian)standard

    例如两个系统事先定义的通信格式为：一个整型、两个长整型、一个长度为8的字符数组，那么转换格式符为：！i2l8s或者！illssssssss

    有了上面这个对应表，格式转换的问题轻松搞定，实验成功可以接收请求和返回内容

    3.虽然实现了但还是觉得不够完美，尤其是response内容的拼装，是一个个写死并一个个传入到pack方法中的，如果response内容很长那么就显得特别丑陋而且也不够灵活。网上找了半天，发现不用一个个传递进去，可以先将内容写入列表或元组中，再在pack方法中用*列表或*元组可实现将这些参数依次传递到pack方法中

    4.经过上一步骤，代码美观不少，但灵活性依旧没有解决，于是想到将这些参数变成配置文件，程序中解析配置文件内容再将内容写入列表中再传递到方法中，于是网上找了找python怎么解析xml,方法很多，最终选择了使用minidom解析，没有理由，也不知道它的优缺点，只是找到了一个具体的示例并且看懂了于是乎就决定用它了，这个东东的特性以及其他解析方式后续再研究，使用它解析时还是费了一番周折，其实原理就是把xml文件中的每一个节点当做叶子节点，可以层层解析也可以解析指定的节点，而且节点内容是该节点的子节点（叶子节点）。

    下面是我构造的xml文件以及python代码：

    <?xml version="1.0" encoding="UTF-8"?>
<response>
 <head>
    <version>420</version>
    <command>420</command>
    <passwd>12345678</passwd>
    <type>1</type>
    <status>0</status>
    <src>4</src>
    <dest>1</dest>
    <serializationHigh>1234567</serializationHigh>
    <serializationLow>7654321</serializationLow>
    <body_length>60</body_length>
    <result_length>2</result_length>
 </head>
 <result>
      <iids>2000000001</iids>
      <gids>10000617005</gids>
      <similaritys>95</similaritys>
      <p_w_picpath_type>0</p_w_picpath_type>
      <dimensions>5</dimensions>
 </result>
 <result>
      <iids>2000000006</iids>
      <gids>10000617006</gids>
      <similaritys>95</similaritys>
      <p_w_picpath_type>0</p_w_picpath_type>
      <dimensions>5</dimensions>
 </result>
</response>

    #!/usr/bin/python
# encoding: utf-8
'''
com.test.mockvsearch -- shortdesc
com.test.mockvsearch is a description
It defines classes_and_methods
@author:     user_name
          
@copyright:  2013 organization_name. All rights reserved.
          
@license:    license
@contact:    user_email
@deffield    updated: Updated
'''
import sys
import struct
import string
import types
from xml.dom import minidom
from SocketServer import (TCPServer as SERV,StreamRequestHandler as SRH)
# 定义获取协议头内容的方法
def gethead(a,b):
    nodes=a.getElementsByTagName(b).item(0);
    head=[];
    for value in nodes.childNodes:
        if value.nodeName=='version':
            head.append(string.atoi(value.firstChild.nodeValue))
        elif value.nodeName=='command':
            head.append(string.atoi(value.firstChild.nodeValue))
        elif value.nodeName=='passwd':
            head.append(str(value.firstChild.nodeValue))
        elif value.nodeName=='type':
            head.append(string.atoi(value.firstChild.nodeValue))
        elif value.nodeName=='status':
            head.append(string.atoi(value.firstChild.nodeValue))
        elif value.nodeName=='command':
            head.append(string.atoi(value.firstChild.nodeValue))
        elif value.nodeName=='src':
            head.append(string.atoi(value.firstChild.nodeValue))
        elif value.nodeName=='dest':
            head.append(string.atoi(value.firstChild.nodeValue))
        elif value.nodeName=='serializationHigh':
            head.append(string.atol(value.firstChild.nodeValue))
        elif value.nodeName=='serializationLow':
            head.append(string.atol(value.firstChild.nodeValue))
        elif value.nodeName=='body_length':
            head.append(string.atoi(value.firstChild.nodeValue))  
        elif value.nodeName=='result_length':
            head.append(string.atoi(value.firstChild.nodeValue))
      
    return head;
#定义获取协议体内容的方法
def getbody(element,bodyname):
    body=[];
    nodes=element.getElementsByTagName(bodyname);
    for i in  range(nodes.length):
        for node in nodes.item(i).childNodes:
            if node.nodeName=='iids':
                body.append(string.atol(node.firstChild.nodeValue));
            if node.nodeName=='gids':
                body.append(string.atol(node.firstChild.nodeValue));
            if node.nodeName=='similaritys':
                body.append(string.atoi(node.firstChild.nodeValue));
            if node.nodeName=='p_w_picpath_type':
                body.append(string.atoi(node.firstChild.nodeValue));
            if node.nodeName=='dimensions':
                body.append(string.atoi(node.firstChild.nodeValue));
    return body;
#定义消息格式的方法
def getbodyformat(list):
    format='';
    for var in list:
        if type(var) is types.IntType:
            format+='i';
        if type(var) is types.LongType:
            format+='Q';
    return format;
doc = minidom.parse(sys.argv[1]);
root= doc.documentElement;
#消息头格式
headformat="!2i8s4i2q2i"
#消息头内容
body1=gethead(root,'head');
#消息体内容
body2=getbody(root, 'result');
#消息体格式化
bodyformat=getbodyformat(body2);
#整个消息的格式化信息
rformat=headformat+bodyformat;
#整个响应消息的内容
response=body1+body2;
HOST='localhost'
PORT=8888;
ADDR=(HOST,PORT);
#定义请求响应处理类
class MyHandler(SRH):
    def handle(self):
        #接收请求
        MyData = self.request.recv(2048000);
        #打印请求长度
        print 'result length is:', len(MyData);
        #打印请求内容和转换格式
        print response,rformat;
        #拼装响应内容
        senddata=struct.pack(rformat,*response);
        #发送响应内容
        self.request.send(senddata);
#创建Socket服务       
myServer=SERV(ADDR,MyHandler);
#进入循环状态
myServer.serve_forever();


关键字：

上一篇： Python编写规则值得收藏

下一篇： 【Python】安装MySQL-pyth