使用Python进行dubbo接口测试

lsekfe

背景
　　大家在测试dubbo接口是不是特别痛苦？因为dubbo接口并不是比较常见的http协议的，而是dubbo协议的，测试dubbo接口的有几种方法，譬如jmeter自定义sampler调用，java连接zookeeper中心调用dubbo，telnet命令调用dubbo等。

　　痛点
　　相信大家都比较熟悉使用jmeter，看了上面的测试方案，肯定是首选jmeter，但是这里踩坑较多，比如下载的插件与dubbo版本不对应，有时候响应参数出现中文乱码，有时候需要反编译jar包查看对应的入参类型等等...

　　解决痛点
　　在网上搜索了一下和看了dubbo接口的用户手册，发现dubbo接口支持telnet命令执行，Python中刚好有个库可以执行telnet命令 telnetlib库 dubbo接口又是通过zookeeper中心进行注册服务的，那我直接通过zookeeper中心查询dubbo接口相关的信息（ip、端口、服务名、方法名和入参类型）然后模拟telnet命令进行dubbo接口调用，岂不是美滋滋！

　　实现方案
　Python + fastapi + telnetlib + kazoo。

　　调用流程
　　telnet命令实操
　　前提，我们得知了一个dubbo接口的ip和端口（通常可以通过服务名在zk中心搜索得知）:

1. telnet 192.168.xx.xx 32024 连接dubbo服务

复制代码

1. ls -l cn.com.api.dubbo.xxxxService 查询该服务的方法列表

复制代码

1. invoke xxxxService.xxxMethod(1234, "test") 调用服务的方法
   

复制代码

通过上述一系列的骚操作，相当于进行了dubbo接口的测试，当然，我们决对不可能通过终端敲命令进行dubbo接口测试，下面我们进行通过代码模拟telnet命令。

　　核心源码分析
　　zk中心搜索服务封装

class GetDubboService(object):
    def __init__(self):
        #测试环境，ZK_CONFIG为zk中心的注册地址，可传入string或者list，譬如ZK_CONFIG = ['xxx','xxx','xxx']或者ZK_CONFIG = 'xxx'
        self.hosts = ZK_CONFIG
        self.zk = self.zk_conn()

    def zk_conn(self):
        try:
            zk = KazooClient(hosts=self.hosts, timeout=2)
            zk.start(2)  # 与zookeeper连接
        except BaseException as e:
            return False
        return zk

    def get_dubbo_info(self, dubbo_service):
        global data
        dubbo_service_data = {}
        try:
            #先查出注册中心所有的dubbo服务
            all_node = self.zk.get_children('/dubbo')
            # 根据传入服务名匹配对应的服务
            node = [i for i in all_node if dubbo_service.lower() in i.lower()]
            # 查询dubbo服务的详细信息
            #遍历数据，过滤掉空数据
            for i in node:
                if self.zk.get_children(f'/dubbo/{i}/providers'):
                    dubbo_data = self.zk.get_children(f'/dubbo/{i}/providers')
                    for index, a in enumerate(dubbo_data):
                        url = parse.urlparse(parse.unquote(a)).netloc
                        host, port = url.split(":")
                        conn = BmDubbo(host, port)
                        #判断获取的ip地址是否连接成功，因为有些开发本地起了dubbo服务
                        status = conn.command("")
                        if status:
                            data = dubbo_data[index]
                            break
            self.zk.stop()
        except BaseException as e:
            return dubbo_service_data
        #parse.unquote 解码
        #parse.urlparse 解析URL
        #parse.query 获取查询参数
        #parse.parse_qsl 返回列表
        url_data = parse.urlparse(parse.unquote(data))
        query_data = dict(parse.parse_qsl(url_data.query))
        query_data['methods'] = query_data['methods'].split(",")
        dubbo_service_data['url'] = url_data.netloc
        dubbo_service_data['dubbo_service'] = dubbo_service
        dubbo_service_data.update(query_data)
        return dubbo_service_data

telnet命令调用封装：

1. class BmDubbo(object):
   
2. 
   
3.     prompt = 'dubbo>'
   
4. 
   
5.     def __init__(self, host, port):
   
6.         self.conn = self.conn(host, port)
   
7. 
   
8.     def conn(self,host, port):
   
9.         conn = telnetlib.Telnet()
   
10.         try:
    
11.             conn.open(host, port, timeout=1)
    
12.         except BaseException:
    
13.             return False
    
14.         return conn
    
15. 
    
16.     def command(self, str_=""):
    
17.         # 模拟cmd控制台 dubbo>invoke ...
    
18.         if self.conn :
    
19.             self.conn.write(str_.encode() + b'\n')
    
20.             data = self.conn.read_until(self.prompt.encode())
    
21.             return data
    
22.         else:
    
23.             return False
    
24. 
    
25.     def invoke(self, service_name, method_name, arg):
    
