examples: Fix Python coding style

* Original commit: espressif/esp-idf@57c54f96f1
This commit is contained in:
Roland Dobai
2018-12-04 08:32:48 +01:00
committed by gabsuren
parent a96c890f97
commit 21c0878f0e
4 changed files with 78 additions and 73 deletions

View File

@ -1,49 +1,52 @@
import re import re
import os import os
import sys import sys
from socket import * import socket
from threading import Thread from threading import Thread
import time import time
# this is a test case write with tiny-test-fw. try:
# to run test cases outside tiny-test-fw, import IDF
# we need to set environment variable `TEST_FW_PATH`, except ImportError:
# then get and insert `TEST_FW_PATH` to sys path before import FW module # this is a test case write with tiny-test-fw.
test_fw_path = os.getenv("TEST_FW_PATH") # to run test cases outside tiny-test-fw,
if test_fw_path and test_fw_path not in sys.path: # we need to set environment variable `TEST_FW_PATH`,
sys.path.insert(0, 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 global g_client_response
import IDF global g_msg_to_client
global g_client_response;
global g_msg_to_client;
g_client_response = b"" g_client_response = b""
g_msg_to_client = b" 3XYZ" g_msg_to_client = b" 3XYZ"
def get_my_ip(): 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)) s1.connect(("8.8.8.8", 80))
my_ip = s1.getsockname()[0] my_ip = s1.getsockname()[0]
s1.close() s1.close()
return my_ip return my_ip
def chat_server_sketch(my_ip): def chat_server_sketch(my_ip):
global g_client_response global g_client_response
print("Starting the server on {}".format(my_ip)) print("Starting the server on {}".format(my_ip))
port=2222 port = 2222
s=socket(AF_INET, SOCK_STREAM) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(600) s.settimeout(600)
s.bind((my_ip, port)) s.bind((my_ip, port))
s.listen(1) s.listen(1)
q,addr=s.accept() q,addr = s.accept()
print("connection accepted") print("connection accepted")
q.settimeout(30) q.settimeout(30)
q.send(g_msg_to_client) q.send(g_msg_to_client)
data = q.recv(1024) data = q.recv(1024)
# check if received initial empty message # check if received initial empty message
if (len(data)>4): if (len(data) > 4):
g_client_response = data g_client_response = data
else: else:
g_client_response = q.recv(1024) g_client_response = q.recv(1024)
@ -51,6 +54,7 @@ def chat_server_sketch(my_ip):
s.close() s.close()
print("server closed") print("server closed")
@IDF.idf_example_test(env_tag="Example_WIFI") @IDF.idf_example_test(env_tag="Example_WIFI")
def test_examples_protocol_asio_chat_client(env, extra_data): 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_client_response
global g_msg_to_client global g_msg_to_client
test_msg="ABC" test_msg = "ABC"
dut1 = env.get_dut("chat_client", "examples/protocols/asio/chat_client") dut1 = env.get_dut("chat_client", "examples/protocols/asio/chat_client")
# check and log bin size # check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "asio_chat_client.bin") binary_file = os.path.join(dut1.app.binary_path, "asio_chat_client.bin")
bin_size = os.path.getsize(binary_file) bin_size = os.path.getsize(binary_file)
IDF.log_performance("asio_chat_client_size", "{}KB".format(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) IDF.check_performance("asio_chat_client_size", bin_size // 1024)
# 1. start a tcp server on the host # 1. start a tcp server on the host
host_ip = get_my_ip() 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() thread1.start()
# 2. start the dut test and wait till client gets IP address # 2. start the dut test and wait till client gets IP address
dut1.start_app() 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` # 3. send host's IP to the client i.e. the `dut1`
dut1.write(host_ip) dut1.write(host_ip)
# 4. client `dut1` should receive a message # 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 # 5. write test message from `dut1` chat_client to the server
dut1.write(test_msg) dut1.write(test_msg)
while len(g_client_response) == 0: 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)) raise ValueError('Wrong data received from asi tcp server: {} (expected:{})'.format(g_client_response[4:7], test_msg))
thread1.join() thread1.join()
if __name__ == '__main__': if __name__ == '__main__':
test_examples_protocol_asio_chat_client() test_examples_protocol_asio_chat_client()

View File

@ -1,21 +1,20 @@
import re import re
import os import os
import sys 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") @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 3. Test connects to server and sends a test message
4. Test evaluates received test message from server 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") dut1 = env.get_dut("chat_server", "examples/protocols/asio/chat_server")
# check and log bin size # check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "asio_chat_server.bin") binary_file = os.path.join(dut1.app.binary_path, "asio_chat_server.bin")
bin_size = os.path.getsize(binary_file) bin_size = os.path.getsize(binary_file)
IDF.log_performance("asio_chat_server_bin_size", "{}KB".format(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) IDF.check_performance("asio_chat_server_size", bin_size // 1024)
# 1. start test # 1. start test
dut1.start_app() dut1.start_app()
# 2. get the server IP address # 2. get the server IP address
data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
# 3. create tcp client and connect to server # 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.settimeout(30)
cli.connect((data[0],80)) cli.connect((data[0],80))
cli.send(test_msg) cli.send(test_msg)

View File

@ -1,21 +1,21 @@
import re import re
import os import os
import sys import sys
from socket import * import socket
# this is a test case write with tiny-test-fw. try:
# to run test cases outside tiny-test-fw, import IDF
# we need to set environment variable `TEST_FW_PATH`, except ImportError:
# then get and insert `TEST_FW_PATH` to sys path before import FW module # this is a test case write with tiny-test-fw.
test_fw_path = os.getenv("TEST_FW_PATH") # to run test cases outside tiny-test-fw,
if test_fw_path and test_fw_path not in sys.path: # we need to set environment variable `TEST_FW_PATH`,
sys.path.insert(0, 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")
import TinyFW if test_fw_path and test_fw_path not in sys.path:
import IDF sys.path.insert(0, test_fw_path)
import IDF
@IDF.idf_example_test(env_tag="Example_WIFI") @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 4. Test evaluates received test message from server
5. Test evaluates received test message on server stdout 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") dut1 = env.get_dut("tcp_echo_server", "examples/protocols/asio/tcp_echo_server")
# check and log bin size # check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "asio_tcp_echo_server.bin") binary_file = os.path.join(dut1.app.binary_path, "asio_tcp_echo_server.bin")
bin_size = os.path.getsize(binary_file) bin_size = os.path.getsize(binary_file)
IDF.log_performance("asio_tcp_echo_server_bin_size", "{}KB".format(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) IDF.check_performance("asio_tcp_echo_server_size", bin_size // 1024)
# 1. start test # 1. start test
dut1.start_app() dut1.start_app()
# 2. get the server IP address # 2. get the server IP address
data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
# 3. create tcp client and connect to server # 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.settimeout(30)
cli.connect((data[0],80)) cli.connect((data[0],80))
cli.send(test_msg) cli.send(test_msg)

View File

@ -1,21 +1,21 @@
import re import re
import os import os
import sys import sys
from socket import * import socket
# this is a test case write with tiny-test-fw. try:
# to run test cases outside tiny-test-fw, import IDF
# we need to set environment variable `TEST_FW_PATH`, except ImportError:
# then get and insert `TEST_FW_PATH` to sys path before import FW module # this is a test case write with tiny-test-fw.
test_fw_path = os.getenv("TEST_FW_PATH") # to run test cases outside tiny-test-fw,
if test_fw_path and test_fw_path not in sys.path: # we need to set environment variable `TEST_FW_PATH`,
sys.path.insert(0, 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")
import TinyFW if test_fw_path and test_fw_path not in sys.path:
import IDF sys.path.insert(0, test_fw_path)
import IDF
@IDF.idf_example_test(env_tag="Example_WIFI") @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 4. Test evaluates received test message from server
5. Test evaluates received test message on server stdout 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") dut1 = env.get_dut("udp_echo_server", "examples/protocols/asio/udp_echo_server")
# check and log bin size # check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "asio_udp_echo_server.bin") binary_file = os.path.join(dut1.app.binary_path, "asio_udp_echo_server.bin")
bin_size = os.path.getsize(binary_file) bin_size = os.path.getsize(binary_file)
IDF.log_performance("asio_udp_echo_server_bin_size", "{}KB".format(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) IDF.check_performance("asio_udp_echo_server_size", bin_size // 1024)
# 1. start test # 1. start test
dut1.start_app() dut1.start_app()
# 2. get the server IP address # 2. get the server IP address
data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) data = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
# 3. create tcp client and connect to server # 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.settimeout(30)
cli.connect((data[0], 80)) cli.connect((data[0], 80))
cli.send(test_msg) 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 # 5. check the client message appears also on server terminal
dut1.expect(test_msg.decode()) dut1.expect(test_msg.decode())
if __name__ == '__main__': if __name__ == '__main__':
test_examples_protocol_asio_udp_server() test_examples_protocol_asio_udp_server()