diff --git a/examples/protocols/asio/chat_client/asio_chat_client_test.py b/examples/protocols/asio/chat_client/asio_chat_client_test.py index 37018ed37..3b141c7d1 100644 --- a/examples/protocols/asio/chat_client/asio_chat_client_test.py +++ b/examples/protocols/asio/chat_client/asio_chat_client_test.py @@ -1,49 +1,52 @@ import re import os import sys -from socket import * +import socket from threading import Thread import time -# 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) +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 TinyFW -import IDF - -global g_client_response; -global g_msg_to_client; +global g_client_response +global g_msg_to_client g_client_response = b"" g_msg_to_client = b" 3XYZ" + 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 chat_server_sketch(my_ip): global g_client_response print("Starting the server on {}".format(my_ip)) - port=2222 - s=socket(AF_INET, SOCK_STREAM) + port = 2222 + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(600) s.bind((my_ip, port)) s.listen(1) - q,addr=s.accept() + q,addr = s.accept() print("connection accepted") q.settimeout(30) q.send(g_msg_to_client) data = q.recv(1024) # check if received initial empty message - if (len(data)>4): + if (len(data) > 4): g_client_response = data else: g_client_response = q.recv(1024) @@ -51,6 +54,7 @@ def chat_server_sketch(my_ip): s.close() print("server closed") + @IDF.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_asio_chat_client(env, extra_data): """ @@ -64,24 +68,24 @@ def test_examples_protocol_asio_chat_client(env, extra_data): """ global g_client_response global g_msg_to_client - test_msg="ABC" + test_msg = "ABC" dut1 = env.get_dut("chat_client", "examples/protocols/asio/chat_client") # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "asio_chat_client.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("asio_chat_client_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("asio_chat_client_size", bin_size//1024) + IDF.log_performance("asio_chat_client_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("asio_chat_client_size", bin_size // 1024) # 1. start a tcp server on the host host_ip = get_my_ip() - thread1 = Thread(target = chat_server_sketch, args = (host_ip,)) + thread1 = Thread(target=chat_server_sketch, args=(host_ip,)) thread1.start() # 2. start the dut test and wait till client gets IP address dut1.start_app() - data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) + dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) # 3. send host's IP to the client i.e. the `dut1` dut1.write(host_ip) # 4. client `dut1` should receive a message - dut1.expect(g_msg_to_client[4:].decode()) # Strip out the front 4 bytes of message len (see chat_message protocol) + dut1.expect(g_msg_to_client[4:].decode()) # Strip out the front 4 bytes of message len (see chat_message protocol) # 5. write test message from `dut1` chat_client to the server dut1.write(test_msg) while len(g_client_response) == 0: @@ -97,5 +101,6 @@ def test_examples_protocol_asio_chat_client(env, extra_data): raise ValueError('Wrong data received from asi tcp server: {} (expected:{})'.format(g_client_response[4:7], test_msg)) thread1.join() + if __name__ == '__main__': test_examples_protocol_asio_chat_client() diff --git a/examples/protocols/asio/chat_server/asio_chat_server_test.py b/examples/protocols/asio/chat_server/asio_chat_server_test.py index 7f25dc512..4f28116aa 100644 --- a/examples/protocols/asio/chat_server/asio_chat_server_test.py +++ b/examples/protocols/asio/chat_server/asio_chat_server_test.py @@ -1,21 +1,20 @@ import re import os import sys -from socket import * - - -# 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 socket +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 @IDF.idf_example_test(env_tag="Example_WIFI") @@ -27,19 +26,19 @@ def test_examples_protocol_asio_chat_server(env, extra_data): 3. Test connects to server and sends a test message 4. Test evaluates received test message from server """ - test_msg=b" 4ABC\n" + test_msg = b" 4ABC\n" dut1 = env.get_dut("chat_server", "examples/protocols/asio/chat_server") # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "asio_chat_server.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("asio_chat_server_bin_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("asio_chat_server_size", bin_size//1024) + IDF.log_performance("asio_chat_server_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("asio_chat_server_size", bin_size // 1024) # 1. start test dut1.start_app() # 2. get the server IP address data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) # 3. create tcp client and connect to server - cli = socket(AF_INET,SOCK_STREAM) + cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM) cli.settimeout(30) cli.connect((data[0],80)) cli.send(test_msg) diff --git a/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py b/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py index 840060d4c..c29b8ce7b 100644 --- a/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py +++ b/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py @@ -1,21 +1,21 @@ import re import os import sys -from socket import * +import socket -# 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 - +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 @IDF.idf_example_test(env_tag="Example_WIFI") @@ -28,19 +28,19 @@ def test_examples_protocol_asio_tcp_server(env, extra_data): 4. Test evaluates received test message from server 5. Test evaluates received test message on server stdout """ - test_msg=b"echo message from client to server" + test_msg = b"echo message from client to server" dut1 = env.get_dut("tcp_echo_server", "examples/protocols/asio/tcp_echo_server") # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "asio_tcp_echo_server.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("asio_tcp_echo_server_bin_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("asio_tcp_echo_server_size", bin_size//1024) + IDF.log_performance("asio_tcp_echo_server_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("asio_tcp_echo_server_size", bin_size // 1024) # 1. start test dut1.start_app() # 2. get the server IP address data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) # 3. create tcp client and connect to server - cli = socket(AF_INET,SOCK_STREAM) + cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM) cli.settimeout(30) cli.connect((data[0],80)) cli.send(test_msg) diff --git a/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py b/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py index 5487700ff..d3a01563c 100644 --- a/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py +++ b/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py @@ -1,21 +1,21 @@ import re import os import sys -from socket import * +import socket -# 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 - +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 @IDF.idf_example_test(env_tag="Example_WIFI") @@ -28,19 +28,19 @@ def test_examples_protocol_asio_udp_server(env, extra_data): 4. Test evaluates received test message from server 5. Test evaluates received test message on server stdout """ - test_msg=b"echo message from client to server" + test_msg = b"echo message from client to server" dut1 = env.get_dut("udp_echo_server", "examples/protocols/asio/udp_echo_server") # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "asio_udp_echo_server.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("asio_udp_echo_server_bin_size", "{}KB".format(bin_size//1024)) - IDF.check_performance("asio_udp_echo_server_size", bin_size//1024) + IDF.log_performance("asio_udp_echo_server_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("asio_udp_echo_server_size", bin_size // 1024) # 1. start test dut1.start_app() # 2. get the server IP address data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) # 3. create tcp client and connect to server - cli = socket(AF_INET, SOCK_DGRAM) + cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) cli.settimeout(30) cli.connect((data[0], 80)) cli.send(test_msg) @@ -55,5 +55,6 @@ def test_examples_protocol_asio_udp_server(env, extra_data): # 5. check the client message appears also on server terminal dut1.expect(test_msg.decode()) + if __name__ == '__main__': test_examples_protocol_asio_udp_server()