26.         command_str = "invoke {0}.{1}({2})".format(service_name, method_name, arg)
    
27.         data = self.command(command_str)
    
28.         try:
    
29.             # 字节数据解码 utf8
    
30.             data = data.decode("utf-8").split('\n')[0].strip()
    
31.         except BaseException:
    
32.             # 字节数据解码 gbk
    
33.             data = data.decode("gbk").split('\n')[0].strip()
    
34.         return data
    
35. 
    
36.     def ls_invoke(self, service_name):
    
37.         command_str = "ls -l {0}".format(service_name)
    
38.         data = self.command(command_str)
    
39.         if "No such service" in data.decode("utf-8"):
    
40.             return False
    
41.         else:
    
42.             data = data.decode("utf-8").split('\n')
    
43.             key = ['methodName', 'paramType','type']
    
44.             dubbo_list = []
    
45.             #这里解析有点复杂，可以自己通过telnet命令实操一下，ls -l xxx
    
46.             for i in range(0, len(data) - 1):
    
47.                 value = []
    
48.                 dubbo_name = data[i].strip().split(' ')[1]
    
49.                 method_name = re.findall(r"(.*?)[(]", dubbo_name)[0]
    
50.                 value.append(method_name)
    
51.                 paramType = re.findall(r"[(](.*?)[)]", dubbo_name)[0]
    
52.                 paramTypeList = paramType.split(',')
    
53.                 if len(paramTypeList) ==1:
    
54.                     paramTypeList = paramTypeList[0]
    
55.                 value.append(paramTypeList)
    
56.                 #这里我将传参类型分成了4大类
    
57.                 if 'java.lang' in paramType or 'java.math' in paramType:
    
58.                     value.append(0)
    
59.                 elif not paramType:
    
60.                     value.append(1)
    
61.                 elif 'List' in paramType:
    
62.                     value.append(2)
    
63.                 else:
    
64.                     value.append(3)
    
65.                 dubbo_list.append(dict(zip(key, value)))
    
66.             return dubbo_list
    
67. 
    
68.     def param_data(self,service_name,method_name):
    
69.         #这里是根据服务名和方法名，找到对应的传参类型
    
70.         dubbo_data = self.ls_invoke(service_name)
    
71.         if dubbo_data:
    
72.             dubbo_list = dubbo_data
    
73.             if dubbo_list:
    
74.                 for i in dubbo_list:
    
75.                     for v in i.values():
    
76.                         if v == method_name:
    
77.                             param_key = ['paramType','type']
    
78.                             param_value = [i.get('paramType'),i.get('type')]
    
79.                             return dict(zip(param_key,param_value))
    
80.             else:
    
81.                 return False
    
82.         else:
    
83.             return False

复制代码

dao层设计：

1. class DubboHandle(object):
   
2.     @staticmethod
   
3.     def invoke(service_name, method_name, data):
   
4.         zk_conn = GetDubboService()
   
5.         if zk_conn.zk:
   
6.             zk_data = zk_conn.get_dubbo_info(service_name)
   
7.             if zk_data:
   
8.                 host, port = zk_data['url'].split(":")
   
9.                 service_name = zk_data['interface']
   
10.                 boby = data.copy()
    
11.                 conn = BmDubbo(host, port)
    
12.                 status = conn.command("")
    
13.                 if status:
    
14.                     # 根据服务名和方法名，返回param方法名和类型
    
15.                     param_data = conn.param_data(service_name, method_name)
    
16.                     if param_data:
    
17.                         type = param_data['type']
    
18.                         param = param_data['paramType']
    
19.                         # 传参类型为枚举值方法
    
20.                         if type == 0 and isinstance(boby, dict):
    
21.                             l_data = []
    
22.                             for v in  boby.values():
    
23.                                 if isinstance(v,str):
    
24.                                     v = f"'{v}'"
    
25.                                 elif isinstance(v,dict) or isinstance(v,list):
    
26.                                     v = json.dumps(v)
    
27.                                     v = f"'{v}'"
    
28.                                 l_data.append(str(v))
    
29.                             boby = ','.join(l_data)
    
30.                         # 无需传参
    
31.                         elif type == 1:
    
32.                             boby = ''
    
33.                         # 传参类型为集合对象
    
34.                         elif type == 2:
    
35.                             # params 只有一个集合对象传参
    
36.                             if isinstance(boby, list):
    
37.                                 boby = boby
    
38.                             # params 一个集合对象后面跟着多个枚举值
    
39.                             elif isinstance(boby, dict):
    
40.                                 set_list = []
    
41.                                 for v in boby.values():
    
42.                                     set_list.append(v)
    
43.                                 set_data = str(set_list)
    
44.                                 boby = set_data[1:-1]
    
45.                         # 传参类型为自定义对象
    
46.                         elif type == 3:
    
47.                             # 兼容多个自定义对象传参
    
48.                             if isinstance(param, list):
    
49.                                 dtoList = []
    
