现原理

根据Dubbo官方文档中提到的：dubbo可以通过telnet命令进行服务治理，可以通过telnet链接dubbo服务，再通过invoke方法调用dubbo接口

详情见http://dubbo.apache.org/zh-cn/docs/user/references/telnet.html

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

通过上面官方文档截图，我们可以看到，当我们拿到dubbo服务的IP和端口号，就能去调用指定的dubbo接口了。下面，让我们一步步来实现

dubbo架构

调用关系说明 服务容器负责启动，加载，运行服务提供者。服务提供者在启动时，向注册中心注册自己提供的服务。

服务消费者在启动时，向注册中心订阅自己所需的服务。

注册中心返回服务提供者地址列表给消费者，如果有变更，注册中心将基于长连接推送变更数据给消费者。

服务消费者，从提供者地址列表中，基于软负载均衡算法，选一台提供者进行调用，如果调用失败，再选另一台调用。

服务消费者和提供者，在内存中累计调用次数和调用时间，定时每分钟发送一次统计数据到监控中心。

Dubbo架构具有以下几个特点，分别是连通性、健壮性、伸缩性、以及向未来架构的升级性。

通过上面架构图我们可以类似 zookeeper 这样的服务注册中心找到对应的服务，所部署的机器和端口

也通过dubbo-monitor上面进行查询

python实现dubbo的调用

通过上述收到查到到要调用的dubbo接口所处的服务器IP和端口，我们就可以通过python实现dubbo的调用了。

详细代码如下：

import reimport telnetlibimport timeimport logging
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')logger=logging.getLogger(__name__)
'''方法调用案例：conn=InvokeDubboApi('127.0.0.1:88888')data={    'dubbo_service': 'xxx.xxx.xx.xxxx.xxxx.xxxx.Service',    'dubbo_method': 'xxxxx',    'parameters': ({"age":41,"name":"tom"},"sh",564645,)    }invoke=json.loads(conn.invoke_dubbo_api(data))conn.logout()'''
class TelnetClient(object):    """通过telnet连接dubbo服务, 执行shell命令, 可用来调用dubbo接口    """
    def __init__(self, server_host, server_port):        self.conn=telnetlib.Telnet()        self.server_host=server_host        self.server_port=server_port
    # telnet登录主机    def connect_dubbo(self):        try:            logging.info("telent连接dubbo服务端: telnet {} {} ……".format(self.server_host, self.server_port))            self.conn.open(self.server_host, port=self.server_port)            return True        except Exception as e:            logging.info('连接失败, 原因是: {}'.format(str(e)))            return False
    # 执行传过来的命令，并输出其执行结果    def execute_command(self, command):        # 执行命令        cmd='invoke {}\n'.format(command).encode("utf-8")        self.conn.write(cmd)        # 初始化调用次数        invoke_count=0        # 若调用无返回时，记录次数并重试        result=self.conn.read_very_eager().decode(encoding='utf-8').split('\r\n')[0]        while result=='':            time.sleep(1)            result=self.conn.read_very_eager().decode(encoding='utf-8').split('\r\n')[0]            invoke_count +=1            if invoke_count>=5:                logging.info("调用dubbo接口超过五次，调用失败")                return '调用dubbo接口失败'        return result    # 退出telnet    def logout_host(self):        self.conn.write(b"exit\n")        logging.info("登出成功")
class InvokeDubboApi(object):
    def __init__(self, content):        #解析dubbo部署的ip和port        try:            dubboaddrre=re.compile(r"([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+)", re.I)            result=dubboaddrre.search(str(content)).group()            server_host=result.split(":")[0]            server_port=result.split(":")[1]            logging.info("获取到dubbo部署信息" + result)        except Exception as e:            raise Exception("获取dubbo部署信息失败：{}".format(e))
        try:            self.telnet_client=TelnetClient(server_host, server_port)            self.login_flag=self.telnet_client.connect_dubbo()        except Exception as e:            logging.info("invokedubboapi init error" + e)
    #调用dubbo接口    def invoke_dubbo_api(self, data):        cmd=data.get("dubbo_service") + "." + data.get("dubbo_method") + "{}".format(data.get("parameters"))        logging.info("调用命令是：{}".format(cmd))        resp=None        try:            if self.login_flag:                result=self.telnet_client.execute_command(cmd)                logging.info("接口响应是,result={}".format(resp))                return result            else:                logging.info("登陆失败！")        except Exception as e:            raise Exception("调用接口异常, 接口响应是result={}, 异常信息为：{}".format(result, e))        self.logout()
    # 调用多个dubbo接口，注：确保所有接口是同一个ip和port    def invoke_dubbo_apis(self,datas):        summary=[]        if isinstance(datas,list):            for i in range(len(datas)):                result=self.invoke_dubbo_api(datas[i])                summary.append({"data":datas[i],"result":result})            return summary        else:            return "请确认入参是list"
    def logout(self):        self.telnet_client.logout_host()
if __name__=='__main__':    data={        'dubbo_service': 'xxx.xxx.xx.xxxx.xxxx.xxxxService',        'dubbo_method': 'xxxxx',        'parameters': ({"id":"123456789","mobile":12456},)    }    i=InvokeDubboApi('127.0.0.1:110741')    i.invoke_dubbo_api(data)    i.logout()

请求结果：

注意事项

• 请求参数

数据data中的参数字段parameters是一个元组，后面的 ‘,’ 不能少

• 请求参数异常

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

• 当要批量请求时

传入的参数必须是list，且需要同样的IP和端口。

多互联网新鲜资讯、工作奇淫技巧关注原创【飞鱼在浪屿】（日更新）

