components/expat/expat,
components/unity/unity,
examples/build_system/cmake/import_lib/main/lib/tinyxml2
+ # autogenerated scripts
+ examples/provisioning/custom_config/components/custom_provisioning/python/custom_config_pb2.py,
# temporary list (should be empty)
components/app_update/dump_otadata.py,
components/app_update/gen_empty_partition.py,
components/ulp/esp32ulp_mapgen.py,
components/wifi_provisioning/python/wifi_config_pb2.py,
components/wifi_provisioning/python/wifi_constants_pb2.py,
- examples/peripherals/can/can_alert_and_recovery/example_test.py,
- examples/peripherals/can/can_network/example_test.py,
- examples/peripherals/can/can_self_test/example_test.py,
- examples/peripherals/i2s_adc_dac/tools/generate_audio_file.py,
- examples/peripherals/sdio/sdio_test.py,
- examples/protocols/asio/chat_client/asio_chat_client_test.py,
- examples/protocols/asio/chat_server/asio_chat_server_test.py,
- examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py,
- examples/protocols/asio/udp_echo_server/asio_udp_server_test.py,
- examples/protocols/esp_http_client/esp_http_client_test.py,
- examples/protocols/http_server/advanced_tests/http_server_advanced_test.py,
- examples/protocols/http_server/advanced_tests/scripts/test.py,
- examples/protocols/http_server/persistent_sockets/http_server_persistence_test.py,
- examples/protocols/http_server/persistent_sockets/scripts/adder.py,
- examples/protocols/http_server/simple/http_server_simple_test.py,
- examples/protocols/http_server/simple/scripts/client.py,
- examples/protocols/https_request/example_test.py,
- examples/protocols/mdns/mdns_example_test.py,
- examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py,
- examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py,
- examples/protocols/mqtt/ws/mqtt_ws_example_test.py,
- examples/protocols/mqtt/wss/mqtt_wss_example_test.py,
- examples/protocols/sockets/scripts/tcpclient.py,
- examples/protocols/sockets/scripts/tcpserver.py,
- examples/protocols/sockets/scripts/udpclient.py,
- examples/protocols/sockets/scripts/udpserver.py,
examples/provisioning/ble_prov/ble_prov_test.py,
- examples/provisioning/custom_config/components/custom_provisioning/python/custom_config_pb2.py,
examples/provisioning/softap_prov/softap_prov_test.py,
examples/provisioning/softap_prov/utils/wifi_tools.py,
- examples/storage/parttool/example_test.py,
- examples/system/cpp_exceptions/example_test.py,
- examples/system/esp_event/default_event_loop/example_test.py,
- examples/system/esp_event/user_event_loops/example_test.py,
- examples/system/esp_timer/example_test.py,
- examples/system/light_sleep/example_test.py,
- examples/system/ota/otatool/example_test.py,
- examples/wifi/iperf/iperf_test.py,
- examples/wifi/iperf/test_report.py,
tools/ci/apply_bot_filter.py,
tools/cmake/convert_to_cmake.py,
tools/esp_app_trace/apptrace_proc.py,
-#Need Python 3 string formatting functions
+# Need Python 3 string formatting functions
from __future__ import print_function
-import re
import os
import sys
-# The test cause is dependent on the Tiny Test Framework. Ensure the
-# `TEST_FW_PATH` environment variable is set to `$IDF_PATH/tools/tiny-test-fw`
-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:
+ # The test cause is dependent on the Tiny Test Framework. Ensure the
+ # `TEST_FW_PATH` environment variable is set to `$IDF_PATH/tools/tiny-test-fw`
+ 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
# CAN Self Test Example constants
STR_EXPECT = ("CAN Alert and Recovery: Driver installed", "CAN Alert and Recovery: Driver uninstalled")
EXPECT_TIMEOUT = 20
+
@IDF.idf_example_test(env_tag='Example_CAN')
def test_can_alert_and_recovery_example(env, extra_data):
- #Get device under test, flash and start example. "dut4" must be defined in EnvConfig
+ # Get device under test, flash and start example. "dut4" must be defined in EnvConfig
dut = env.get_dut('dut4', 'examples/peripherals/can/can_alert_and_recovery')
dut.start_app()
for string in STR_EXPECT:
- dut.expect(string, timeout = EXPECT_TIMEOUT)
+ dut.expect(string, timeout=EXPECT_TIMEOUT)
+
if __name__ == '__main__':
test_can_alert_and_recovery_example()
-#Need Python 3 string formatting functions
+# Need Python 3 string formatting functions
from __future__ import print_function
-import re
import os
import sys
-import time
from threading import Thread
-# The test cause is dependent on the Tiny Test Framework. Ensure the
-# `TEST_FW_PATH` environment variable is set to `$IDF_PATH/tools/tiny-test-fw`
-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
-
-#Define tuple of strings to expect for each DUT.
+try:
+ import IDF
+except ImportError:
+ # The test cause is dependent on the Tiny Test Framework. Ensure the
+ # `TEST_FW_PATH` environment variable is set to `$IDF_PATH/tools/tiny-test-fw`
+ 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
+
+# Define tuple of strings to expect for each DUT.
master_expect = ("CAN Master: Driver installed", "CAN Master: Driver uninstalled")
slave_expect = ("CAN Slave: Driver installed", "CAN Slave: Driver uninstalled")
listen_only_expect = ("CAN Listen Only: Driver installed", "Listen Only: Driver uninstalled")
+
def dut_thread_callback(**kwargs):
- #Parse keyword arguments
- dut = kwargs['dut'] #Get DUT from kwargs
+ # Parse keyword arguments
+ dut = kwargs['dut'] # Get DUT from kwargs
expected = kwargs['expected']
- result = kwargs['result'] #Get result[out] from kwargs. MUST be of mutable type e.g. list
-
- #Must reset again as flashing during start_app will reset multiple times, causing unexpected results
+ result = kwargs['result'] # Get result[out] from kwargs. MUST be of mutable type e.g. list
+
+ # Must reset again as flashing during start_app will reset multiple times, causing unexpected results
dut.reset()
for string in expected:
dut.expect(string, 20)
- #Mark thread has run to completion without any exceptions
- result[0] = True
+ # Mark thread has run to completion without any exceptions
+ result[0] = True
+
@IDF.idf_example_test(env_tag='Example_CAN')
def test_can_network_example(env, extra_data):
-
- #Get device under test. "dut1", "dut2", and "dut3" must be properly defined in EnvConfig
+
+ # Get device under test. "dut1", "dut2", and "dut3" must be properly defined in EnvConfig
dut_master = env.get_dut("dut1", "examples/peripherals/can/can_network/can_network_master")
dut_slave = env.get_dut("dut2", "examples/peripherals/can/can_network/can_network_slave")
dut_listen_only = env.get_dut("dut3", "examples/peripherals/can/can_network/can_network_listen_only")
- #Flash app onto each DUT, each DUT is reset again at the start of each thread
+ # Flash app onto each DUT, each DUT is reset again at the start of each thread
dut_master.start_app()
dut_slave.start_app()
dut_listen_only.start_app()
- #Create dict of keyword arguments for each dut
+ # Create dict of keyword arguments for each dut
results = [[False], [False], [False]]
- master_kwargs = {"dut" : dut_master, "result" : results[0], "expected" : master_expect}
- slave_kwargs = {"dut" : dut_slave, "result" : results[1], "expected" : slave_expect}
- listen_only_kwargs = {"dut" : dut_listen_only, "result" : results[2], "expected" : listen_only_expect}
+ master_kwargs = {"dut": dut_master, "result": results[0], "expected": master_expect}
+ slave_kwargs = {"dut": dut_slave, "result": results[1], "expected": slave_expect}
+ listen_only_kwargs = {"dut": dut_listen_only, "result": results[2], "expected": listen_only_expect}
- #Create thread for each dut
- dut_master_thread = Thread(target = dut_thread_callback, name = "Master Thread", kwargs = master_kwargs)
- dut_slave_thread = Thread(target = dut_thread_callback, name = "Slave Thread", kwargs = slave_kwargs)
- dut_listen_only_thread = Thread(target = dut_thread_callback, name = "Listen Only Thread", kwargs = listen_only_kwargs)
+ # Create thread for each dut
+ dut_master_thread = Thread(target=dut_thread_callback, name="Master Thread", kwargs=master_kwargs)
+ dut_slave_thread = Thread(target=dut_thread_callback, name="Slave Thread", kwargs=slave_kwargs)
+ dut_listen_only_thread = Thread(target=dut_thread_callback, name="Listen Only Thread", kwargs=listen_only_kwargs)
- #Start each thread
+ # Start each thread
dut_listen_only_thread.start()
dut_master_thread.start()
dut_slave_thread.start()
- #Wait for threads to complete
+ # Wait for threads to complete
dut_listen_only_thread.join()
dut_master_thread.join()
dut_slave_thread.join()
- #check each thread ran to completion
+ # check each thread ran to completion
for result in results:
- if result[0] != True:
+ if result[0] is not True:
raise Exception("One or more threads did not run successfully")
+
if __name__ == '__main__':
test_can_network_example()
-#Need Python 3 string formatting functions
+# Need Python 3 string formatting functions
from __future__ import print_function
-import re
import os
import sys
-# The test cause is dependent on the Tiny Test Framework. Ensure the
-# `TEST_FW_PATH` environment variable is set to `$IDF_PATH/tools/tiny-test-fw`
-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:
+ # The test cause is dependent on the Tiny Test Framework. Ensure the
+ # `TEST_FW_PATH` environment variable is set to `$IDF_PATH/tools/tiny-test-fw`
+ 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
+
# CAN Self Test Example constants
STR_EXPECT = ("CAN Self Test: Driver installed", "CAN Self Test: Driver uninstalled")
EXPECT_TIMEOUT = 20
+
@IDF.idf_example_test(env_tag='Example_CAN')
def test_can_self_test_example(env, extra_data):
- #Get device under test, flash and start example. "dut4" must be defined in EnvConfig
+ # Get device under test, flash and start example. "dut4" must be defined in EnvConfig
dut = env.get_dut('dut4', 'examples/peripherals/can/can_self_test')
dut.start_app()
for string in STR_EXPECT:
- dut.expect(string, timeout = EXPECT_TIMEOUT)
+ dut.expect(string, timeout=EXPECT_TIMEOUT)
+
if __name__ == '__main__':
test_can_self_test_example()
import wave
import struct
+
def get_wave_array_str(filename, target_bits):
wave_read = wave.open(filename, "r")
array_str = ""
val, = struct.unpack("<H", wave_read.readframes(1))
scale_val = (1 << target_bits) - 1
cur_lim = (1 << sampwidth) - 1
- #scale current data to 8-bit data
+ # scale current data to 8-bit data
val = val * scale_val / cur_lim
val = int(val + ((scale_val + 1) // 2)) & scale_val
- array_str += "0x%x, "%(val)
+ array_str += "0x%x, " % (val)
if (i + 1) % 16 == 0:
array_str += "\n"
return array_str
-def gen_wave_table(wav_file_list, target_file_name, scale_bits = 8):
+
+def gen_wave_table(wav_file_list, target_file_name, scale_bits=8):
with open(target_file_name, "w") as audio_table:
print('#include <stdio.h>', file=audio_table)
print('const unsigned char audio_table[] = {', file=audio_table)
for wav in wav_file_list:
print("processing: {}".format(wav))
- print(get_wave_array_str(filename = wav, target_bits = scale_bits), file=audio_table)
+ print(get_wave_array_str(filename=wav, target_bits=scale_bits), file=audio_table)
print('};\n', file=audio_table)
print("Done...")
-if __name__=='__main__':
+
+if __name__ == '__main__':
print("Generating audio array...")
wav_list = []
for filename in os.listdir("./"):
if filename.endswith(".wav"):
wav_list.append(filename)
- gen_wave_table(wav_file_list = wav_list, target_file_name = "audio_example_file.h")
+ gen_wave_table(wav_file_list=wav_list, target_file_name="audio_example_file.h")
# limitations under the License.
""" example of writing test with TinyTestFW """
-import re
import os
import sys
-# if we want to run test case outside `tiny-test-fw` folder,
-# we need to insert tiny-test-fw path into sys path
-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 TinyFW
+except ImportError:
+ # if we want to run test case outside `tiny-test-fw` folder,
+ # we need to insert tiny-test-fw path into sys path
+ 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 TinyFW
import IDF
dut1 = env.get_dut("sdio_host", "examples/peripherals/sdio/host")
dut2 = env.get_dut("sdio_slave", "examples/peripherals/sdio/slave")
dut1.start_app()
- #wait until the master is ready to setup the slave
+ # wait until the master is ready to setup the slave
dut1.expect("host ready, start initializing slave...")
dut2.start_app()
dut2.expect("recv len: 6")
dut2.expect("recv len: 12")
dut2.expect("recv len: 128")
- #511
+ # 511
dut2.expect("recv len: 128")
dut2.expect("recv len: 128")
dut2.expect("recv len: 128")
dut2.expect("recv len: 127")
- #512
+ # 512
dut2.expect("recv len: 128")
dut2.expect("recv len: 128")
dut2.expect("recv len: 128")
dut1.expect("receive data, size: 128")
dut1.expect("receive data, size: 128")
- #the last valid line of one round
+ # the last valid line of one round
dut1.expect("ce d3 d8 dd e2 e7 ec f1 f6 fb 00 05 0a 0f 14 19")
- #the first 2 lines of the second round
+ # the first 2 lines of the second round
dut1.expect("46 4b 50")
dut1.expect("5a 5f 64 69 6e 73")
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)
s.close()
print("server closed")
+
@IDF.idf_example_test(env_tag="Example_WIFI")
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:
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()
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")
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)
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")
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)
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")
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)
# 5. check the client message appears also on server terminal
dut1.expect(test_msg.decode())
+
if __name__ == '__main__':
test_examples_protocol_asio_udp_server()
import os
import sys
-# 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 TinyFW
-import IDF
+ import IDF
@IDF.idf_example_test(env_tag="Example_WIFI", ignore=True)
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "esp-http-client-example.bin")
bin_size = os.path.getsize(binary_file)
- IDF.log_performance("esp_http_client_bin_size", "{}KB".format(bin_size//1024))
- IDF.check_performance("esp_http_client_bin_size", bin_size//1024)
+ IDF.log_performance("esp_http_client_bin_size", "{}KB".format(bin_size // 1024))
+ IDF.check_performance("esp_http_client_bin_size", bin_size // 1024)
# start test
dut1.start_app()
dut1.expect("Connected to AP, begin http example", timeout=30)
import re
import os
import sys
-import string
-import random
-import socket
-# This environment variable is expected on the host machine
-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 environment variable is expected on the host machine
+ 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 Utility
# When running on local machine execute the following before running this script
# > make app bootloader
# > make print_flash_cmd | tail -n 1 > build/download.config
# > export TEST_FW_PATH=~/esp/esp-idf/tools/tiny-test-fw
-import TinyFW
-import IDF
-import Utility
-
# Import client module
expath = os.path.dirname(os.path.realpath(__file__))
client = imp.load_source("client", expath + "/scripts/test.py")
+
# Due to connectivity issues (between runner host and DUT) in the runner environment,
# some of the `advanced_tests` are ignored. These tests are intended for verifying
# the expected limits of the http_server capabilities, and implement sending and receiving
# Get binary file
binary_file = os.path.join(dut1.app.binary_path, "tests.bin")
bin_size = os.path.getsize(binary_file)
- IDF.log_performance("http_server_bin_size", "{}KB".format(bin_size//1024))
- IDF.check_performance("http_server_bin_size", bin_size//1024)
+ IDF.log_performance("http_server_bin_size", "{}KB".format(bin_size // 1024))
+ IDF.check_performance("http_server_bin_size", bin_size // 1024)
# Upload binary and start testing
Utility.console_log("Starting http_server advanced test app")
got_ip = dut1.expect(re.compile(r"(?:[\s\S]*)Got IP: '(\d+.\d+.\d+.\d+)'"), timeout=30)[0]
got_port = dut1.expect(re.compile(r"(?:[\s\S]*)Started HTTP server on port: '(\d+)'"), timeout=15)[0]
- result = dut1.expect(re.compile(r"(?:[\s\S]*)Max URI handlers: '(\d+)'(?:[\s\S]*)Max Open Sessions: '(\d+)'(?:[\s\S]*)Max Header Length: '(\d+)'(?:[\s\S]*)Max URI Length: '(\d+)'(?:[\s\S]*)Max Stack Size: '(\d+)'"), timeout=15)
- max_uri_handlers = int(result[0])
+ result = dut1.expect(re.compile(r"(?:[\s\S]*)Max URI handlers: '(\d+)'(?:[\s\S]*)Max Open Sessions: " # noqa: W605
+ r"'(\d+)'(?:[\s\S]*)Max Header Length: '(\d+)'(?:[\s\S]*)Max URI Length: "
+ r"'(\d+)'(?:[\s\S]*)Max Stack Size: '(\d+)'"), timeout=15)
+ # max_uri_handlers = int(result[0])
max_sessions = int(result[1])
max_hdr_len = int(result[2])
max_uri_len = int(result[3])
if not client.recv_timeout_test(got_ip, got_port):
failed = True
- ## This test fails a lot! Enable when connection is stable
- #test_size = 50*1024 # 50KB
- #if not client.packet_size_limit_test(got_ip, got_port, test_size):
+ # This test fails a lot! Enable when connection is stable
+ # test_size = 50*1024 # 50KB
+ # if not client.packet_size_limit_test(got_ip, got_port, test_size):
# Utility.console_log("Ignoring failure")
Utility.console_log("Getting initial stack usage...")
inital_stack = int(dut1.expect(re.compile(r"(?:[\s\S]*)Free Stack for server task: '(\d+)'"), timeout=15)[0])
- if inital_stack < 0.1*max_stack_size:
+ if inital_stack < 0.1 * max_stack_size:
Utility.console_log("More than 90% of stack being used on server start")
failed = True
final_stack = int(dut1.expect(re.compile(r"(?:[\s\S]*)Free Stack for server task: '(\d+)'"), timeout=15)[0])
- if final_stack < 0.05*max_stack_size:
+ if final_stack < 0.05 * max_stack_size:
Utility.console_log("More than 95% of stack got used during tests")
failed = True
if failed:
raise RuntimeError
+
if __name__ == '__main__':
test_examples_protocol_http_server_advanced()
# - Server should automatically close the socket
-############# TODO TESTS #############
+# ############ TODO TESTS #############
# 3. Stress Tests
#
from __future__ import division
from __future__ import print_function
-from future import standard_library
-standard_library.install_aliases()
from builtins import str
from builtins import range
from builtins import object
_verbose_ = False
+
class Session(object):
- def __init__(self, addr, port, timeout = 15):
- self.client = socket.create_connection((addr, int(port)), timeout = timeout)
+ def __init__(self, addr, port, timeout=15):
+ self.client = socket.create_connection((addr, int(port)), timeout=timeout)
self.target = addr
self.status = 0
self.encoding = ''
def send_err_check(self, request, data=None):
rval = True
try:
- self.client.sendall(request.encode());
+ self.client.sendall(request.encode())
if data:
self.client.sendall(data.encode())
except socket.error as err:
request = "GET " + path + " HTTP/1.1\r\nHost: " + self.target
if headers:
for field, value in headers.items():
- request += "\r\n"+field+": "+value
+ request += "\r\n" + field + ": " + value
request += "\r\n\r\n"
return self.send_err_check(request)
def send_put(self, path, data, headers=None):
- request = "PUT " + path + " HTTP/1.1\r\nHost: " + self.target
+ request = "PUT " + path + " HTTP/1.1\r\nHost: " + self.target
if headers:
for field, value in headers.items():
- request += "\r\n"+field+": "+value
- request += "\r\nContent-Length: " + str(len(data)) +"\r\n\r\n"
+ request += "\r\n" + field + ": " + value
+ request += "\r\nContent-Length: " + str(len(data)) + "\r\n\r\n"
return self.send_err_check(request, data)
def send_post(self, path, data, headers=None):
- request = "POST " + path + " HTTP/1.1\r\nHost: " + self.target
+ request = "POST " + path + " HTTP/1.1\r\nHost: " + self.target
if headers:
for field, value in headers.items():
- request += "\r\n"+field+": "+value
- request += "\r\nContent-Length: " + str(len(data)) +"\r\n\r\n"
+ request += "\r\n" + field + ": " + value
+ request += "\r\nContent-Length: " + str(len(data)) + "\r\n\r\n"
return self.send_err_check(request, data)
def read_resp_hdrs(self):
def close(self):
self.client.close()
+
def test_val(text, expected, received):
if expected != received:
Utility.console_log(" Fail!")
return False
return True
+
class adder_thread (threading.Thread):
def __init__(self, id, dut, port):
threading.Thread.__init__(self)
def close(self):
self.session.close()
+
def get_hello(dut, port):
# GET /hello should return 'Hello World!'
Utility.console_log("[test] GET /hello returns 'Hello World!' =>", end=' ')
conn.close()
return True
+
def put_hello(dut, port):
# PUT /hello returns 405'
Utility.console_log("[test] PUT /hello returns 405 =>", end=' ')
conn.close()
return True
+
def post_hello(dut, port):
# POST /hello returns 405'
Utility.console_log("[test] POST /hello returns 404 =>", end=' ')
conn.close()
return True
+
def post_echo(dut, port):
# POST /echo echoes data'
Utility.console_log("[test] POST /echo echoes data =>", end=' ')
conn.close()
return True
+
def put_echo(dut, port):
# PUT /echo echoes data'
Utility.console_log("[test] PUT /echo echoes data =>", end=' ')
conn.close()
return True
+
def get_echo(dut, port):
# GET /echo returns 404'
Utility.console_log("[test] GET /echo returns 405 =>", end=' ')
conn.close()
return True
+
def get_hello_type(dut, port):
# GET /hello/type_html returns text/html as Content-Type'
Utility.console_log("[test] GET /hello/type_html has Content-Type of text/html =>", end=' ')
conn.close()
return True
+
def get_hello_status(dut, port):
# GET /hello/status_500 returns status 500'
Utility.console_log("[test] GET /hello/status_500 returns status 500 =>", end=' ')
conn.close()
return True
+
def get_false_uri(dut, port):
# GET /false_uri returns status 404'
Utility.console_log("[test] GET /false_uri returns status 404 =>", end=' ')
conn.close()
return True
+
def parallel_sessions_adder(dut, port, max_sessions):
# POSTs on /adder in parallel sessions
Utility.console_log("[test] POST {pipelined} on /adder in " + str(max_sessions) + " sessions =>", end=' ')
Utility.console_log("Success")
return res
+
def async_response_test(dut, port):
# Test that an asynchronous work is executed in the HTTPD's context
# This is tested by reading two responses over the same session
Utility.console_log("Success")
return True
+
def leftover_data_test(dut, port):
# Leftover data in POST is purged (valid and invalid URIs)
Utility.console_log("[test] Leftover data in POST is purged (valid and invalid URIs) =>", end=' ')
Utility.console_log("Success")
return True
+
def spillover_session(dut, port, max_sess):
# Session max_sess_sessions + 1 is rejected
Utility.console_log("[test] Session max_sess_sessions (" + str(max_sess) + ") + 1 is rejected =>", end=' ')
a.close()
break
s.append(a)
- except:
+ except Exception:
if (_verbose_):
Utility.console_log("Connection " + str(i) + " rejected")
a.close()
Utility.console_log(["Fail","Success"][len(s) == max_sess])
return (len(s) == max_sess)
+
def recv_timeout_test(dut, port):
Utility.console_log("[test] Timeout occurs if partial packet sent =>", end=' ')
s = Session(dut, port)
Utility.console_log("Success")
return True
+
def packet_size_limit_test(dut, port, test_size):
Utility.console_log("[test] send size limit test =>", end=' ')
retry = 5
retry -= 1
Utility.console_log("data size = ", test_size)
s = http.client.HTTPConnection(dut + ":" + port, timeout=15)
- random_data = ''.join(string.printable[random.randint(0,len(string.printable))-1] for _ in list(range(test_size)))
+ random_data = ''.join(string.printable[random.randint(0,len(string.printable)) - 1] for _ in list(range(test_size)))
path = "/echo"
s.request("POST", url=path, body=random_data)
resp = s.getresponse()
if not test_val("Error", "200", str(resp.status)):
if test_val("Error", "500", str(resp.status)):
Utility.console_log("Data too large to be allocated")
- test_size = test_size//10
+ test_size = test_size // 10
else:
Utility.console_log("Unexpected error")
s.close()
Utility.console_log("Failed")
return False
+
def code_500_server_error_test(dut, port):
Utility.console_log("[test] 500 Server Error test =>", end=' ')
s = Session(dut, port)
content_len = 2**31
s.client.sendall(("POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: " + str(content_len) + "\r\n\r\nABCD").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
if not test_val("Server Error", "500", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def code_501_method_not_impl(dut, port):
Utility.console_log("[test] 501 Method Not Implemented =>", end=' ')
s = Session(dut, port)
path = "/hello"
s.client.sendall(("ABC " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
# Presently server sends back 400 Bad Request
- #if not test_val("Server Error", "501", s.status):
- #s.close()
- #return False
+ # if not test_val("Server Error", "501", s.status):
+ # s.close()
+ # return False
if not test_val("Server Error", "400", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def code_505_version_not_supported(dut, port):
Utility.console_log("[test] 505 Version Not Supported =>", end=' ')
s = Session(dut, port)
path = "/hello"
s.client.sendall(("GET " + path + " HTTP/2.0\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
if not test_val("Server Error", "505", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def code_400_bad_request(dut, port):
Utility.console_log("[test] 400 Bad Request =>", end=' ')
s = Session(dut, port)
path = "/hello"
s.client.sendall(("XYZ " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
if not test_val("Client Error", "400", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def code_404_not_found(dut, port):
Utility.console_log("[test] 404 Not Found =>", end=' ')
s = Session(dut, port)
path = "/dummy"
s.client.sendall(("GET " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
if not test_val("Client Error", "404", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def code_405_method_not_allowed(dut, port):
Utility.console_log("[test] 405 Method Not Allowed =>", end=' ')
s = Session(dut, port)
path = "/hello"
s.client.sendall(("POST " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
if not test_val("Client Error", "405", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def code_408_req_timeout(dut, port):
Utility.console_log("[test] 408 Request Timeout =>", end=' ')
s = Session(dut, port)
s.client.sendall(("POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: 10\r\n\r\nABCD").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
if not test_val("Client Error", "408", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def code_411_length_required(dut, port):
Utility.console_log("[test] 411 Length Required =>", end=' ')
s = Session(dut, port)
path = "/echo"
s.client.sendall(("POST " + path + " HTTP/1.1\r\nHost: " + dut + "\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
# Presently server sends back 400 Bad Request
- #if not test_val("Client Error", "411", s.status):
- #s.close()
- #return False
+ # if not test_val("Client Error", "411", s.status):
+ # s.close()
+ # return False
if not test_val("Client Error", "400", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
def send_getx_uri_len(dut, port, length):
s = Session(dut, port)
method = "GET "
version = " HTTP/1.1\r\n"
- path = "/"+"x"*(length - len(method) - len(version) - len("/"))
+ path = "/" + "x" * (length - len(method) - len(version) - len("/"))
s.client.sendall(method.encode())
time.sleep(1)
s.client.sendall(path.encode())
time.sleep(1)
s.client.sendall((version + "Host: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
s.close()
return s.status
+
def code_414_uri_too_long(dut, port, max_uri_len):
Utility.console_log("[test] 414 URI Too Long =>", end=' ')
status = send_getx_uri_len(dut, port, max_uri_len)
Utility.console_log("Success")
return True
+
def send_postx_hdr_len(dut, port, length):
s = Session(dut, port)
path = "/echo"
host = "Host: " + dut
custom_hdr_field = "\r\nCustom: "
- custom_hdr_val = "x"*(length - len(host) - len(custom_hdr_field) - len("\r\n\r\n") + len("0"))
+ custom_hdr_val = "x" * (length - len(host) - len(custom_hdr_field) - len("\r\n\r\n") + len("0"))
request = ("POST " + path + " HTTP/1.1\r\n" + host + custom_hdr_field + custom_hdr_val + "\r\n\r\n").encode()
- s.client.sendall(request[:length//2])
+ s.client.sendall(request[:length // 2])
time.sleep(1)
- s.client.sendall(request[length//2:])
+ s.client.sendall(request[length // 2:])
hdr = s.read_resp_hdrs()
resp = s.read_resp_data()
s.close()
return (hdr["Custom"] == custom_hdr_val), resp
return False, s.status
+
def code_431_hdr_too_long(dut, port, max_hdr_len):
Utility.console_log("[test] 431 Header Too Long =>", end=' ')
res, status = send_postx_hdr_len(dut, port, max_hdr_len)
Utility.console_log("Success")
return True
+
def test_upgrade_not_supported(dut, port):
Utility.console_log("[test] Upgrade Not Supported =>", end=' ')
s = Session(dut, port)
- path = "/hello"
+ # path = "/hello"
s.client.sendall(("OPTIONS * HTTP/1.1\r\nHost:" + dut + "\r\nUpgrade: TLS/1.0\r\nConnection: Upgrade\r\n\r\n").encode())
s.read_resp_hdrs()
- resp = s.read_resp_data()
+ s.read_resp_data()
if not test_val("Client Error", "200", s.status):
s.close()
return False
Utility.console_log("Success")
return True
+
if __name__ == '__main__':
- ########### Execution begins here...
+ # Execution begins here...
# Configuration
# Max number of threads/sessions
max_sessions = 7
test_upgrade_not_supported(dut, port)
# Not supported yet (Error on chunked request)
- ###code_411_length_required(dut, port)
+ # code_411_length_required(dut, port)
Utility.console_log("### Sessions and Context Tests")
parallel_sessions_adder(dut, port, max_sessions)
async_response_test(dut, port)
spillover_session(dut, port, max_sessions)
recv_timeout_test(dut, port)
- packet_size_limit_test(dut, port, 50*1024)
+ packet_size_limit_test(dut, port, 50 * 1024)
get_hello(dut, port)
sys.exit()
import re
import os
import sys
-import string
import random
-import socket
-# This environment variable is expected on the host machine
-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 environment variable is expected on the host machine
+ 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 Utility
# When running on local machine execute the following before running this script
# > make app bootloader
# > make print_flash_cmd | tail -n 1 > build/download.config
# > export TEST_FW_PATH=~/esp/esp-idf/tools/tiny-test-fw
-import TinyFW
-import IDF
-import Utility
-
# Import client module
expath = os.path.dirname(os.path.realpath(__file__))
client = imp.load_source("client", expath + "/scripts/adder.py")
+
@IDF.idf_example_test(env_tag="Example_WIFI")
def test_examples_protocol_http_server_persistence(env, extra_data):
# Acquire DUT
# Get binary file
binary_file = os.path.join(dut1.app.binary_path, "persistent_sockets.bin")
bin_size = os.path.getsize(binary_file)
- IDF.log_performance("http_server_bin_size", "{}KB".format(bin_size//1024))
- IDF.check_performance("http_server_bin_size", bin_size//1024)
+ IDF.log_performance("http_server_bin_size", "{}KB".format(bin_size // 1024))
+ IDF.check_performance("http_server_bin_size", bin_size // 1024)
# Upload binary and start testing
Utility.console_log("Starting http_server persistance test app")
dut1.expect("PUT allocating new session", timeout=30)
# Not expected
raise RuntimeError
- except:
+ except Exception:
# As expected
pass
Utility.console_log("Validating user context data")
# Start another session to check user context data
- conn2 = client.start_session(got_ip, got_port)
+ client.start_session(got_ip, got_port)
num = random.randint(0,100)
client.putreq(conn, "/adder", str(num))
visitor += 1
client.end_session(conn)
dut1.expect("/adder Free Context function called", timeout=30)
+
if __name__ == '__main__':
test_examples_protocol_http_server_persistence()
from __future__ import print_function
from __future__ import unicode_literals
-from future import standard_library
-standard_library.install_aliases()
from builtins import str
from builtins import range
import http.client
import argparse
+import Utility
-def start_session (ip, port):
+
+def start_session(ip, port):
return http.client.HTTPConnection(ip, int(port), timeout=15)
-def end_session (conn):
+
+def end_session(conn):
conn.close()
-def getreq (conn, path, verbose = False):
+
+def getreq(conn, path, verbose=False):
conn.request("GET", path)
resp = conn.getresponse()
data = resp.read()
Utility.console_log("Data content : " + data)
return data
-def postreq (conn, path, data, verbose = False):
+
+def postreq(conn, path, data, verbose=False):
conn.request("POST", path, data)
resp = conn.getresponse()
data = resp.read()
Utility.console_log("Data content : " + data)
return data
-def putreq (conn, path, body, verbose = False):
+
+def putreq(conn, path, body, verbose=False):
conn.request("PUT", path, body)
resp = conn.getresponse()
data = resp.read()
Utility.console_log("Data content : " + data)
return data
+
if __name__ == '__main__':
# Configure argument parser
parser = argparse.ArgumentParser(description='Run HTTPd Test')
- parser.add_argument('IP' , metavar='IP' , type=str, help='Server IP')
+ parser.add_argument('IP', metavar='IP', type=str, help='Server IP')
parser.add_argument('port', metavar='port', type=str, help='Server port')
- parser.add_argument('N' , metavar='integer', type=int, help='Integer to sum upto')
+ parser.add_argument('N', metavar='integer', type=int, help='Integer to sum upto')
args = vars(parser.parse_args())
# Get arguments
# Establish HTTP connection
Utility.console_log("Connecting to => " + ip + ":" + port)
- conn = start_session (ip, port)
+ conn = start_session(ip, port)
# Reset adder context to specified value(0)
# -- Not needed as new connection will always
# -- have zero value of the accumulator
Utility.console_log("Reset the accumulator to 0")
- putreq (conn, "/adder", str(0))
+ putreq(conn, "/adder", str(0))
# Sum numbers from 1 to specified value(N)
Utility.console_log("Summing numbers from 1 to " + str(N))
- for i in range(1, N+1):
- postreq (conn, "/adder", str(i))
+ for i in range(1, N + 1):
+ postreq(conn, "/adder", str(i))
# Fetch the result
- Utility.console_log("Result :" + getreq (conn, "/adder"))
+ Utility.console_log("Result :" + getreq(conn, "/adder"))
# Close HTTP connection
- end_session (conn)
+ end_session(conn)
import sys
import string
import random
-import socket
-# This environment variable is expected on the host machine
-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 environment variable is expected on the host machine
+ 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 Utility
# When running on local machine execute the following before running this script
# > make app bootloader
# > make print_flash_cmd | tail -n 1 > build/download.config
# > export TEST_FW_PATH=~/esp/esp-idf/tools/tiny-test-fw
-import TinyFW
-import IDF
-import Utility
-
# Import client module
expath = os.path.dirname(os.path.realpath(__file__))
client = imp.load_source("client", expath + "/scripts/client.py")
+
@IDF.idf_example_test(env_tag="Example_WIFI")
def test_examples_protocol_http_server_simple(env, extra_data):
# Acquire DUT
# Get binary file
binary_file = os.path.join(dut1.app.binary_path, "simple.bin")
bin_size = os.path.getsize(binary_file)
- IDF.log_performance("http_server_bin_size", "{}KB".format(bin_size//1024))
- IDF.check_performance("http_server_bin_size", bin_size//1024)
+ IDF.log_performance("http_server_bin_size", "{}KB".format(bin_size // 1024))
+ IDF.check_performance("http_server_bin_size", bin_size // 1024)
# Upload binary and start testing
Utility.console_log("Starting http_server simple test app")
raise RuntimeError
# Acquire host IP. Need a way to check it
- host_ip = dut1.expect(re.compile(r"(?:[\s\S]*)Found header => Host: (\d+.\d+.\d+.\d+)"), timeout=30)[0]
+ dut1.expect(re.compile(r"(?:[\s\S]*)Found header => Host: (\d+.\d+.\d+.\d+)"), timeout=30)[0]
# Match additional headers sent in the request
dut1.expect("Found header => Test-Header-2: Test-Value-2", timeout=30)
dut1.expect("Registering /hello and /echo URIs", timeout=30)
# Generate random data of 10KB
- random_data = ''.join(string.printable[random.randint(0,len(string.printable))-1] for _ in range(10*1024))
+ random_data = ''.join(string.printable[random.randint(0,len(string.printable)) - 1] for _ in range(10 * 1024))
Utility.console_log("Test /echo POST handler with random data")
if not client.test_post_handler(got_ip, got_port, random_data):
raise RuntimeError
raise RuntimeError
dut1.expect("400 Bad Request - Server unable to understand request due to invalid syntax", timeout=30)
+
if __name__ == '__main__':
test_examples_protocol_http_server_simple()
from __future__ import print_function
from __future__ import unicode_literals
-from future import standard_library
-standard_library.install_aliases()
from builtins import str
import http.client
import argparse
+import Utility
+
def verbose_print(verbosity, *args):
if (verbosity):
Utility.console_log(''.join(str(elems) for elems in args))
-def test_get_handler(ip, port, verbosity = False):
+
+def test_get_handler(ip, port, verbosity=False):
verbose_print(verbosity, "======== GET HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
uri = "/hello?query1=value1&query2=value2&query3=value3"
# GET hello response
return False
if resp.getheader("Custom-Header-2") != "Custom-Value-2":
return False
- except:
+ except Exception:
return False
verbose_print(verbosity, "vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv")
sess.close()
return (resp_data == "Hello World!")
-def test_post_handler(ip, port, msg, verbosity = False):
+
+def test_post_handler(ip, port, msg, verbosity=False):
verbose_print(verbosity, "======== POST HANDLER TEST ============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
# POST message to /echo and get back response
sess.request("POST", url="/echo", body=msg)
sess.close()
return (resp_data == msg)
-def test_put_handler(ip, port, verbosity = False):
+
+def test_put_handler(ip, port, verbosity=False):
verbose_print(verbosity, "======== PUT HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
# PUT message to /ctrl to disable /hello URI handler
verbose_print(verbosity, "Disabling /hello handler")
sess.close()
return ((resp_data2 == "Hello World!") and (resp_data1 == "This URI doesn't exist"))
-def test_custom_uri_query(ip, port, query, verbosity = False):
+
+def test_custom_uri_query(ip, port, query, verbosity=False):
verbose_print(verbosity, "======== GET HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
uri = "/hello?" + query
# GET hello response
sess.close()
return (resp_data == "Hello World!")
+
if __name__ == '__main__':
# Configure argument parser
parser = argparse.ArgumentParser(description='Run HTTPd Test')
- parser.add_argument('IP' , metavar='IP' , type=str, help='Server IP')
+ parser.add_argument('IP', metavar='IP', type=str, help='Server IP')
parser.add_argument('port', metavar='port', type=str, help='Server port')
parser.add_argument('msg', metavar='message', type=str, help='Message to be sent to server')
args = vars(parser.parse_args())
port = args['port']
msg = args['msg']
- if not test_get_handler (ip, port, True):
+ if not test_get_handler(ip, port, True):
Utility.console_log("Failed!")
if not test_post_handler(ip, port, msg, True):
Utility.console_log("Failed!")
- if not test_put_handler (ip, port, True):
+ if not test_put_handler(ip, port, True):
Utility.console_log("Failed!")
import os
import sys
-# 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", ignore=True)
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "https_request.bin")
bin_size = os.path.getsize(binary_file)
- IDF.log_performance("https_request_bin_size", "{}KB".format(bin_size//1024))
- IDF.check_performance("https_request_bin_size", bin_size//1024)
+ IDF.log_performance("https_request_bin_size", "{}KB".format(bin_size // 1024))
+ IDF.check_performance("https_request_bin_size", bin_size // 1024)
# start test
dut1.start_app()
dut1.expect("Connection established...", timeout=30)
import sys
import socket
import time
-import imp
import struct
-import dpkt, dpkt.dns
+import dpkt
+import dpkt.dns
from threading import Thread
-
# 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:
+ 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
+import DUT
g_run_server = True
g_done = False
+
def mdns_server(esp_host):
global g_run_server
global g_done
- UDP_IP="0.0.0.0"
- UDP_PORT=5353
+ UDP_IP = "0.0.0.0"
+ UDP_PORT = 5353
MCAST_GRP = '224.0.0.251'
- sock = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
+ sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
- sock.bind( (UDP_IP,UDP_PORT) )
+ sock.bind((UDP_IP,UDP_PORT))
mreq = struct.pack("4sl", socket.inet_aton(MCAST_GRP), socket.INADDR_ANY)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
dns = dpkt.dns.DNS(b'\x00\x00\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01')
arr.cls = dpkt.dns.DNS_IN
arr.type = dpkt.dns.DNS_A
arr.name = u'tinytester.local'
- arr.ip =socket.inet_aton('127.0.0.1')
+ arr.ip = socket.inet_aton('127.0.0.1')
resp_dns. an.append(arr)
sock.sendto(resp_dns.pack(),(MCAST_GRP,UDP_PORT))
while g_run_server:
try:
- m=sock.recvfrom( 1024 );
+ m = sock.recvfrom(1024)
dns = dpkt.dns.DNS(m[0])
- if len(dns.qd)>0 and dns.qd[0].type == dpkt.dns.DNS_A:
+ if len(dns.qd) > 0 and dns.qd[0].type == dpkt.dns.DNS_A:
if dns.qd[0].name == u'tinytester.local':
- print (dns.__repr__(),dns.qd[0].name)
+ print(dns.__repr__(),dns.qd[0].name)
sock.sendto(resp_dns.pack(),(MCAST_GRP,UDP_PORT))
- if len(dns.an)>0 and dns.an[0].type == dpkt.dns.DNS_A:
- if dns.an[0].name == esp_host + u'.local':
+ if len(dns.an) > 0 and dns.an[0].type == dpkt.dns.DNS_A:
+ if dns.an[0].name == esp_host + u'.local':
print("Received answer esp32-mdns query")
g_done = True
- print (dns.an[0].name)
+ print(dns.an[0].name)
dns = dpkt.dns.DNS(b'\x00\x00\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01')
- dns.qd[0].name= esp_host + u'.local'
+ dns.qd[0].name = esp_host + u'.local'
sock.sendto(dns.pack(),(MCAST_GRP,UDP_PORT))
print("Sending esp32-mdns query")
time.sleep(0.5)
except socket.timeout:
break
+
@IDF.idf_example_test(env_tag="Example_WIFI")
def test_examples_protocol_mdns(env, extra_data):
global g_run_server
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "mdns-test.bin")
bin_size = os.path.getsize(binary_file)
- IDF.log_performance("mdns-test_bin_size", "{}KB".format(bin_size//1024))
- IDF.check_performance("mdns-test_bin_size", bin_size//1024)
+ IDF.log_performance("mdns-test_bin_size", "{}KB".format(bin_size // 1024))
+ IDF.check_performance("mdns-test_bin_size", bin_size // 1024)
# 1. start mdns application
dut1.start_app()
# 2. get the dut host name (and IP address)
specific_host = dut1.expect(re.compile(r"mdns hostname set to: \[([^\]]+)\]"), timeout=30)
specific_host = str(specific_host[0])
- dut_ip = ""
try:
- dut_ip = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
+ dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)
except DUT.ExpectTimeout:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
# 3. check the mdns name is accessible
- thread1 = Thread(target = mdns_server, args = (specific_host,))
+ thread1 = Thread(target=mdns_server, args=(specific_host,))
thread1.start()
start = time.time()
while (time.time() - start) <= 60:
break
g_run_server = False
thread1.join()
- if g_done == False:
+ if g_done is False:
raise ValueError('Test has failed: did not receive mdns answer within timeout')
# 4. check DUT output if mdns advertized host is resolved
dut1.expect(re.compile(r"mdns-test: Query A: tinytester.local resolved to: 127.0.0.1"), timeout=30)
+
if __name__ == '__main__':
test_examples_protocol_mdns()
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")
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
# 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
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...")
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!")
# 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()
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
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):
# 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)
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)
print("Failure!")
raise ValueError('Mismatch of msgid: received: {}, enqueued {}, deleted {}'.format(msgid, msgid_enqueued, msgid_deleted))
+
if __name__ == '__main__':
test_examples_protocol_mqtt_qos1()
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")
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
# 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
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...")
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!")
# 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()
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")
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
# 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
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...")
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!")
# 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()
import socket
import sys
-# ----------- Config ----------
-PORT = 3333;
+# ----------- Config ----------
+PORT = 3333
IP_VERSION = 'IPv4'
IPV4 = '192.168.0.167'
IPV6 = 'FE80::32AE:A4FF:FE80:5288'
sock = socket.socket(family_addr, socket.SOCK_STREAM)
except socket.error as msg:
print('Could not create socket: ' + str(msg[0]) + ': ' + msg[1])
- sys.exit(1);
+ sys.exit(1)
try:
sock.connect((host, PORT))
except socket.error as msg:
print('Could not open socket: ', msg)
sock.close()
- sys.exit(1);
+ sys.exit(1)
while True:
msg = input('Enter message to send: ')
msg = msg.encode()
sock.sendall(msg)
data = sock.recv(1024)
- if not data: break;
- print( 'Reply: ' + data.decode())
-sock.close()
\ No newline at end of file
+ if not data:
+ break
+ print('Reply: ' + data.decode())
+sock.close()
import socket
import sys
-# ----------- Config ----------
+# ----------- Config ----------
IP_VERSION = 'IPv4'
-PORT = 3333;
+PORT = 3333
# -------------------------------
if IP_VERSION == 'IPv4':
sys.exit(1)
print('Socket created')
-
+
try:
sock.bind(('', PORT))
print('Socket binded')
while True:
data = conn.recv(128)
- if not data: break
+ if not data:
+ break
data = data.decode()
print('Received data: ' + data)
reply = 'OK: ' + data
from builtins import input
import socket
-import sys
+import sys
-# ----------- Config ----------
+# ----------- Config ----------
PORT = 3333
IP_VERSION = 'IPv4'
IPV4 = '192.168.0.167'
# -------------------------------
if IP_VERSION == 'IPv4':
- host = IPV4
+ host = IPV4
family_addr = socket.AF_INET
elif IP_VERSION == 'IPv6':
host = IPV6
try:
- sock = socket.socket(family_addr, socket.SOCK_DGRAM)
-except socket.error as msg:
+ sock = socket.socket(family_addr, socket.SOCK_DGRAM)
+except socket.error:
print('Failed to create socket')
sys.exit()
-
+
while True:
msg = input('Enter message to send : ')
try:
sock.sendto(msg.encode(), (host, PORT))
reply, addr = sock.recvfrom(128)
- if not reply: break
+ if not reply:
+ break
print('Reply[' + addr[0] + ':' + str(addr[1]) + '] - ' + str(reply))
except socket.error as msg:
print('Error Code : ' + str(msg[0]) + ' Message: ' + msg[1])
- sys.exit()
\ No newline at end of file
+ sys.exit()
import socket
import sys
-# ----------- Config ----------
+# ----------- Config ----------
IP_VERSION = 'IPv4'
-PORT = 3333;
+PORT = 3333
# -------------------------------
if IP_VERSION == 'IPv4':
sys.exit(1)
-try :
+try:
sock = socket.socket(family_addr, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-except socket.error as msg :
+except socket.error as msg:
print('Failed to create socket. Error Code : ' + str(msg[0]) + ' Message ' + msg[1])
sys.exit()
sys.exit()
while True:
- try :
+ try:
print('Waiting for data...')
data, addr = sock.recvfrom(1024)
- if not data: break
+ if not data:
+ break
data = data.decode()
print('Reply[' + addr[0] + ':' + str(addr[1]) + '] - ' + data)
reply = 'OK ' + data
import sys
import subprocess
-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:
+ 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')
import os
import sys
-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:
+ 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
@IDF.idf_example_test(env_tag='Example_WIFI')
def test_examples_system_cpp_exceptions(env, extra_data):
'In destructor, m_arg=42',
'Exception caught: Exception in constructor',
'app_main done'
- ]
+ ]
for line in lines:
dut.expect(line, timeout=2)
+
if __name__ == '__main__':
test_examples_system_cpp_exceptions()
from __future__ import print_function
-import re
import os
import sys
-# 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
# Timer events
TIMER_EVENT_LIMIT = 3
TASK_ITERATION_POST = "TASK_EVENTS:TASK_ITERATION_EVENT: posting to default loop, {} out of " + str(TASK_ITERATION_LIMIT)
TASK_ITERATION_HANDLING = "TASK_EVENTS:TASK_ITERATION_EVENT: task_iteration_handler, executed {} times"
+
def _test_timer_events(dut):
dut.start_app()
if expiries < TIMER_EVENT_LIMIT:
dut.expect_all("TIMER_EVENTS:TIMER_EVENT_EXPIRY: all_event_handler",
- "TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_any_handler",
- TIMER_EXPIRY_HANDLING.format(expiries))
+ "TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_any_handler",
+ TIMER_EXPIRY_HANDLING.format(expiries))
else:
dut.expect_all("TIMER_EVENTS:TIMER_EVENT_STOPPED: posting to default loop",
- "TIMER_EVENTS:TIMER_EVENT_EXPIRY: all_event_handler",
- "TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_any_handler",
- TIMER_EXPIRY_HANDLING.format(expiries))
+ "TIMER_EVENTS:TIMER_EVENT_EXPIRY: all_event_handler",
+ "TIMER_EVENTS:TIMER_EVENT_EXPIRY: timer_any_handler",
+ TIMER_EXPIRY_HANDLING.format(expiries))
print("Posted timer stopped event")
print("Handled timer expiry event {} out of {}".format(expiries, TIMER_EVENT_LIMIT))
dut.expect("TIMER_EVENTS:TIMER_EVENT_STOPPED: deleted timer event source")
print("Handled timer stopped event")
+
def _test_iteration_events(dut):
dut.start_app()
dut.expect("TASK_EVENTS:TASK_ITERATION_EVENT: deleting task event source")
print("Deleted task event source")
+
@IDF.idf_example_test(env_tag='Example_WIFI')
def test_default_event_loop_example(env, extra_data):
dut = env.get_dut('default_event_loop', 'examples/system/event/default_event_loop')
_test_iteration_events(dut)
_test_timer_events(dut)
+
if __name__ == '__main__':
test_default_event_loop_example()
from __future__ import print_function
-import re
import os
import sys
-# 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
TASK_ITERATION_LIMIT = 10
TASK_ITERATION_POSTING = "posting TASK_EVENTS:TASK_ITERATION_EVENT to {}, iteration {} out of " + str(TASK_ITERATION_LIMIT)
TASK_ITERATION_HANDLING = "handling TASK_EVENTS:TASK_ITERATION_EVENT from {}, iteration {}"
+
@IDF.idf_example_test(env_tag='Example_WIFI')
def test_user_event_loops_example(env, extra_data):
dut = env.get_dut('user_event_loops', 'examples/system/event/user_event_loops')
dut.expect("deleting task event source")
print("Deleted task event source")
+
if __name__ == '__main__':
test_user_event_loops_example()
import os
import sys
-# 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
STARTING_TIMERS_REGEX = re.compile(r'Started timers, time since boot: (\d+) us')
LIGHT_SLEEP_TIME = 500000
ONE_SHOT_TIMER_PERIOD = 5000000
+
@IDF.idf_example_test(env_tag='Example_WIFI')
def test_examples_system_esp_timer(env, extra_data):
dut = env.get_dut('esp_timer_example', 'examples/system/esp_timer')
groups = dut.expect(STARTING_TIMERS_REGEX, timeout=30)
start_time = int(groups[0])
print('Start time: {} us'.format(start_time))
-
+
groups = dut.expect(TIMER_DUMP_LINE_REGEX, timeout=2)
assert(groups[0] == 'periodic' and int(groups[1]) == INITIAL_TIMER_PERIOD)
groups = dut.expect(TIMER_DUMP_LINE_REGEX, timeout=2)
assert(groups[0] == 'one-shot' and int(groups[1]) == 0)
-
+
for i in range(0, 5):
groups = dut.expect(PERIODIC_TIMER_REGEX, timeout=2)
cur_time = int(groups[0])
dut.expect(STOP_REGEX, timeout=2)
+
if __name__ == '__main__':
test_examples_system_esp_timer()
import sys
import time
-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:
+ 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
ENTERING_SLEEP_STR = 'Entering light sleep'
EXIT_SLEEP_REGEX = re.compile(r'Returned from light sleep, reason: (\w+), t=(\d+) ms, slept for (\d+) ms')
WAKEUP_INTERVAL_MS = 2000
+
@IDF.idf_example_test(env_tag='Example_WIFI')
def test_examples_system_light_sleep(env, extra_data):
dut = env.get_dut('light_sleep_example', 'examples/system/light_sleep')
assert(groups[0] == 'timer' and int(groups[2]) == WAKEUP_INTERVAL_MS)
print('Woke up from timer again')
+
if __name__ == '__main__':
test_examples_system_light_sleep()
import sys
import subprocess
-# 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')
import time
import subprocess
-# add current folder to system path for importing test_report
-sys.path.append(os.path.dirname(__file__))
-# 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
+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
import Utility
from Utility import (Attenuator, PowerControl, LineChart)
-from test_report import (ThroughputForConfigsReport, ThroughputVsRssiReport)
+
+try:
+ from test_report import (ThroughputForConfigsReport, ThroughputVsRssiReport)
+except ImportError:
+ # add current folder to system path for importing test_report
+ sys.path.append(os.path.dirname(__file__))
+ from test_report import (ThroughputForConfigsReport, ThroughputVsRssiReport)
# configurations
TEST_TIME = TEST_TIMEOUT = 60
BAD_POINT_PERCENTAGE_THRESHOLD = 0.3
# we need at least 1/2 valid points to qualify the test result
- THROUGHPUT_QUALIFY_COUNT = TEST_TIME//2
+ THROUGHPUT_QUALIFY_COUNT = TEST_TIME // 2
def __init__(self, proto, direction, config_name):
self.proto = proto
current_config = self.sdkconfigs[_config_name]
if i > 0:
- previous_config_name = self.sort_order[i-1]
+ previous_config_name = self.sort_order[i - 1]
previous_config = self.sdkconfigs[previous_config_name]
else:
previous_config = previous_config_name = None
projects / esp-idf / commitdiff