examples: Fix Python coding style

examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py

@@ -2,15 +2,28 @@ import re
 import os
 import sys
 import time
-import socket
-import imp
 import ssl
 import paho.mqtt.client as mqtt
 
+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
@@ -18,6 +31,7 @@ def on_connect(client, userdata, flags, 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
@@ -29,19 +43,6 @@ def on_message(client, userdata, msg):
         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
-
-
 
 @IDF.idf_example_test(env_tag="Example_WIFI")
 def test_examples_protocol_mqtt_ssl(env, extra_data):
@@ -81,7 +82,7 @@ def test_examples_protocol_mqtt_ssl(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...")
@@ -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()

examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py

@@ -1,33 +1,50 @@
 import re
 import os
 import sys
-from socket import *
+import socket
 from threading import Thread
 import struct
 import time
 
+
+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.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]))
@@ -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):
@@ -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()

examples/protocols/mqtt/ws/mqtt_ws_example_test.py

@@ -5,14 +5,28 @@ import re
 import os
 import sys
 import time
-import socket
-import imp
 import paho.mqtt.client as mqtt
 
+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
@@ -20,6 +34,7 @@ def on_connect(client, userdata, flags, 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
@@ -31,18 +46,6 @@ def on_message(client, userdata, msg):
         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
-
 
 @IDF.idf_example_test(env_tag="Example_WIFI")
 def test_examples_protocol_mqtt_ws(env, extra_data):
@@ -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...")
@@ -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()

examples/protocols/mqtt/wss/mqtt_wss_example_test.py

@@ -3,15 +3,28 @@ import re
 import os
 import sys
 import time
-import socket
-import imp
 import ssl
 import paho.mqtt.client as mqtt
 
+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
@@ -19,6 +32,7 @@ def on_connect(client, userdata, flags, 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
@@ -30,19 +44,6 @@ def on_message(client, userdata, msg):
         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
-
-
 
 @IDF.idf_example_test(env_tag="Example_WIFI")
 def test_examples_protocol_mqtt_wss(env, extra_data):
@@ -81,7 +82,7 @@ def test_examples_protocol_mqtt_wss(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...")
@@ -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()