Python 实现自动化测试 dubbo 协议接口

前言

在工作或学习过程中，可能会遇到后端服务里有使用 dubbo 协议实现的接口，dubbo 协议接口的测试方法不同于 http/https 类型的接口，不能简单使用request.post的方法来完成自动化测试。

如果需要对 dubbo 协议的接口进行自动化测试，你会想到哪种解决方案呢？

今天的分享将提供一种方案由 Python 来实现 dubbo 协议的接口自动化测试。

准备工作：

• 安装Python3+的运行环境

• 使用python -m pip install telnetlib安装依赖包

PS：使用依赖包telnetlib是想通过 Python 来模拟 Linux 下对 dubbo 接口调用的命令行操作

Python 实现源码

  

   

    

     

    
    

     

      import json
     
    
   

    

     

    
    

     

      import telnetlib
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      class 
      Dubbo(telnetlib.Telnet):
     
    
   

    

     

    
    

     

          prompt = 
      'dubbo>'
     
    
   

    

     

    
    

     

          
     
    
   

    

     

    
    

     

          def 
      __init__(self, host=None, port=
      0):
     
    
   

    

     

    
    

     

            
      super().
      __init__(host, port)
     
    
   

    

     

    
    

     

            self.
      write(b
      '\n')
     
    
   

    

     

    
    

     

            
     
    
   

    

     

    
    

     

          def 
      command(self, flag, str_=
      ''):
     
    
   

    

     

    
    

     

            data = self.
      read_until(flag.
      encode())
     
    
   

    

     

    
    

     

            self.
      write(str_.
      encode() + b
      '\n')
     
    
   

    

     

    
    

     

            return data
     
    
   

    

     

    
    

     

            
     
    
   

    

     

    
    

     

          def 
      invoke(self, service_name, method_name):
     
    
   

    

     

    
    

     

            command_str = 
      "invoke {0}.{1}()".
      format(service_name, method_name)
     
    
   

    

     

    
    

     

            self.
      command(Dubbo.prompt, command_str)
     
    
   

    

     

    
    

     

            data = self.
      command(Dubbo.prompt, 
      "")
     
    
   

    

     

    
    

     

            data = json.
      loads(data.
      decode(Dubbo.coding, errors=
      'ignore').
      split(
      '\n')[
      0].
      strip())
     
    
   

    

     

    
    

     

            
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      if __name__ == 
      '__main__':
     
    
   

    

     

    
    

     

        conn = 
      Dubbo(
      'serviceIp', port)
     
    
   

    

     

    
    

     

        result = conn.
      invoke(
     
    
   

    

     

    
    

     

          
      "serviceName",
     
    
   

    

     

    
    

     

          
      "serviceName.method"
     
    
   

    

     

    
    

     

        )
     
    
   

    

     

    
    

     

        
      print(result)
     
    
  
 

  
 

源码重点详解

1、代码段：

command_str = "invoke {0}.{1}()".format(service_name, method_name)

(1)、invoke 为调用 dubbo 接口的命令名称；

(2)、{0} 为 invoke 命令后需要接的第一个参数，它的值将由传入进来的 service_name 取代，一般是传入服务名称，这里的写法只是为了占位；

(3)、{1} 为 invoke 命令后需要接的第二个参数，它的值将由传入进来的 method_name 取代，一般是传入被调用的方法名称，这里的写法只是为了占位；

(4)、() 为 invoke 命令后调用 method_name 方法时需要的格式，类似 Python 中如果想要调用一个函数是需要加上 () 才能调用成功的，如果有参数还需要在 () 里传入对应的参数；

(5)、问题：如果被调用的dubbo服务的方法需要传入对应的参数，怎么处理呢？

答：

将代码改写为："invoke {0}{1}({2})".format(service_name,method_name,param_data)，需要在 () 里添加占位符，并在调用该方法时传入对应的参数，一般是 key:value 格式的参数，即 Python 中的字典格式的数据；

2、代码段：

conn = Dubbo('serviceIp', port)

(1)、serviceIp：被调用服务所属的IP地址；

(2)、port：被调用服务所对应的可访问的外部暴露的端口；

PS：如果端口号为内部端口，这种方式调用会报错，必须是可访问的外部暴露端口。

绵薄之力

最后感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走

​这些资料，对于想进阶【自动化测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴我走过了最艰难的路程，希望也能帮助到你！凡事要趁早，特别是技术行业，一定要提升技术功底。希望对大家有所帮助…….