python广泛用于科学计算领域，并由于其易用性和可扩展性强广泛使用。Javascript用于浏览器前端领域。可是你是否想象Javascript中可以执行python代码吗？

梦想实现了，github源码Brython就是这么一个混血产物（github地址：https://github.com/brython-dev/brython）。

brython.js的接口brython()通过捕获“script type="text/python"”的python代码片段，转化为js函数，通过加载到eval()执行。更多的实现原理参考：https://github.com/brython-dev/brython/wiki/How%20Brython%20works

一个python写网页的例子：https://brython.info/

、首先，我们想，什么是 API 测试？

API 测试其实是一种用程序或工具来发送数据，同时验收系统的返回值的方法。这种测试更偏向于业务实现逻辑。常见的网络协议有 TCP、Http、webservice、socket 等，http?和 webservice 都是基于 TCP/IP 协议的应用层协议，webservice 是基于 http 的 soap 协议传输数据。

二、接口自动化测试的基本流程有（如图）：

1、在测试工具中登记待测交易的接口报文格式；

2、编写测试案例，向案例中添加交易接口并进行配置关联；

3、准备测试数据并对测试数据进行参数化；

4、测试工具自动执行自动化测试案例；

5、测试工具比对预期结果和返回结果，验证案例是否执行成功。

三、接口测试发现的典型问题

接口测试经常遇到的bug和问题，如下：

（1）传入参数处理不当，导致程序crash；

（2）类型溢出，导致数据读出和写入不一致；

（3）因对象权限未进行校验，可以访问其他用户敏感信息；

（4）状态处理不当，导致逻辑出现错乱；

（5）逻辑校验不完善，可利用漏洞获取非正当利益等。

四、举例天气API接口实战

我们今天就主要以天气API接口为例，使用python语言实现用例编写、封装及报告生成功能。

API信息：

天气API：http://www.51testing.com/html/88/n-4465288.html

URL:http://t.weather.sojson.com/api/weather/city/101030100

请求方式：get

参数：city 城市名称

①代码实现查询北京的天气信息

步骤：

1、新建 weather_api_test.py文件

代码实现

#-*-coding:GBK -*-

import requests

from pip._vendor.requests.models import Response

url='http://t.weather.sojson.com/api/weather/city/101030100'

r=requests.get(url)

response_data=r.json()

print(r.text)

返回结果：

②用例集成到Unittest

1、针对不同的参数场景进行测试

2、设置断言判断执行结果是否符合预期

实现原理：

首先导入requests 库、unitest 、时间库

其次，创建天气class类

然后，分别创建4个函数，分别实现存放路径、正常传参、异常传参、缺省参数功能

3、用例设计

代码实现：

新建 weather_api_unitest.py文件

#-*-coding:GBK -*-

import unittest

import requests

from time import sleep

class weathertest(unittest.TestCase):

def setUp(self):

self.url='http://t.weather.sojson.com/api/weather/city/101030100'

self.url_error='http://t.weather.sojson.com/api/weather/city/101030101'

self.url_no='http://t.weather.sojson.com/api/weather/city'

#参数正常

def test_weather_tianjing(self):

r=requests.get(self.url)

result=r.json()

self.assertEqual(result['status'],200)

self.assertEqual(result['message'],'success感谢又拍云(upyun.com)提供CDN赞助')

sleep(3)

#参数异常

def test_weather_param_error(self):

r=requests.get(self.url_error)

result=r.json()

self.assertEqual(result['status'],400)

self.assertEqual(result['message'],'获取失败')

sleep(3)

#参数缺省

def test_weather_no_param(self):

r=requests.get(self.url_no)

result=r.json()

self.assertEqual(result['status'],404)

self.assertEqual(result['message'],'Request resource not found.')

sleep(3)

if __name__=='_main_':

unittest.main()

③测试报告生成

1、创建文件夹如图，把测试用例放到test_case目录下

2、下载BSTestRunner模块并放置到python Lib文件夹下

如路径 C:\Python34\Lib

3、创建run.py 文件

代码：

import unittest

from BSTestRunner import BSTestRunner

import time

#指定测试用例和报告路径

test_dir='./test_case'

report_dir='./reports'

#加载测试用例

discover=unittest.defaultTestLoader.discover(test_dir, pattern='weather_api_unittest.py')

#定义报告格式

now=time.strftime('%Y-%m-%d %H_%M_%S')

report_name=report_dir+'/'+now+'test_report.html'

#运行用例并生成测试报告

with open(report_name,'wb') as f:

runner=BSTestRunner(stream=f,title="weather api test report",description="china city weather test report")

runner.run(discover)

4、运行run.py，在reports文件夹下查看生成的报告

五、总结

最后我们再来总结一下接口测试的常用知识点和你需要掌握的。

1、requests发送get请求和post请求的方法

get(url, params=None, **kwargs)

post(url, data=None, json=None, **kwargs)

2、parmas参数和data参数的区别

由于get请求无请求体，post请求有请求体。

使用params参数时，默认会把参数附加到url后面，所以发送get请求时应使用params参数。

使用data参数时，参数会存放到请求体中，所以发送post请求时不能使用params，应使用data，除非接口及支持get又支持post，同样get请求也不能使用data参数。

3、如何使用Seesion解决接口保持状态的问题

初始化Session实例，通过这个实例调用request()方法发送请求。

4、最重要的一个封装方法，并掌握这个封装该如何使用

主要针对get和post请求的接口。

总之，API 测试上手很简单，但做得好，做成工程化还真需要费一点力气，一些技术细节的把控和提升，会无形中提升整体的测试水准；而如何让 API 测试真正在我们的日常工作中发挥出最大作用，也需慢慢研究和调整的。

请关注+私信回复：“测试”就可以免费拿到软件测试学习资料。