現原理

根據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 測試真正在我們的日常工作中發揮出最大作用，也需慢慢研究和調整的。

請關注+私信回復：“測試”就可以免費拿到軟件測試學習資料。