Dubbo接口测试的Python实践-第2 章：使用Telnet测试

Posted 2021-04-06 翼支付自动化测试团队

就职于甜橙金融质量监控中心。

专注测试领域6年，致力于自动化测试开发以提高测试效率保证产品质量。

“6年小测试。”

                                                                                                ——夏晖

前言

Python语言实现对Dubbo接口的调用与测试，原理基于Dubbo官方文档中给出的一套通过telnet命令行的方式去调试Dubbo接口，测试Dubbo接口功能是否正确的方法。

0

核心测试源码（dubbo_core）

核心测试源码（dubbo_core）如下：

import json
import telnetlib
import socket
import sys
import io
import time

# 系统默认编码格式为 'utf-8'
coding = sys.getdefaultencoding()
class Dubbo(telnetlib.Telnet):
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding=coding)
    prompt = 'dubbo>'
    coding = coding

    def __init__(self, host=None, port=0, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
        super().__init__(host, port)
        self.write(b'\n')

    def command(self, flag, str_=""):
        self.write(str_.encode() + b"\n")
        # retry50次，如果没有返回，则等待0.1秒后再次获取输出
        for x in range(50):
            time.sleep(0.1)
            data = self.read_very_eager()
            data = data.replace(b'\r\n' + flag.encode(), b'')
            data = data.replace(b'dubbo>', b'')
            if data:
                return data
        return None

    def invoke(self, service_name, method_name, arg):
        # 返回 dubbo 结果 及 执行时间
        json_dumps = arg
        command_str = "invoke {0}.{1}({2})".format(service_name, method_name, json_dumps)
        data = self.command(Dubbo.prompt, command_str)
        # 中文显示 用gbk
        data = data.decode('gbk', errors='ignore').split('\r\n')
        print(data)
        print(data[0])
        data[0].replace('dubbo>', '')
        return data[0].replace('dubbo>', ''), data[1]

   # 断开dubbo返回
    def diss_connect(self):
        self.write(b"exit\n")

使用步骤：

2. 使用对象invoke方法吗，传入服务名称、方法名及传入json

3. 如若没有后续操作，使用对象diss_connect方法断开连接

返回：

1. 结果为一个长度为2的tuple

2. 第一部分为Dubbo接口的json返回

3. 第二部分为这次接口调用耗时。

单接口调用示例：

import json
import time
from dubbotest import dubbo_core

json_data = '{"xx":"yy","xxx":"yyy"}'
dubbo_service = "xxxService"
dubbo_method = "yyymethod "
# 1.建立dubbo实例，连接服务器
conn = dubbo_core.Dubbo("ervicesIp", 20880)
# 2.执行并获取dubbo返回
result = conn.invoke(dubbo_service, dubbo_method, json_data)
print("----测试结果-----")
print(result)
# 3.断开dubbo返回，最好每次调用后都断开，否则telnet输出可能会有不明输出，影响后续输出结果
conn.diss_connect()

调用dubbo核心测试类，将测试的ip端口和service.method传入即可测试，注意断开dubbo连结.

1

接口联调

测试时很多接口需要关联使用，组成测试链路，上一个接口返回值里某些数据作为下一个接口的入参：

步骤：

1. 建立Dubbo连接

2. 调用接口1，提取输出组成下个接口的入参

3. 调用接口2

4. 断开连接

【如若链路中还有更多接口，那么重复第二部操作就可以了，python处理json还是很方便的】

import json
import time
from dubbotest import dubbo_core

json_data = '{"xx":"yy","xxx":"yyy"}'
dubbo_service = "xxService"
dubbo_method = "xxmethod"
# 1.建立dubbo实例，连接服务器
conn = dubbo_core.Dubbo("services", 20880)
# 2.执行测试1并获取dubbo返回
result = conn.invoke(dubbo_service, dubbo_method, json_data)
print("----测试case1-----")
print(result)
#取出需要的参数
needparam = json.loads(result[0])["result"]["needparam"]
# 2.执行测试2并获取dubbo返回
json_data1 = '{"aa":"bb","needparam":"'+needparam+'"}'
dubbo_service1 = "aaservices"
dubbo_method1 = "aamethod"
result2 = conn.invoke(dubbo_service1, dubbo_method1, json_data1)
print("----testcase2-----")
print(result1)
# 3.断开dubbo返回，最好每次调用后都断开，否则telnet输出可能会有不明输出，影响后续输出结果
conn.diss_connect()

2

对json断言

对于测试来说，我们需要判断测试的结果是否符合我们预期，加上断言来实现，断言至少要有以下几种：“期望值”及“预期值”。

当“期望值”及“预期值”匹配则测试是通过。

考虑到json验证点的复杂性，“期望值”的关系有以下几种关系（==，！=，>,>=,<,<=,包含，不包含，被包含）。我们通过python的字典进行了对应关系的分类处理。

 

“期望值”设计规则如下：

1.  期望值通过jsonpath的方式指定jsonpath的值及其期望的判断方式。

    格式如下: jsonpath|action|value

2.  如若有多条验证点，那么使用&&符号进行验证点拆分。

三个验证点格式如下: jsonpath1|action1|value1 && jsonpath2|action2|value2 && jsonpath3|action3|value3

3.  Action包含：in/contains/==/!=/>/>=/</<=/notin这几类关键字关系的实现

 

示例:

import json
import jsonpath
import common.utils

#data是模拟接口返回的数据
data='{"result":{"aa":"bb","cc":{"ee":"ff","qq":"pp"},"errorCode":null,"errorMsg":null}'
#期望值。期望ee字段的值等于ff，并且希望qq字段的值大于0。以&&连接
expect_str = '$...ee|==|ff&&$...qq|>=|0'
def convert_str(val):
    return '"'+val+'"'

def convert_strs(val):
    return ",".join([convert_str(x) for x in val.split(",")])

#对期望值和实际值的一个断言
def judge(expect_str, actual):
    cb=common.utils.commonLib()
    status = "Pass"
    flag = True
    for x in expect_str.split("&&"):
        y = x.split("|")
        jpath = y[0]
        signal = y[1]
        value = y[2]
       signal_dict = {"in": lambda:value in (convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0]))),  
#期望值包含在实际值中
        "contains":  lambda:convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0])) in ",".join([convert_str(x) for x in value.split(",")]),    #实际值包含在期望值中
        "==": lambda:eval(            convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0])) + signal + (convert_strs(value))),#期望值包==实际值
        "!=": lambda:eval(            convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0])) + signal + (convert_strs(value))),#期望值包！=实际值
        ">": lambda :convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0]))>convert_str(value),#期望值包>实际值
        ">=": lambda: convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0])) >= (convert_str(value)),#期望值包>=实际值
        "<": lambda: convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0])) < (convert_str(value)),#期望值包<实际值
        "<=": lambda: convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0]))<=(convert_str(value)),#期望值包<=实际值
        "not in": lambda:convert_str(str(jsonpath.jsonpath(json.loads(actual), jpath)[0])) not in convert_strs(value)#期望值不包含实际值
        }
        flag=signal_dict[signal]()
        if not flag:
            status = "Fail"
    return [status]print(judge(expect_str,data))

写在最后

今天的分享就到这里，下次再会。有兴趣的同学一起讨论学习吧。

翼支付质量平台

翼支付质量平台 © 2018