国内大部分公司目前都是使用基于 Java 语言的 Dubbo 技术栈,而测试同事普遍对 Python 技术栈更为熟悉。为了使不懂 JAVA 代码的测试同事也能进行 Dubbo 接口层的测试,故对 HttpRunner 进行二次开发,添加对 Dubbo 接口的支持

1、实现原理
关于 HttpRunner 我想不用多做介绍,测试小伙伴应该都了解,这是一款非常优秀的面向 HTTP(S) 协议的通用测试框架,我们要做的是基于这个框架进行二次开发。

根据 Dubbo 官方文档中提到的:dubbo 可以通过 telnet 命令进行服务治理,详情见​​Dubbo 官方文档

而在 Python 中有一个第三方包 telnetlib,所以我们可以通过这个包来执行 telnet 命令,进而对 dubbo 接口进行调用

通过上图我们还了解到一个信息点,那就是如果我要通过 telnet 连接服务器,我需要 ip 还有 端口号。下面,让我们一步步来实现

2、Dubbo 服务图解
​​
​​

我们可以通过 Dubbo 服务的架构图了解到,要获取 Dubbo 服务,我们需要去 zookeeper 服务注册中心找到对应的服务即可

3、telnet 连接 Dubbo 服务
使用 Docker 容器部署微服务,每次重启服务 ip 都会发生变化(原因:容器间则不用 ip 直接通讯,而使用主机名、复服务名、网络别名),所以在使用 telnet 连接时需要获取动态的 ip,我司项目目前使用的 dubbo-monitor 进行管理

通过爬虫的思想,可动态获取服务 ip 和端口号,连接上服务器之后,我们就可以通过模拟命令行对服务进行治理

4、如何使用
4.1、请求服务相关代码

这段代码是处理请求的代码,放在 httprunner 源码文件下的 utils 文件夹内

import time
import os
import sys
import telnetlib


class TelnetClient(object):
    """通过telnet连接dubbo服务, 执行shell命令, 可用来调用dubbo接口
    """

    def __init__(self, server_host, server_port):
        self.tn = telnetlib.Telnet()
        self.server_host = server_host
        self.server_port = server_port

    # 此函数实现telnet登录主机
    def connect_dubbo(self):
        try:
            print("telent连接dubbo服务端: telnet {} {} ……".format(self.server_host, self.server_port))
            self.tn.open(self.server_host, port=self.server_port)
            return True
        except Exception as e:
            print('连接失败, 原因是: {}'.format(str(e)))
            return False

    # 执行传过来的命令,并输出其执行结果
    def execute_some_command(self, command):
        # 执行命令
        cmd = (command + '\n').encode("gbk")
        self.tn.write(cmd)

        # 获取命令结果,字符串类型
        retry_count = 0
        # 如果响应未及时返回,则等待后重新读取,并记录重试次数
        result = self.tn.read_very_eager().decode(encoding='gbk')
        while result == '':
            time.sleep(1)
            result = self.tn.read_very_eager().decode(encoding='gbk')
            retry_count += 1
        return result

    # 退出telnet
    def logout_host(self):
        self.tn.write(b"exit\n")
        print("登出成功")


class InvokeDubboApi(object):

    def __init__(self, server_host, server_port):
        try:
            self.telnet_client = TelnetClient(server_host, server_port)
            self.login_flag = self.telnet_client.connect_dubbo()
        except Exception as e:
            print("invokedubboapi init error" + str(e))



    def invoke_dubbo_api(self, dubbo_service, dubbor_method, *args):
        api_name = dubbo_service + "." + dubbor_method + "{}"
        cmd = "invoke " + api_name.format(args)
        print("调用命令是:{}".format(cmd))
        resp0 = None
        try:
            if self.login_flag:
                resp0 = self.telnet_client.execute_some_command(cmd)
                print("接口响应是,resp={}".format(resp0))
                # dubbo接口返回的数据中有 elapsed: 4 ms. 耗时,需要使用elapsed 进行切割
                return str(re.compile(".+").findall(resp0).pop(0)).split("elapsed").pop(0).strip()
            else:
                print("登陆失败!")
        except Exception as e:
            raise Exception("调用接口异常, 接口响应是resp={}, 异常信息为:{}".format(resp0, str(e)))
        self.logout()

    def logout(self):
        self.telnet_client.logout_host()

class GetDubboService2(object):
    def __init__(self):
        pass

    def get_dubbo_info2(self,content):
        try:
            dubbore = re.compile(r"([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+)", re.I)
            result = dubbore.search(str(content)).group()
            print("获取到dubbo部署信息" + result)
            return {"server_host": result.split(":")[0], "server_port": result.split(":")[1]}
        except Exception as e:
            raise Exception("获取dubbo部署信息失败:{}".format(str(e)))

这段代码放在 functions 中,这样在 yml 文件中可以直接通过 ${方法名}进行调用

​​

代码如下:

def invoke_dubbo2(content, dubbo_service, dubbo_method, *args):
    '''
    通过 zk管理后台查找dubbo注册信息
    :param dubbo_service:   dubbo中 服务名 如:com.zl.mall.api.IItemService
    :param dubbor_method:  服务中的方法 如:updateItem
    :param args: 方法请求需要的参数
    :return:
    '''
    dubbo_info = GetDubboService2().get_dubbo_info2(content)
    invokeDubboApi = InvokeDubboApi(server_host=dubbo_info.get("server_host"),
                                    server_port=dubbo_info.get("server_port"))
    return invokeDubboApi.invoke_dubbo_api(dubbo_service, dubbo_method, *args)

将代码添加至框架之后,即可在 yml 文件中使用

4.2、添加 testcase

和在项目中填写 Http testcase 类似,其中新增的几个 variables 参数为

dubbo_service: com.zl.item.api.IItemService ---Dubbo 服务名称(必填)
dubbo_method: queryItemByLstItemId ---Dubbo 服务方法名(必填)
iItemIdList: [123,12323] ---方法的请求参数(根据 Dubbo 定义的方法来调整入参)

${invoke_dubbo2()}    请求 dubbo 接口的方法,按照对应的格式填写即可请求 DUbbo 接口

4.3、运行 testcase

运行 testcase,查看控制台信息,其中 resp 即为 dubbo 接口返回的信息,对比 validate 中设置的预期值,即可完成对 Dubbo 接口的测试

5、注意事项
5.0、请求参数异常

 请求 Dubbo 接口如果填入的参数有误,会报 no such method 的错误,请检查一下参数是否正常

5.1、枚举类请求:

枚举类的类名:  com.zl.item.entity.StudentEnum 

需要使用到的枚举类:GOOD_STUDENT,填写格式如下 :

{"name": "GOOD_STUDENT", "class": "com.zl.item.entity.StudentEnum "}

5.2、实体类请求:

实体类类名:com.zl.item.entity.Student

实体类的字段:

public class student{
Long id;
String code}

{"class":"com.zl.item.entity.Student","Id":123,"code":"abc"}

5.3、对于请求方法是 void 方法的校验设置

java 代码中 void 这种方法是没有返回值的,所以我们直接用 “null”。如果需要进一步校验数据的准确性,可以校验这个方法改变的特性

5.4、对于入参是 Boolean 类型的数据

在 json 中,直接 使用{“data”:true}即可,但是在 python 请求时需要使用 “true”,加上双引号


↙↙↙阅读原文可查看相关链接,并与作者交流