diff --git a/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py b/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py index be6d1e3..5f47580 100644 --- a/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py +++ b/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py @@ -2,45 +2,46 @@ import re import os import sys import time -import socket -import imp import ssl import paho.mqtt.client as mqtt -g_recv_data="" -g_recv_topic="" -g_broker_connected=0 +try: + import IDF +except ImportError: + # this is a test case write with tiny-test-fw. + # to run test cases outside tiny-test-fw, + # we need to set environment variable `TEST_FW_PATH`, + # then get and insert `TEST_FW_PATH` to sys path before import FW module + test_fw_path = os.getenv("TEST_FW_PATH") + if test_fw_path and test_fw_path not in sys.path: + sys.path.insert(0, test_fw_path) + import IDF + +import DUT + +g_recv_data = "" +g_recv_topic = "" +g_broker_connected = 0 + # The callback for when the client receives a CONNACK response from the server. def on_connect(client, userdata, flags, rc): global g_broker_connected - print("Connected with result code "+str(rc)) + print("Connected with result code " + str(rc)) g_broker_connected = 1 client.subscribe("/topic/qos0") + # The callback for when a PUBLISH message is received from the server. def on_message(client, userdata, msg): global g_recv_topic global g_recv_data payload = msg.payload.decode() - if g_recv_data == "" and payload == "data": + if g_recv_data == "" and payload == "data": client.publish("/topic/qos0", "data_to_esp32") g_recv_topic = msg.topic g_recv_data = payload - print(msg.topic+" "+str(payload)) - -# this is a test case write with tiny-test-fw. -# to run test cases outside tiny-test-fw, -# we need to set environment variable `TEST_FW_PATH`, -# then get and insert `TEST_FW_PATH` to sys path before import FW module -test_fw_path = os.getenv("TEST_FW_PATH") -if test_fw_path and test_fw_path not in sys.path: - sys.path.insert(0, test_fw_path) - -import TinyFW -import IDF -import DUT - + print(msg.topic + " " + str(payload)) @IDF.idf_example_test(env_tag="Example_WIFI") @@ -48,7 +49,7 @@ def test_examples_protocol_mqtt_ssl(env, extra_data): global g_recv_topic global g_recv_data global g_broker_connected - broker_url="iot.eclipse.org" + broker_url = "iot.eclipse.org" """ steps: | 1. join AP and connects to ssl broker @@ -60,8 +61,8 @@ def test_examples_protocol_mqtt_ssl(env, extra_data): # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "mqtt_ssl.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("mqtt_ssl_bin_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("mqtt_ssl_size", bin_size//1024) + IDF.log_performance("mqtt_ssl_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("mqtt_ssl_size", bin_size // 1024) # 1. start test (and check the environment is healthy) dut1.start_app() client = None @@ -73,15 +74,15 @@ def test_examples_protocol_mqtt_ssl(env, extra_data): client.on_connect = on_connect client.on_message = on_message client.tls_set(None, - None, - None, cert_reqs=ssl.CERT_NONE, tls_version=ssl.PROTOCOL_TLSv1, ciphers=None) + None, + None, cert_reqs=ssl.CERT_NONE, tls_version=ssl.PROTOCOL_TLSv1, ciphers=None) client.tls_insecure_set(True) print("Connecting...") client.connect(broker_url, 8883, 60) print("...done") except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - except: + except Exception: print("ENV_TEST_FAILURE: Unexpected error while connecting to broker {}: {}:".format(broker_url, sys.exc_info()[0])) raise print("Start Looping...") @@ -92,7 +93,7 @@ def test_examples_protocol_mqtt_ssl(env, extra_data): if g_broker_connected == 0: raise ValueError('ENV_TEST_FAILURE: Test script cannot connect to broker: {}'.format(broker_url)) # 3. check the message received back from the server - if g_recv_topic == "/topic/qos0" and g_recv_data == "data" : + if g_recv_topic == "/topic/qos0" and g_recv_data == "data": print("PASS: Received correct message") else: print("Failure!") @@ -100,5 +101,6 @@ def test_examples_protocol_mqtt_ssl(env, extra_data): # 4. check that the esp32 client received data sent by this python client dut1.expect(re.compile(r"DATA=data_to_esp32"), timeout=30) + if __name__ == '__main__': test_examples_protocol_mqtt_ssl() diff --git a/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py b/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py index e2568d5..05e5142 100644 --- a/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py +++ b/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py @@ -1,36 +1,53 @@ import re import os import sys -from socket import * +import socket from threading import Thread import struct import time -msgid=-1 + +try: + import IDF +except ImportError: + # this is a test case write with tiny-test-fw. + # to run test cases outside tiny-test-fw, + # we need to set environment variable `TEST_FW_PATH`, + # then get and insert `TEST_FW_PATH` to sys path before import FW module + test_fw_path = os.getenv("TEST_FW_PATH") + if test_fw_path and test_fw_path not in sys.path: + sys.path.insert(0, test_fw_path) + import IDF + +import DUT + +msgid = -1 + def get_my_ip(): - s1 = socket(AF_INET, SOCK_DGRAM) + s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s1.connect(("8.8.8.8", 80)) my_ip = s1.getsockname()[0] s1.close() return my_ip + def mqqt_server_sketch(my_ip, port): global msgid print("Starting the server on {}".format(my_ip)) s = None try: - s=socket(AF_INET, SOCK_STREAM) + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(60) s.bind((my_ip, port)) s.listen(1) - q,addr=s.accept() + q,addr = s.accept() q.settimeout(30) print("connection accepted") - except: + except Exception: print("Local server on {}:{} listening/accepting failure: {}" - "Possibly check permissions or firewall settings" - "to accept connections on this address".format(my_ip, port, sys.exc_info()[0])) + "Possibly check permissions or firewall settings" + "to accept connections on this address".format(my_ip, port, sys.exc_info()[0])) raise data = q.recv(1024) # check if received initial empty message @@ -47,20 +64,6 @@ def mqqt_server_sketch(my_ip, port): s.close() print("server closed") -# this is a test case write with tiny-test-fw. -# to run test cases outside tiny-test-fw, -# we need to set environment variable `TEST_FW_PATH`, -# then get and insert `TEST_FW_PATH` to sys path before import FW module -test_fw_path = os.getenv("TEST_FW_PATH") -if test_fw_path and test_fw_path not in sys.path: - sys.path.insert(0, test_fw_path) - -import TinyFW -import IDF -import DUT - - - @IDF.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_mqtt_qos1(env, extra_data): @@ -76,14 +79,14 @@ def test_examples_protocol_mqtt_qos1(env, extra_data): # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "mqtt_tcp.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("mqtt_tcp_bin_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("mqtt_tcp_size", bin_size//1024) + IDF.log_performance("mqtt_tcp_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("mqtt_tcp_size", bin_size // 1024) # 1. start mqtt broker sketch host_ip = get_my_ip() - thread1 = Thread(target = mqqt_server_sketch, args = (host_ip,1883)) + thread1 = Thread(target=mqqt_server_sketch, args=(host_ip,1883)) thread1.start() # 2. start the dut test and wait till client gets IP address - dut1.start_app() + dut1.start_app() # waiting for getting the IP address try: ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) @@ -91,10 +94,10 @@ def test_examples_protocol_mqtt_qos1(env, extra_data): except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - print ("writing to device: {}".format("mqtt://" + host_ip + "\n")) + print("writing to device: {}".format("mqtt://" + host_ip + "\n")) dut1.write("mqtt://" + host_ip + "\n") thread1.join() - print ("Message id received from server: {}".format(msgid)) + print("Message id received from server: {}".format(msgid)) # 3. check the message id was enqueued and then deleted msgid_enqueued = dut1.expect(re.compile(r"OUTBOX: ENQUEUE msgid=([0-9]+)"), timeout=30) msgid_deleted = dut1.expect(re.compile(r"OUTBOX: DELETED msgid=([0-9]+)"), timeout=30) @@ -105,5 +108,6 @@ def test_examples_protocol_mqtt_qos1(env, extra_data): print("Failure!") raise ValueError('Mismatch of msgid: received: {}, enqueued {}, deleted {}'.format(msgid, msgid_enqueued, msgid_deleted)) + if __name__ == '__main__': test_examples_protocol_mqtt_qos1() diff --git a/examples/protocols/mqtt/ws/mqtt_ws_example_test.py b/examples/protocols/mqtt/ws/mqtt_ws_example_test.py index 276f1d3..3567cae 100644 --- a/examples/protocols/mqtt/ws/mqtt_ws_example_test.py +++ b/examples/protocols/mqtt/ws/mqtt_ws_example_test.py @@ -5,43 +5,46 @@ import re import os import sys import time -import socket -import imp import paho.mqtt.client as mqtt -g_recv_data="" -g_recv_topic="" -g_broker_connected=0 +try: + import IDF +except Exception: + # this is a test case write with tiny-test-fw. + # to run test cases outside tiny-test-fw, + # we need to set environment variable `TEST_FW_PATH`, + # then get and insert `TEST_FW_PATH` to sys path before import FW module + test_fw_path = os.getenv("TEST_FW_PATH") + if test_fw_path and test_fw_path not in sys.path: + sys.path.insert(0, test_fw_path) + import IDF + +import DUT + + +g_recv_data = "" +g_recv_topic = "" +g_broker_connected = 0 + # The callback for when the client receives a CONNACK response from the server. def on_connect(client, userdata, flags, rc): global g_broker_connected - print("Connected with result code "+str(rc)) + print("Connected with result code " + str(rc)) g_broker_connected = 1 client.subscribe("/topic/qos0") + # The callback for when a PUBLISH message is received from the server. def on_message(client, userdata, msg): global g_recv_topic global g_recv_data payload = msg.payload.decode() - if g_recv_data == "" and payload == "data": + if g_recv_data == "" and payload == "data": client.publish("/topic/qos0", "data_to_esp32") g_recv_topic = msg.topic g_recv_data = payload - print(msg.topic+" "+payload) - -# this is a test case write with tiny-test-fw. -# to run test cases outside tiny-test-fw, -# we need to set environment variable `TEST_FW_PATH`, -# then get and insert `TEST_FW_PATH` to sys path before import FW module -test_fw_path = os.getenv("TEST_FW_PATH") -if test_fw_path and test_fw_path not in sys.path: - sys.path.insert(0, test_fw_path) - -import TinyFW -import IDF -import DUT + print(msg.topic + " " + payload) @IDF.idf_example_test(env_tag="Example_WIFI") @@ -49,7 +52,7 @@ def test_examples_protocol_mqtt_ws(env, extra_data): global g_recv_topic global g_recv_data global g_broker_connected - broker_url="iot.eclipse.org" + broker_url = "iot.eclipse.org" """ steps: | 1. join AP and connects to ws broker @@ -61,8 +64,8 @@ def test_examples_protocol_mqtt_ws(env, extra_data): # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "mqtt_websocket.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("mqtt_websocket_bin_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("mqtt_websocket_size", bin_size//1024) + IDF.log_performance("mqtt_websocket_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("mqtt_websocket_size", bin_size // 1024) # 1. start test (and check the environment is healthy) dut1.start_app() client = None @@ -79,7 +82,7 @@ def test_examples_protocol_mqtt_ws(env, extra_data): print("...done") except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - except: + except Exception: print("ENV_TEST_FAILURE: Unexpected error while connecting to broker {}: {}:".format(broker_url, sys.exc_info()[0])) raise print("Start Looping...") @@ -90,7 +93,7 @@ def test_examples_protocol_mqtt_ws(env, extra_data): if g_broker_connected == 0: raise ValueError('ENV_TEST_FAILURE: Test script cannot connect to broker: {}'.format(broker_url)) # 3. check the message received back from the server - if g_recv_topic == "/topic/qos0" and g_recv_data == "data" : + if g_recv_topic == "/topic/qos0" and g_recv_data == "data": print("PASS: Received correct message") else: print("Failure!") @@ -98,5 +101,6 @@ def test_examples_protocol_mqtt_ws(env, extra_data): # 4. check that the esp32 client received data sent by this python client dut1.expect(re.compile(r"DATA=data_to_esp32"), timeout=30) + if __name__ == '__main__': test_examples_protocol_mqtt_ws() diff --git a/examples/protocols/mqtt/wss/mqtt_wss_example_test.py b/examples/protocols/mqtt/wss/mqtt_wss_example_test.py index ade43c1..b1515ae 100644 --- a/examples/protocols/mqtt/wss/mqtt_wss_example_test.py +++ b/examples/protocols/mqtt/wss/mqtt_wss_example_test.py @@ -3,45 +3,46 @@ import re import os import sys import time -import socket -import imp import ssl import paho.mqtt.client as mqtt -g_recv_data="" -g_recv_topic="" -g_broker_connected=0 +try: + import IDF +except ImportError: + # this is a test case write with tiny-test-fw. + # to run test cases outside tiny-test-fw, + # we need to set environment variable `TEST_FW_PATH`, + # then get and insert `TEST_FW_PATH` to sys path before import FW module + test_fw_path = os.getenv("TEST_FW_PATH") + if test_fw_path and test_fw_path not in sys.path: + sys.path.insert(0, test_fw_path) + import IDF + +import DUT + +g_recv_data = "" +g_recv_topic = "" +g_broker_connected = 0 + # The callback for when the client receives a CONNACK response from the server. def on_connect(client, userdata, flags, rc): global g_broker_connected - print("Connected with result code "+str(rc)) + print("Connected with result code " + str(rc)) g_broker_connected = 1 client.subscribe("/topic/qos0") + # The callback for when a PUBLISH message is received from the server. def on_message(client, userdata, msg): global g_recv_topic global g_recv_data payload = msg.payload.decode() - if g_recv_data == "" and payload == "data": + if g_recv_data == "" and payload == "data": client.publish("/topic/qos0", "data_to_esp32") g_recv_topic = msg.topic g_recv_data = payload - print(msg.topic+" "+str(payload)) - -# this is a test case write with tiny-test-fw. -# to run test cases outside tiny-test-fw, -# we need to set environment variable `TEST_FW_PATH`, -# then get and insert `TEST_FW_PATH` to sys path before import FW module -test_fw_path = os.getenv("TEST_FW_PATH") -if test_fw_path and test_fw_path not in sys.path: - sys.path.insert(0, test_fw_path) - -import TinyFW -import IDF -import DUT - + print(msg.topic + " " + str(payload)) @IDF.idf_example_test(env_tag="Example_WIFI") @@ -49,7 +50,7 @@ def test_examples_protocol_mqtt_wss(env, extra_data): global g_recv_topic global g_recv_data global g_broker_connected - broker_url="iot.eclipse.org" + broker_url = "iot.eclipse.org" """ steps: | 1. join AP and connects to wss broker @@ -61,8 +62,8 @@ def test_examples_protocol_mqtt_wss(env, extra_data): # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "mqtt_websocket_secure.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("mqtt_websocket_secure_bin_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("mqtt_websocket_secure_size", bin_size//1024) + IDF.log_performance("mqtt_websocket_secure_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("mqtt_websocket_secure_size", bin_size // 1024) # 1. start test (and check the environment is healthy) dut1.start_app() client = None @@ -74,14 +75,14 @@ def test_examples_protocol_mqtt_wss(env, extra_data): client.on_connect = on_connect client.on_message = on_message client.tls_set(None, - None, - None, cert_reqs=ssl.CERT_NONE, tls_version=ssl.PROTOCOL_TLSv1, ciphers=None) + None, + None, cert_reqs=ssl.CERT_NONE, tls_version=ssl.PROTOCOL_TLSv1, ciphers=None) print("Connecting...") client.connect(broker_url, 443, 60) print("...done") except DUT.ExpectTimeout: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - except: + except Exception: print("ENV_TEST_FAILURE: Unexpected error while connecting to broker {}: {}:".format(broker_url, sys.exc_info()[0])) raise print("Start Looping...") @@ -92,7 +93,7 @@ def test_examples_protocol_mqtt_wss(env, extra_data): if g_broker_connected == 0: raise ValueError('ENV_TEST_FAILURE: Test script cannot connect to broker: {}'.format(broker_url)) # 3. check the message received back from the server - if g_recv_topic == "/topic/qos0" and g_recv_data == "data" : + if g_recv_topic == "/topic/qos0" and g_recv_data == "data": print("PASS: Received correct message") else: print("Failure!") @@ -100,5 +101,6 @@ def test_examples_protocol_mqtt_wss(env, extra_data): # 4. check that the esp32 client received data sent by this python client dut1.expect(re.compile(r"DATA=data_to_esp32"), timeout=30) + if __name__ == '__main__': test_examples_protocol_mqtt_wss()