Django测试工具平台（二）---请求dubbo接口

前言

本系列的第二章内容，主要完成Dubbo接口的请求。如果你对python如何请求Dubbo不熟悉，可以参考之前写的一篇文章httprunner自动化实例（五）—Dubbo接口测试，本文主要讲解如何通过django实现功能。

---

按照平台的规划，“Dubbo接口请求”作为主打功能，需要放在一个单独的APP内，所以我们需要按照上一章介绍的步骤：

1、新建APP python manage.py startapp dubbo

2、将APP注册在settings.py文件中 INSTALLED_APPS=[ + dubbo.apps.DubboConfig ]

3、实现业务逻辑

4、将url注册在urls.py文件中

这里我们直接进入业务逻辑的开发，第一步和第二步可参考上一篇文章。

业务逻辑实现

1、配置文件的读取

上一章，我们直接将redis的连接信息写在代码中，这肯定是不合理的，所以这一章我们改进一下，对于这种服务的信息直接放在一个配置文件当中，方便修改。 在根目录下新建 conf.ini

[zookeeper_conf] zookeeper_address = 127.0.0.1,x.x.x.x # 这里填ZK的服务器ip

然后在TestPlatform/settings.py中加上

# 配置文件的路径 CONF_DIR = os.path.join(BASE_DIR, "conf.ini")

最后在utilsapp/common.py中完成一个通用的读取配置的方法

class ReadConf: ''' 通用的读取配置文件的方法 ''' def __int__(self): pass def get_conf(self, section, option): conf_path = settings.CONF_DIR cf = configparser.ConfigParser() cf.read(conf_path) conf = cf.get(section, option) return conf

2、从zk上获取dubbo注册信息

因为Dubbo接口是注册在zookeeper 上，因此 连接Dubbo接口的第一步肯定是去zookeeper 获取 Dubbo服务的注册信息（ip和端口号），这里我们使用python的第三方包来实现zookeeper的连接

pip install -i https://pypi.douban.com/simple/ kazoo

在dubbo目录下新建dubbo_client.py 用来获取服务的注册信息

class GetDubboService(object): def __init__(self): self.hosts = ReadConf().get_conf("zookeeper_conf", "zookeeper_address") if self.hosts: self.hosts = self.hosts.split(',') self.zk = KazooClient(hosts=self.hosts) self.zk.start() # 与zookeeper连接 else: print("请配置zk地址信息") sys.exit(0) def get_dubbo_info(self, dubbo_service): node = self.zk.get_children('/dubbo/' + dubbo_service + '/providers') from urllib import parse if node: server = parse.unquote(node[0]) dubbore = re.compile(r"^dubbo://([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+)", re.I) result = dubbore.match(server) if result: result = result.group(1) print("获取到dubbo部署信息" + result) return {"server_host": result.split(":")[0], "server_port": result.split(":")[1]} self.zk.stop()

3、请求Dubbo接口实现

获取注册信息之后，我们就可以通过python的三方包 telnetlib ，来建立连接，以及通过shell命令来 请求Dubbo接口（原理在之前的一篇文章有详细说明httprunner自动化实例（五）—Dubbo接口测试） 同样在dubbo_client.py 实现相关逻辑

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 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()

4、views.py页面逻辑实现

这里我们使用django restframework的 类视图的APIView类来实现，其中get方法通过dubbo的服务名来获取相关注册信息（ip和port）。post方法则是请求Dubbo接口。 在java 中一个方法通常的入参为 一个实体类比如 getName(User user) ，或者 不确定的数据类型以及不确定的入参个数，比如 getName(Integer id,String name) ，所以这里我们需要分两种情况传参

实体类 ： “params_type”:“class” 其他 ： “params_type”:“others”

class DubboApi(APIView): def get(self, request): ''' 获取服务的ip和端口 :param request: :return: ''' service_name = request.GET.get('service_name') dubbo_info = GetDubboService().get_dubbo_info(service_name) return Response(dubbo_info) def post(self, request, *args): """ 请求Dubbo接口 :param request: :return: """ service_name = request.data.get('service_name') dubbo_method = request.data.get('dubbo_method') # 多参数类型，多参数 params_type = request.data.get('params_type') params = request.data.get('params') dubbo_info = GetDubboService().get_dubbo_info(service_name) server_host = dubbo_info.get("server_host") server_port = dubbo_info.get("server_port") # 判断参数类型 , if params_type == "class": result = InvokeDubboApi(server_host, server_port).invoke_dubbo_api(service_name, dubbo_method, params) else: args = params result = InvokeDubboApi(server_host, server_port).invoke_dubbo_api(service_name, dubbo_method, *args) return Response(json.loads(result))

业务逻辑实现之后，将相关方法注册在urls.py上，启动服务即可进行调试

5、调试

这里给两个传参的示例。java中还有类似枚举类，没有返回值的方法 在httprunner自动化实例（五）—Dubbo接口测试也有说明，有需要的可以翻看这篇文章，避免踩坑

//实体类 { "service_name": "com.test.zl.GetService", "dubbo_method": "getName", "params_type":"class", "params": { "class": "com.test.model.student", "name": "可乐", "age": 18 } } // 其他 { "service_name": "ccom.test.zl.GetService", "dubbo_method": "getName", "params_type":"others", "params": [1,"可乐"] }

总结

本章相对上一章的内容，难度会有一些提升，如果你对其中有不理解的地方，欢迎留言！乐意为你解答，共同进步嘛 ！ 如果你有想实现的功能欢迎提交 本项目的代码已上传git https://github.com/627886474/TestPlatform（本章内容请切换分支 — git chechkout dubbo ）

如果你觉得项目对你有帮助，可以关注一下微信公众号，持续分享干货