espressif/esp-protocols
1
0
Fork 1
mirror of https://github.com/espressif/esp-protocols.git synced 2025-07-17 12:32:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

style: format python files with isort and double-quote-string-fixer

Browse Source 
* Original commit: espressif/esp-idf@0146f258d7
This commit is contained in:
Fu Hanxi
2021-01-26 10:49:01 +08:00
committed by gabsuren
parent 88d3eda156
commit 0d1d73eb35
5 changed files with 45 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
examples/protocols/asio/chat_client/asio_chat_client_test.py
View File

@ -1,21 +1,21 @@
import re
import os import os
import re
import socket import socket
from threading import Thread
import time import time
from threading import Thread
import ttfw_idf import ttfw_idf
global g_client_response global g_client_response
global g_msg_to_client 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.socket(socket.AF_INET, socket.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
@ -23,14 +23,14 @@ def get_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.socket(socket.AF_INET, socket.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)
@ -39,12 +39,12 @@ def chat_server_sketch(my_ip):
g_client_response = data g_client_response = data
else: else:
g_client_response = q.recv(1024) g_client_response = q.recv(1024)
print("received from client {}".format(g_client_response)) print('received from client {}'.format(g_client_response))
s.close() s.close()
print("server closed") print('server closed')
@ttfw_idf.idf_example_test(env_tag="Example_WIFI") @ttfw_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):
""" """
steps: | steps: |
@ -57,19 +57,19 @@ 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", dut_class=ttfw_idf.ESP32DUT) dut1 = env.get_dut('chat_client', 'examples/protocols/asio/chat_client', dut_class=ttfw_idf.ESP32DUT)
# 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)
ttfw_idf.log_performance("asio_chat_client_size", "{}KB".format(bin_size // 1024)) ttfw_idf.log_performance('asio_chat_client_size', '{}KB'.format(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()
dut1.expect(re.compile(r" IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"), timeout=30) dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), 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
@ -82,10 +82,10 @@ def test_examples_protocol_asio_chat_client(env, extra_data):
print(g_client_response) print(g_client_response)
# 6. evaluate host_server received this message # 6. evaluate host_server received this message
if (g_client_response[4:7] == test_msg): if (g_client_response[4:7] == test_msg):
print("PASS: Received correct message") print('PASS: Received correct message')
pass pass
else: else:
print("Failure!") print('Failure!')
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()

18
examples/protocols/asio/chat_server/asio_chat_server_test.py
View File

@ -1,11 +1,11 @@
import re
import os import os
import re
import socket import socket
import ttfw_idf import ttfw_idf
@ttfw_idf.idf_example_test(env_tag="Example_WIFI") @ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_asio_chat_server(env, extra_data): def test_examples_protocol_asio_chat_server(env, extra_data):
""" """
steps: | steps: |
@ -14,16 +14,16 @@ 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", dut_class=ttfw_idf.ESP32DUT) dut1 = env.get_dut('chat_server', 'examples/protocols/asio/chat_server', dut_class=ttfw_idf.ESP32DUT)
# 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)
ttfw_idf.log_performance("asio_chat_server_bin_size", "{}KB".format(bin_size // 1024)) ttfw_idf.log_performance('asio_chat_server_bin_size', '{}KB'.format(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" IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"), timeout=30) data = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)
# 3. create tcp client and connect to server # 3. create tcp client and connect to server
cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM) cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
cli.settimeout(30) cli.settimeout(30)
@ -32,10 +32,10 @@ def test_examples_protocol_asio_chat_server(env, extra_data):
data = cli.recv(1024) data = cli.recv(1024)
# 4. check the message received back from the server # 4. check the message received back from the server
if (data == test_msg): if (data == test_msg):
print("PASS: Received correct message {}".format(data)) print('PASS: Received correct message {}'.format(data))
pass pass
else: else:
print("Failure!") print('Failure!')
raise ValueError('Wrong data received from asi tcp server: {} (expoected:{})'.format(data, test_msg)) raise ValueError('Wrong data received from asi tcp server: {} (expoected:{})'.format(data, test_msg))

1
examples/protocols/asio/ssl_client_server/example_test.py
View File

@ -1,4 +1,5 @@
from __future__ import unicode_literals from __future__ import unicode_literals
import ttfw_idf import ttfw_idf

18
examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py
View File

@ -1,11 +1,11 @@
import re
import os import os
import re
import socket import socket
import ttfw_idf import ttfw_idf
@ttfw_idf.idf_example_test(env_tag="Example_WIFI") @ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_asio_tcp_server(env, extra_data): def test_examples_protocol_asio_tcp_server(env, extra_data):
""" """
steps: | steps: |
@ -15,16 +15,16 @@ 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", dut_class=ttfw_idf.ESP32DUT) dut1 = env.get_dut('tcp_echo_server', 'examples/protocols/asio/tcp_echo_server', dut_class=ttfw_idf.ESP32DUT)
# 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)
ttfw_idf.log_performance("asio_tcp_echo_server_bin_size", "{}KB".format(bin_size // 1024)) ttfw_idf.log_performance('asio_tcp_echo_server_bin_size', '{}KB'.format(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" IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"), timeout=30) data = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)
# 3. create tcp client and connect to server # 3. create tcp client and connect to server
cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM) cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
cli.settimeout(30) cli.settimeout(30)
@ -33,10 +33,10 @@ def test_examples_protocol_asio_tcp_server(env, extra_data):
data = cli.recv(1024) data = cli.recv(1024)
# 4. check the message received back from the server # 4. check the message received back from the server
if (data == test_msg): if (data == test_msg):
print("PASS: Received correct message") print('PASS: Received correct message')
pass pass
else: else:
print("Failure!") print('Failure!')
raise ValueError('Wrong data received from asi tcp server: {} (expected:{})'.format(data, test_msg)) raise ValueError('Wrong data received from asi tcp server: {} (expected:{})'.format(data, test_msg))
# 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())

18
examples/protocols/asio/udp_echo_server/asio_udp_server_test.py
View File

@ -1,11 +1,11 @@
import re
import os import os
import re
import socket import socket
import ttfw_idf import ttfw_idf
@ttfw_idf.idf_example_test(env_tag="Example_WIFI") @ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_asio_udp_server(env, extra_data): def test_examples_protocol_asio_udp_server(env, extra_data):
""" """
steps: | steps: |
@ -15,16 +15,16 @@ 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", dut_class=ttfw_idf.ESP32DUT) dut1 = env.get_dut('udp_echo_server', 'examples/protocols/asio/udp_echo_server', dut_class=ttfw_idf.ESP32DUT)
# 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)
ttfw_idf.log_performance("asio_udp_echo_server_bin_size", "{}KB".format(bin_size // 1024)) ttfw_idf.log_performance('asio_udp_echo_server_bin_size', '{}KB'.format(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" IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"), timeout=30) data = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)
# 3. create tcp client and connect to server # 3. create tcp client and connect to server
cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
cli.settimeout(30) cli.settimeout(30)
@ -33,10 +33,10 @@ def test_examples_protocol_asio_udp_server(env, extra_data):
data = cli.recv(1024) data = cli.recv(1024)
# 4. check the message received back from the server # 4. check the message received back from the server
if (data == test_msg): if (data == test_msg):
print("PASS: Received correct message") print('PASS: Received correct message')
pass pass
else: else:
print("Failure!") print('Failure!')
raise ValueError('Wrong data received from asio udp server: {} (expected:{})'.format(data, test_msg)) raise ValueError('Wrong data received from asio udp server: {} (expected:{})'.format(data, test_msg))
# 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())