python-mqtt自動(dòng)化測(cè)試

scq000 發(fā)布于2021-09-26 09:47 / 2595人閱讀

摘要：基礎(chǔ)數(shù)據(jù)獲取地址端口發(fā)送訂閱主題請(qǐng)求參數(shù)獲取這里是發(fā)送的函數(shù)

# -*- coding:utf-8 -*-from multiprocessing import Processimport paho.mqtt.publish as publishfrom Data.data import *import paho.mqtt.subscribe as subscribefrom common.yaml_common_method import *import time,json,yaml,os"""基礎(chǔ)數(shù)據(jù)yaml獲取"""data = yaml_method_read("data")host = data["host"]#IP地址prot = data["port"]#端口pub_url = data["pub_url"]+"/register"#發(fā)送sub_url = data["sub_url"]+"/register/response"#訂閱"""主題請(qǐng)求參數(shù)yaml獲取"""publish_yaml = yaml_method_read("publish")publish_msg = publish_yaml["upload_service"]client_id = time.strftime("mq:test:%Y%m%d%H%M%S",time.localtime(time.time()))class Mqtt_method_info():    def mqtt_publish(self):        #這里是發(fā)送的函數(shù)        msg = publish_msg        print("------------------------------Send_sub------------------------------")        print(pub_url)        print("----------------------------Send_message----------------------------")        print(msg)        msg = json.dumps(msg)        publish.single(pub_url, msg, qos=1, hostname=host, port=prot, client_id=client_id)        print("------------------------------End_send------------------------------")    def mqtt_subscribe(self):        #這里是訂閱的函數(shù)        print("--------------------------Subscribe_topics--------------------------")        print(sub_url)        msg = subscribe.simple(sub_url, qos=1, hostname=host, port=prot, client_id=client_id)        msg_payload = json.loads(msg.payload.decode("utf-8"))        print("---------------------------Return_message----------------------------")        print(msg_payload)        yaml_method_write("mqtt_return_data",msg_payload)    def implement(self):        pub = Process(target=Mqtt_method_info().mqtt_publish)#發(fā)送線程        sub = Process(target=Mqtt_method_info().mqtt_subscribe)#訂閱線程        sub.start()        time.sleep(1)        pub.start()        sub.join()        print("執(zhí)行完畢")if __name__ == "__main__":    Mqtt_method_info().implement()

?這里講一下python-mqtt測(cè)試腳本：

我們會(huì)使用到python的paho-mqtt庫，同樣pip安裝下就行，失敗的話大部分是需要用鏡像站的情況

這份代碼分為yaml文件讀取發(fā)送的數(shù)據(jù)，這里只有操作的方法，大家如果需要使用可以把發(fā)送、訂閱的函數(shù)中修改，只運(yùn)行方法，然后自己再通過各種文件管理數(shù)據(jù)