50.                                 for index, dto in enumerate(boby):
    
51.                                     dto.update({"class": param[index]})
    
52.                                     dtoList.append(json.dumps(dto))
    
53.                                 boby = ','.join(dtoList)
    
54.                             elif isinstance(boby, dict):
    
55.                                 boby.update({"class": param})
    
56.                                 boby = json.dumps(boby)
    
57.                         else:
    
58.                             return None, f"data请求参数有误,请检查！"
    
59.                         response_data = conn.invoke(service_name, method_name, boby)
    
60.                         try:
    
61.                             response_data = json.loads(response_data)
    
62.                         except Exception as e:
    
63.                             return None, f"解析json失败:{response_data}"
    
64.                         return response_data, None
    
65.                     else:
    
66.                         return None, f"{service_name.split('.')[-1]}服务下不存在{method_name}方法"
    
67.                 else:
    
68.                     return None, f"{service_name}服务连接出错"
    
69.             else:
    
70.                 return None, f"{service_name}没有在zk中心注册"
    
71.         else:
    
72.             return None, "zk服务连接失败"

复制代码

view层引用：

1. @router.post('/invoke', name='dubbo业务请求接口')
   
2. async def dubboInvoke(data: DubboInvokeBody):
   
3.     res_data, err = DubboHandle.invoke(data.serviceName, data.methodName, data.data)
   
4.     if err:
   
5.         return res_400(msg=err)
   
6.     return res_200(data=res_data)

复制代码

　invoke接口传参说明
　　原生对象或者自定义对象传参(xxDto、jsonObj、java.util.HashMap)：

1. {
   
2.     "serviceName": "xxxxxx",
   
3.     "methodName": "xxxxxx",
   
4.     "data": {        //data传入对应的对象数据,一般为json格式的
   
5.         "productStoreQueryDTOS": [
   
6.             {
   
7.                 "productNoNumDTOList": [
   
8.                     {
   
9.                         "num": 13,
   
10.                         "productNo": "10000620"
    
11.                     },
    
12.                     {
    
13.                         "num": 13,
    
14.                         "productNo": "10000014"
    
15.                     }
    
16.                 ],
    
17.                 "storeCode": "4401S1389"
    
18.             }
    
19.         ]
    
20.     }
    
21. }

复制代码

枚举值类型传参(java.lang.String、java.lang.Integer)：

1. {
   
2.     "serviceName": "xxxx",
   
3.     "methodName": "xxxxx",
   
4.     "data": {         //格式为json，枚举值顺序必须按照dubbo接口定义的传参顺序，注意是否为int还是string
   
5.         "account":"123456",
   
6.         "password":"3fd6ebe43dab8b6ce6d033a5da6e6ac5"
   
7.     }
   
8. }

复制代码

方法名无需传参：

1. {
   
2.     "serviceName": "xxxx",
   
3.     "methodName": "xxxxxx",
   
4.     "data":{}      //传入空对象
   
5. }

复制代码

集合对象传参(java.util.List)：

1. {
   
2.     "serviceName": "xxxx",
   
3.     "methodName": "xxxxxx",
   
4.     "data":{
   
5.         "List": [
   
6.             "1221323",
   
7.             "3242442"
   
8.         ]
   
9.     } //传入对象，里面嵌套数组
   
10. }

复制代码

　　集合对象传参,后面跟着枚举值(java.util.List 、 java.lang.String 、 java.lang.Integer)：

1. {
   
2.     "serviceName": "xxxx",
   
3.     "methodName": "xxxxxx",
   
4.     "data":{
   
5.         "userCode": ["12345","686838"],
   
6.         "startTime": "2021-04-16 13:30:00",
   
7.         "endTime": "2021-04-16 14:30:00"
   
8. }
   
9. }
   

复制代码

多个自定义对象传参,对象顺序按照dubbo接口定义的传参顺序(xxdtox、xxdto)：

1. {
   
2.     "serviceName": "xxxx",
   
3.     "methodName": "xxxxxx",
   
4.     "data":[
   
5.       {
   
6.         "userCode": "7932723",
   
7.         "startTime": "2021-04-16 13:30:00",
   
8.         "endTime": "2021-04-16 14:30:00"
   
9. },
   
10.       {
    
11.         "name": "fang",
    
12.         "age": "18"
    
13. }
    
14.     ]
    
15. }

复制代码

上述传参可以满足大部分入参类型~

　　疑惑
　　问：dubbo的传输协议，本身支持http协议，跟开发沟通，测试环境切换为http协议，不就方便测试了么？
　　答：公司内部系统对接大多走的是 Dubbo 协议，这是公司的开发规范，只能从外部绕了。
　　问：为什么我部署之后，连接不上zk服务或者出现dubbo服务连接出错？
　　答：部署服务的主机必须可以连通dubbo服务。

　　总结
　　本期解决了测试dubbo接口的痛点，希望能对大家有帮助~