diff --git a/examples/bluetooth/nimble/blecent/blecent_test.py b/examples/bluetooth/nimble/blecent/blecent_test.py index eabe8df26..9a85afc06 100644 --- a/examples/bluetooth/nimble/blecent/blecent_test.py +++ b/examples/bluetooth/nimble/blecent/blecent_test.py @@ -16,43 +16,20 @@ from __future__ import print_function import os -import sys import re import uuid import subprocess -try: - # 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 - from IDF.IDFDUT import ESP32DUT -except ImportError as e: - print(e) - print("\nCheck your IDF_PATH\nOR") - print("Try `export TEST_FW_PATH=$IDF_PATH/tools/tiny-test-fw` for resolving the issue\nOR") - print("Try `pip install -r $IDF_PATH/tools/tiny-test-fw/requirements.txt` for resolving the issue") - import IDF - -try: - import lib_ble_client -except ImportError: - lib_ble_client_path = os.getenv("IDF_PATH") + "/tools/ble" - if lib_ble_client_path and lib_ble_client_path not in sys.path: - sys.path.insert(0, lib_ble_client_path) - import lib_ble_client - - -import Utility +from tiny_test_fw import Utility +import ttfw_idf +from ble import lib_ble_client # 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 -@IDF.idf_example_test(env_tag="Example_WIFI_BT") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI_BT") def test_example_app_ble_central(env, extra_data): """ Steps: @@ -68,12 +45,12 @@ def test_example_app_ble_central(env, extra_data): subprocess.check_output(['rm','-rf','/var/lib/bluetooth/*']) subprocess.check_output(['hciconfig','hci0','reset']) # Acquire DUT - dut = env.get_dut("blecent", "examples/bluetooth/nimble/blecent", dut_class=ESP32DUT) + dut = env.get_dut("blecent", "examples/bluetooth/nimble/blecent", dut_class=ttfw_idf.ESP32DUT) # Get binary file binary_file = os.path.join(dut.app.binary_path, "blecent.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("blecent_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.log_performance("blecent_bin_size", "{}KB".format(bin_size // 1024)) # Upload binary and start testing Utility.console_log("Starting blecent example test app") diff --git a/examples/bluetooth/nimble/blehr/blehr_test.py b/examples/bluetooth/nimble/blehr/blehr_test.py index 432a16013..84de1eed5 100644 --- a/examples/bluetooth/nimble/blehr/blehr_test.py +++ b/examples/bluetooth/nimble/blehr/blehr_test.py @@ -16,42 +16,19 @@ from __future__ import print_function import os -import sys import re import threading import traceback import Queue import subprocess -try: - # 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 - from IDF.IDFDUT import ESP32DUT -except ImportError as e: - print(e) - print("\nCheck your IDF_PATH\nOR") - print("Try `export TEST_FW_PATH=$IDF_PATH/tools/tiny-test-fw` for resolving the issue\nOR") - print("Try `pip install -r $IDF_PATH/tools/tiny-test-fw/requirements.txt` for resolving the issue\n") - import IDF - -try: - import lib_ble_client -except ImportError: - lib_ble_client_path = os.getenv("IDF_PATH") + "/tools/ble" - if lib_ble_client_path and lib_ble_client_path not in sys.path: - sys.path.insert(0, lib_ble_client_path) - import lib_ble_client - - -import Utility +from tiny_test_fw import Utility +import ttfw_idf +from ble import lib_ble_client # 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 def blehr_client_task(hr_obj, dut_addr): @@ -115,7 +92,7 @@ class BleHRThread(threading.Thread): self.exceptions_queue.put(traceback.format_exc(), block=False) -@IDF.idf_example_test(env_tag="Example_WIFI_BT") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI_BT") def test_example_app_ble_hr(env, extra_data): """ Steps: @@ -129,13 +106,13 @@ def test_example_app_ble_hr(env, extra_data): subprocess.check_output(['hciconfig','hci0','reset']) # Acquire DUT - dut = env.get_dut("blehr", "examples/bluetooth/nimble/blehr", dut_class=ESP32DUT) + dut = env.get_dut("blehr", "examples/bluetooth/nimble/blehr", dut_class=ttfw_idf.ESP32DUT) # Get binary file binary_file = os.path.join(dut.app.binary_path, "blehr.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("blehr_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("blehr_bin_size", bin_size // 1024) + ttfw_idf.log_performance("blehr_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("blehr_bin_size", bin_size // 1024) # Upload binary and start testing Utility.console_log("Starting blehr simple example test app") diff --git a/examples/bluetooth/nimble/bleprph/bleprph_test.py b/examples/bluetooth/nimble/bleprph/bleprph_test.py index 70615c12b..f55729571 100644 --- a/examples/bluetooth/nimble/bleprph/bleprph_test.py +++ b/examples/bluetooth/nimble/bleprph/bleprph_test.py @@ -16,36 +16,15 @@ from __future__ import print_function import os -import sys import re import Queue import traceback import threading import subprocess -try: - # 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 - from IDF.IDFDUT import ESP32DUT -except ImportError as e: - print(e) - print("Try `export TEST_FW_PATH=$IDF_PATH/tools/tiny-test-fw` for resolving the issue") - print("Try `pip install -r $IDF_PATH/tools/tiny-test-fw/requirements.txt` for resolving the issue") - import IDF - -try: - import lib_ble_client -except ImportError: - lib_ble_client_path = os.getenv("IDF_PATH") + "/tools/ble" - if lib_ble_client_path and lib_ble_client_path not in sys.path: - sys.path.insert(0, lib_ble_client_path) - import lib_ble_client - - -import Utility +from tiny_test_fw import Utility +import ttfw_idf +from ble import lib_ble_client # When running on local machine execute the following before running this script # > make app bootloader @@ -136,7 +115,7 @@ class BlePrphThread(threading.Thread): self.exceptions_queue.put(traceback.format_exc(), block=False) -@IDF.idf_example_test(env_tag="Example_WIFI_BT") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI_BT") def test_example_app_ble_peripheral(env, extra_data): """ Steps: @@ -150,13 +129,13 @@ def test_example_app_ble_peripheral(env, extra_data): subprocess.check_output(['hciconfig','hci0','reset']) # Acquire DUT - dut = env.get_dut("bleprph", "examples/bluetooth/nimble/bleprph", dut_class=ESP32DUT) + dut = env.get_dut("bleprph", "examples/bluetooth/nimble/bleprph", dut_class=ttfw_idf.ESP32DUT) # Get binary file binary_file = os.path.join(dut.app.binary_path, "bleprph.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("bleprph_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("bleprph_bin_size", bin_size // 1024) + ttfw_idf.log_performance("bleprph_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("bleprph_bin_size", bin_size // 1024) # Upload binary and start testing Utility.console_log("Starting bleprph simple example test app") diff --git a/examples/get-started/blink/example_test.py b/examples/get-started/blink/example_test.py index fbe274bea..10815cf08 100644 --- a/examples/get-started/blink/example_test.py +++ b/examples/get-started/blink/example_test.py @@ -5,21 +5,10 @@ from __future__ import print_function from __future__ import unicode_literals import re import os -import sys import hashlib -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import Utility +import ttfw_idf def verify_elf_sha256_embedding(dut): @@ -39,13 +28,13 @@ def verify_elf_sha256_embedding(dut): raise ValueError('ELF file SHA256 mismatch') -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_blink(env, extra_data): - dut = env.get_dut("blink", "examples/get-started/blink", dut_class=ESP32DUT) + dut = env.get_dut("blink", "examples/get-started/blink", dut_class=ttfw_idf.ESP32DUT) binary_file = os.path.join(dut.app.binary_path, "blink.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("blink_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("blink_bin_size", bin_size // 1024) + ttfw_idf.log_performance("blink_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("blink_bin_size", bin_size // 1024) dut.start_app() diff --git a/examples/get-started/hello_world/loadable_elf_example_test.py b/examples/get-started/hello_world/loadable_elf_example_test.py index c85dbd121..3afaf270a 100644 --- a/examples/get-started/hello_world/loadable_elf_example_test.py +++ b/examples/get-started/hello_world/loadable_elf_example_test.py @@ -1,19 +1,12 @@ import os -import pexpect -import serial -import sys import threading import time -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 pexpect +import serial -import Utility +from tiny_test_fw import Utility +import ttfw_idf class CustomProcess(object): @@ -125,13 +118,13 @@ class SerialThread(object): Utility.console_log('The pyserial thread is still alive', 'O') -@IDF.idf_example_test(env_tag="test_jtag_arm") +@ttfw_idf.idf_example_test(env_tag="test_jtag_arm") def test_examples_loadable_elf(env, extra_data): idf_path = os.environ['IDF_PATH'] rel_project_path = os.path.join('examples', 'get-started', 'hello_world') proj_path = os.path.join(idf_path, rel_project_path) - example = IDF.Example(rel_project_path, target="esp32") + example = ttfw_idf.Example(rel_project_path, target="esp32") sdkconfig = example.get_sdkconfig() elf_path = os.path.join(example.get_binary_path(rel_project_path), 'hello-world.elf') esp_log_path = os.path.join(proj_path, 'esp.log') diff --git a/examples/peripherals/can/can_alert_and_recovery/example_test.py b/examples/peripherals/can/can_alert_and_recovery/example_test.py index 77110053d..4c1a5d461 100644 --- a/examples/peripherals/can/can_alert_and_recovery/example_test.py +++ b/examples/peripherals/can/can_alert_and_recovery/example_test.py @@ -1,28 +1,16 @@ # Need Python 3 string formatting functions from __future__ import print_function -import os -import sys - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +import ttfw_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_CAN1') +@ttfw_idf.idf_example_test(env_tag='Example_CAN1') def test_can_alert_and_recovery_example(env, extra_data): - dut = env.get_dut('dut1', 'examples/peripherals/can/can_alert_and_recovery', dut_class=ESP32DUT) + dut = env.get_dut('dut1', 'examples/peripherals/can/can_alert_and_recovery', dut_class=ttfw_idf.ESP32DUT) dut.start_app() for string in STR_EXPECT: diff --git a/examples/peripherals/can/can_network/example_test.py b/examples/peripherals/can/can_network/example_test.py index 3183baa75..d59873aa7 100644 --- a/examples/peripherals/can/can_network/example_test.py +++ b/examples/peripherals/can/can_network/example_test.py @@ -1,19 +1,9 @@ # Need Python 3 string formatting functions from __future__ import print_function -import os -import sys from threading import Thread -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 + +import ttfw_idf # Define tuple of strings to expect for each DUT. master_expect = ("CAN Master: Driver installed", "CAN Master: Driver uninstalled") @@ -37,13 +27,16 @@ def dut_thread_callback(**kwargs): result[0] = True -@IDF.idf_example_test(env_tag='Example_CAN2') +@ttfw_idf.idf_example_test(env_tag='Example_CAN2') def test_can_network_example(env, extra_data): # 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_class=ESP32DUT) - dut_slave = env.get_dut("dut2", "examples/peripherals/can/can_network/can_network_slave", dut_class=ESP32DUT) - dut_listen_only = env.get_dut("dut3", "examples/peripherals/can/can_network/can_network_listen_only", dut_class=ESP32DUT) + dut_master = env.get_dut("dut1", "examples/peripherals/can/can_network/can_network_master", + dut_class=ttfw_idf.ESP32DUT) + dut_slave = env.get_dut("dut2", "examples/peripherals/can/can_network/can_network_slave", + dut_class=ttfw_idf.ESP32DUT) + dut_listen_only = env.get_dut("dut3", "examples/peripherals/can/can_network/can_network_listen_only", + dut_class=ttfw_idf.ESP32DUT) # Flash app onto each DUT, each DUT is reset again at the start of each thread dut_master.start_app() diff --git a/examples/peripherals/can/can_self_test/example_test.py b/examples/peripherals/can/can_self_test/example_test.py index 9a888643c..a220f960f 100644 --- a/examples/peripherals/can/can_self_test/example_test.py +++ b/examples/peripherals/can/can_self_test/example_test.py @@ -1,18 +1,7 @@ # Need Python 3 string formatting functions from __future__ import print_function -import os -import sys -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +import ttfw_idf # CAN Self Test Example constants @@ -20,10 +9,10 @@ STR_EXPECT = ("CAN Self Test: Driver installed", "CAN Self Test: Driver uninstal EXPECT_TIMEOUT = 20 -@IDF.idf_example_test(env_tag='Example_CAN1') +@ttfw_idf.idf_example_test(env_tag='Example_CAN1') def test_can_self_test_example(env, extra_data): # Get device under test, flash and start example. "dut1" must be defined in EnvConfig - dut = env.get_dut('dut1', 'examples/peripherals/can/can_self_test', dut_class=ESP32DUT) + dut = env.get_dut('dut1', 'examples/peripherals/can/can_self_test', dut_class=ttfw_idf.ESP32DUT) dut.start_app() for string in STR_EXPECT: diff --git a/examples/peripherals/i2c/i2c_tools/example_test.py b/examples/peripherals/i2c/i2c_tools/example_test.py index 0f0c2271d..3731c54d8 100644 --- a/examples/peripherals/i2c/i2c_tools/example_test.py +++ b/examples/peripherals/i2c/i2c_tools/example_test.py @@ -1,23 +1,14 @@ from __future__ import print_function -import os -import sys + +import ttfw_idf EXPECT_TIMEOUT = 20 -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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_I2C_CCS811_SENSOR') +@ttfw_idf.idf_example_test(env_tag='Example_I2C_CCS811_SENSOR') def test_i2ctools_example(env, extra_data): # Get device under test, flash and start example. "i2ctool" must be defined in EnvConfig - dut = env.get_dut('i2ctools', 'examples/peripherals/i2c/i2c_tools', dut_class=ESP32DUT) + dut = env.get_dut('i2ctools', 'examples/peripherals/i2c/i2c_tools', dut_class=ttfw_idf.ESP32DUT) dut.start_app() dut.expect("esp32>", timeout=EXPECT_TIMEOUT) # Get i2c address diff --git a/examples/peripherals/rmt/ir_protocols/example_test.py b/examples/peripherals/rmt/ir_protocols/example_test.py index 9fd9fd816..e61010e4f 100644 --- a/examples/peripherals/rmt/ir_protocols/example_test.py +++ b/examples/peripherals/rmt/ir_protocols/example_test.py @@ -1,19 +1,11 @@ from __future__ import print_function -import os -import sys -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 ttfw_idf EXPECT_TIMEOUT = 20 -@IDF.idf_example_test(env_tag='Example_RMT_IR_PROTOCOLS') +@ttfw_idf.idf_example_test(env_tag='Example_RMT_IR_PROTOCOLS') def test_examples_rmt_ir_protocols(env, extra_data): dut = env.get_dut('ir_protocols_example', 'examples/peripherals/rmt/ir_protocols', app_config_name='nec') print("Using binary path: {}".format(dut.app.binary_path)) diff --git a/examples/peripherals/sdio/sdio_test.py b/examples/peripherals/sdio/sdio_test.py index 0fa8b2979..4025d5394 100644 --- a/examples/peripherals/sdio/sdio_test.py +++ b/examples/peripherals/sdio/sdio_test.py @@ -12,25 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. -""" example of writing test with TinyTestFW """ -import os -import sys - -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 IDF -from IDF.IDFDUT import ESP32DUT +from tiny_test_fw import TinyFW +import ttfw_idf -@IDF.idf_example_test(env_tag="Example_SDIO", ignore=True) +@ttfw_idf.idf_example_test(env_tag="Example_SDIO", ignore=True) def test_example_sdio_communication(env, extra_data): """ Configurations @@ -50,8 +36,8 @@ def test_example_sdio_communication(env, extra_data): or use sdio test board, which has two wrover modules connect to a same FT3232 Assume that first dut is host and second is slave """ - dut1 = env.get_dut("sdio_host", "examples/peripherals/sdio/host", dut_class=ESP32DUT) - dut2 = env.get_dut("sdio_slave", "examples/peripherals/sdio/slave", dut_class=ESP32DUT) + dut1 = env.get_dut("sdio_host", "examples/peripherals/sdio/host", dut_class=ttfw_idf.ESP32DUT) + dut2 = env.get_dut("sdio_slave", "examples/peripherals/sdio/slave", dut_class=ttfw_idf.ESP32DUT) dut1.start_app() # wait until the master is ready to setup the slave dut1.expect("host ready, start initializing slave...") @@ -133,5 +119,5 @@ def test_example_sdio_communication(env, extra_data): if __name__ == '__main__': - TinyFW.set_default_config(env_config_file="EnvConfigTemplate.yml", dut=IDF.IDFDUT) + TinyFW.set_default_config(env_config_file="EnvConfigTemplate.yml", dut=ttfw_idf.IDFDUT) test_example_sdio_communication() diff --git a/examples/protocols/asio/chat_client/asio_chat_client_test.py b/examples/protocols/asio/chat_client/asio_chat_client_test.py index 4325806a3..d59b459c7 100644 --- a/examples/protocols/asio/chat_client/asio_chat_client_test.py +++ b/examples/protocols/asio/chat_client/asio_chat_client_test.py @@ -1,22 +1,10 @@ import re import os -import sys import socket from threading import Thread import time -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf global g_client_response global g_msg_to_client @@ -56,7 +44,7 @@ def chat_server_sketch(my_ip): print("server closed") -@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): """ steps: | @@ -70,12 +58,12 @@ def test_examples_protocol_asio_chat_client(env, extra_data): global g_client_response global g_msg_to_client test_msg = "ABC" - dut1 = env.get_dut("chat_client", "examples/protocols/asio/chat_client", dut_class=ESP32DUT) + dut1 = env.get_dut("chat_client", "examples/protocols/asio/chat_client", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("asio_chat_client_size", "{}KB".format(bin_size // 1024)) + ttfw_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,)) diff --git a/examples/protocols/asio/chat_server/asio_chat_server_test.py b/examples/protocols/asio/chat_server/asio_chat_server_test.py index ccdd4a557..2bfa026fc 100644 --- a/examples/protocols/asio/chat_server/asio_chat_server_test.py +++ b/examples/protocols/asio/chat_server/asio_chat_server_test.py @@ -1,24 +1,11 @@ import re import os -import sys import socket - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@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): """ steps: | @@ -28,12 +15,12 @@ def test_examples_protocol_asio_chat_server(env, extra_data): 4. Test evaluates received test message from server """ test_msg = b" 4ABC\n" - dut1 = env.get_dut("chat_server", "examples/protocols/asio/chat_server", dut_class=ESP32DUT) + dut1 = env.get_dut("chat_server", "examples/protocols/asio/chat_server", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("asio_chat_server_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("asio_chat_server_size", bin_size // 1024) # 1. start test dut1.start_app() # 2. get the server IP address diff --git a/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py b/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py index d5ee5cf07..b7dc16817 100644 --- a/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py +++ b/examples/protocols/asio/tcp_echo_server/asio_tcp_server_test.py @@ -1,25 +1,11 @@ import re import os -import sys import socket - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@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): """ steps: | @@ -30,12 +16,12 @@ def test_examples_protocol_asio_tcp_server(env, extra_data): 5. Test evaluates received test message on server stdout """ test_msg = b"echo message from client to server" - dut1 = env.get_dut("tcp_echo_server", "examples/protocols/asio/tcp_echo_server", dut_class=ESP32DUT) + dut1 = env.get_dut("tcp_echo_server", "examples/protocols/asio/tcp_echo_server", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("asio_tcp_echo_server_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("asio_tcp_echo_server_size", bin_size // 1024) # 1. start test dut1.start_app() # 2. get the server IP address diff --git a/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py b/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py index 726a39f4f..3b6d0bff9 100644 --- a/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py +++ b/examples/protocols/asio/udp_echo_server/asio_udp_server_test.py @@ -1,25 +1,11 @@ import re import os -import sys import socket - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@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): """ steps: | @@ -30,12 +16,12 @@ def test_examples_protocol_asio_udp_server(env, extra_data): 5. Test evaluates received test message on server stdout """ test_msg = b"echo message from client to server" - dut1 = env.get_dut("udp_echo_server", "examples/protocols/asio/udp_echo_server", dut_class=ESP32DUT) + dut1 = env.get_dut("udp_echo_server", "examples/protocols/asio/udp_echo_server", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("asio_udp_echo_server_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("asio_udp_echo_server_size", bin_size // 1024) # 1. start test dut1.start_app() # 2. get the server IP address diff --git a/examples/protocols/esp_http_client/esp_http_client_test.py b/examples/protocols/esp_http_client/esp_http_client_test.py index cdd0888c4..ce7072cdf 100644 --- a/examples/protocols/esp_http_client/esp_http_client_test.py +++ b/examples/protocols/esp_http_client/esp_http_client_test.py @@ -1,35 +1,22 @@ import re import os -import sys -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@IDF.idf_example_test(env_tag="Example_WIFI", ignore=True) +@ttfw_idf.idf_example_test(env_tag="Example_WIFI", ignore=True) def test_examples_protocol_esp_http_client(env, extra_data): """ steps: | 1. join AP 2. Send HTTP request to httpbin.org """ - dut1 = env.get_dut("esp_http_client", "examples/protocols/esp_http_client", dut_class=ESP32DUT) + dut1 = env.get_dut("esp_http_client", "examples/protocols/esp_http_client", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("esp_http_client_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_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) diff --git a/examples/protocols/http_server/advanced_tests/http_server_advanced_test.py b/examples/protocols/http_server/advanced_tests/http_server_advanced_test.py index 754eb5564..c38bfbd53 100644 --- a/examples/protocols/http_server/advanced_tests/http_server_advanced_test.py +++ b/examples/protocols/http_server/advanced_tests/http_server_advanced_test.py @@ -19,31 +19,14 @@ from __future__ import print_function from __future__ import unicode_literals import re import os -import sys -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import Utility +import ttfw_idf +from idf_http_server_test import test as client # 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 client module -# TODO: replace with import -expath = os.path.dirname(os.path.realpath(__file__)) -client = Utility.load_source(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 @@ -51,16 +34,16 @@ client = Utility.load_source(expath + "/scripts/test.py") # of large HTTP packets and malformed requests, running multiple parallel sessions, etc. # It is advised that all these tests be run locally, when making changes or adding new # features to this component. -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_http_server_advanced(env, extra_data): # Acquire DUT - dut1 = env.get_dut("http_server", "examples/protocols/http_server/advanced_tests", dut_class=ESP32DUT) + dut1 = env.get_dut("http_server", "examples/protocols/http_server/advanced_tests", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("http_server_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("http_server_bin_size", bin_size // 1024) # Upload binary and start testing Utility.console_log("Starting http_server advanced test app") diff --git a/examples/protocols/http_server/persistent_sockets/http_server_persistence_test.py b/examples/protocols/http_server/persistent_sockets/http_server_persistence_test.py index 71cf0db96..1f635a13a 100644 --- a/examples/protocols/http_server/persistent_sockets/http_server_persistence_test.py +++ b/examples/protocols/http_server/persistent_sockets/http_server_persistence_test.py @@ -21,42 +21,28 @@ from builtins import str from builtins import range import re import os -import sys import random -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import Utility +import ttfw_idf +from idf_http_server_test import adder as client # 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 client module -# TODO: replace with import -expath = os.path.dirname(os.path.realpath(__file__)) -client = Utility.load_source(expath + "/scripts/adder.py") -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_http_server_persistence(env, extra_data): # Acquire DUT - dut1 = env.get_dut("http_server", "examples/protocols/http_server/persistent_sockets", dut_class=ESP32DUT) + dut1 = env.get_dut("http_server", "examples/protocols/http_server/persistent_sockets", + dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("http_server_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("http_server_bin_size", bin_size // 1024) # Upload binary and start testing Utility.console_log("Starting http_server persistance test app") diff --git a/examples/protocols/http_server/simple/http_server_simple_test.py b/examples/protocols/http_server/simple/http_server_simple_test.py index bb209e204..0e9242fc0 100644 --- a/examples/protocols/http_server/simple/http_server_simple_test.py +++ b/examples/protocols/http_server/simple/http_server_simple_test.py @@ -20,43 +20,29 @@ from __future__ import unicode_literals from builtins import range import re import os -import sys import string import random -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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) +from tiny_test_fw import Utility +import ttfw_idf +from idf_http_server_test import client - 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 client module -expath = os.path.dirname(os.path.realpath(__file__)) -client = Utility.load_source(expath + "/scripts/client.py") -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_http_server_simple(env, extra_data): # Acquire DUT - dut1 = env.get_dut("http_server", "examples/protocols/http_server/simple", dut_class=ESP32DUT) + dut1 = env.get_dut("http_server", "examples/protocols/http_server/simple", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("http_server_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("http_server_bin_size", bin_size // 1024) # Upload binary and start testing Utility.console_log("Starting http_server simple test app") diff --git a/examples/protocols/https_request/example_test.py b/examples/protocols/https_request/example_test.py index 543edc3da..c86213f89 100644 --- a/examples/protocols/https_request/example_test.py +++ b/examples/protocols/https_request/example_test.py @@ -1,22 +1,10 @@ import re import os -import sys -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@IDF.idf_example_test(env_tag="Example_WIFI", ignore=True) +@ttfw_idf.idf_example_test(env_tag="Example_WIFI", ignore=True) def test_examples_protocol_https_request(env, extra_data): """ steps: | @@ -24,12 +12,12 @@ def test_examples_protocol_https_request(env, extra_data): 2. connect to www.howsmyssl.com:443 3. send http request """ - dut1 = env.get_dut("https_request", "examples/protocols/https_request", dut_class=ESP32DUT) + dut1 = env.get_dut("https_request", "examples/protocols/https_request", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("https_request_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("https_request_bin_size", bin_size // 1024) # start test dut1.start_app() dut1.expect("Connection established...", timeout=30) diff --git a/examples/protocols/mdns/mdns_example_test.py b/examples/protocols/mdns/mdns_example_test.py index e86b84e42..af897bb79 100644 --- a/examples/protocols/mdns/mdns_example_test.py +++ b/examples/protocols/mdns/mdns_example_test.py @@ -1,6 +1,5 @@ import re import os -import sys import socket import time import struct @@ -8,22 +7,8 @@ import dpkt import dpkt.dns from threading import Thread, Event - -# 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 - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 DUT +from tiny_test_fw import DUT +import ttfw_idf # g_run_server = True # g_done = False @@ -103,7 +88,7 @@ def mdns_server(esp_host): continue -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_mdns(env, extra_data): global stop_mdns_server """ @@ -113,12 +98,12 @@ def test_examples_protocol_mdns(env, extra_data): 3. check the mdns name is accessible 4. check DUT output if mdns advertized host is resolved """ - dut1 = env.get_dut("mdns-test", "examples/protocols/mdns", dut_class=ESP32DUT) + dut1 = env.get_dut("mdns-test", "examples/protocols/mdns", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("mdns-test_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_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) diff --git a/examples/protocols/modbus/serial/example_test.py b/examples/protocols/modbus/serial/example_test.py index e1ffa9736..0902e1cb8 100644 --- a/examples/protocols/modbus/serial/example_test.py +++ b/examples/protocols/modbus/serial/example_test.py @@ -2,20 +2,11 @@ from __future__ import print_function import os -import sys import re import logging from threading import Thread -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 +import ttfw_idf LOG_LEVEL = logging.DEBUG LOGGER_NAME = "modbus_test" @@ -174,13 +165,13 @@ def test_check_mode(dut=None, mode_str=None, value=None): return False -@IDF.idf_example_test(env_tag='Example_T2_RS485') +@ttfw_idf.idf_example_test(env_tag='Example_T2_RS485') def test_modbus_communication(env, comm_mode): global logger # Get device under test. "dut1 - master", "dut2 - slave" must be properly connected through RS485 interface driver - dut_master = env.get_dut("modbus_master", "examples/protocols/modbus/serial/mb_master") - dut_slave = env.get_dut("modbus_slave", "examples/protocols/modbus/serial/mb_slave") + dut_master = env.get_dut("modbus_master", "examples/protocols/modbus/serial/mb_master", dut_class=ttfw_idf.ESP32DUT) + dut_slave = env.get_dut("modbus_slave", "examples/protocols/modbus/serial/mb_slave", dut_class=ttfw_idf.ESP32DUT) try: logger.debug("Environment vars: %s\r\n" % os.environ) diff --git a/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py b/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py index 43d4cec8b..a392b2cae 100644 --- a/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py +++ b/examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py @@ -8,21 +8,8 @@ import ssl import paho.mqtt.client as mqtt from threading import Thread, Event - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import DUT +import ttfw_idf event_client_connected = Event() @@ -73,7 +60,7 @@ def on_message(client, userdata, msg): message_log += "Received data:" + msg.topic + " " + payload + "\n" -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_mqtt_ssl(env, extra_data): broker_url = "" broker_port = 0 @@ -85,13 +72,13 @@ def test_examples_protocol_mqtt_ssl(env, extra_data): 4. Test ESP32 client received correct qos0 message 5. Test python client receives binary data from running partition and compares it with the binary """ - dut1 = env.get_dut("mqtt_ssl", "examples/protocols/mqtt/ssl", dut_class=ESP32DUT) + dut1 = env.get_dut("mqtt_ssl", "examples/protocols/mqtt/ssl", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("mqtt_ssl_bin_size", "{}KB" + .format(bin_size // 1024)) + ttfw_idf.check_performance("mqtt_ssl_size", bin_size // 1024) # Look for host:port in sdkconfig try: value = re.search(r'\:\/\/([^:]+)\:([0-9]+)', dut1.app.get_sdkconfig()["CONFIG_BROKER_URI"]) diff --git a/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py b/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py index 4e3970564..2d8dbe341 100644 --- a/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py +++ b/examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py @@ -6,21 +6,8 @@ from threading import Thread import struct import time - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import DUT +import ttfw_idf msgid = -1 @@ -66,7 +53,7 @@ def mqqt_server_sketch(my_ip, port): print("server closed") -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_mqtt_qos1(env, extra_data): global msgid """ @@ -76,12 +63,12 @@ def test_examples_protocol_mqtt_qos1(env, extra_data): 3. Test evaluates that qos1 message is queued and removed from queued after ACK received 4. Test the broker received the same message id evaluated in step 3 """ - dut1 = env.get_dut("mqtt_tcp", "examples/protocols/mqtt/tcp", dut_class=ESP32DUT) + dut1 = env.get_dut("mqtt_tcp", "examples/protocols/mqtt/tcp", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("mqtt_tcp_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_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)) diff --git a/examples/protocols/mqtt/ws/mqtt_ws_example_test.py b/examples/protocols/mqtt/ws/mqtt_ws_example_test.py index 19ed395d6..cf40c9a3a 100644 --- a/examples/protocols/mqtt/ws/mqtt_ws_example_test.py +++ b/examples/protocols/mqtt/ws/mqtt_ws_example_test.py @@ -7,21 +7,8 @@ import sys import paho.mqtt.client as mqtt from threading import Thread, Event - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import DUT +import ttfw_idf event_client_connected = Event() event_stop_client = Event() @@ -52,7 +39,7 @@ def on_message(client, userdata, msg): message_log += "Received data:" + msg.topic + " " + payload + "\n" -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_mqtt_ws(env, extra_data): broker_url = "" broker_port = 0 @@ -63,12 +50,12 @@ def test_examples_protocol_mqtt_ws(env, extra_data): 3. Test evaluates it received correct qos0 message 4. Test ESP32 client received correct qos0 message """ - dut1 = env.get_dut("mqtt_websocket", "examples/protocols/mqtt/ws", dut_class=ESP32DUT) + dut1 = env.get_dut("mqtt_websocket", "examples/protocols/mqtt/ws", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("mqtt_websocket_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("mqtt_websocket_size", bin_size // 1024) # Look for host:port in sdkconfig try: value = re.search(r'\:\/\/([^:]+)\:([0-9]+)', dut1.app.get_sdkconfig()["CONFIG_BROKER_URI"]) diff --git a/examples/protocols/mqtt/wss/mqtt_wss_example_test.py b/examples/protocols/mqtt/wss/mqtt_wss_example_test.py index 486d65d31..a4c4e6c02 100644 --- a/examples/protocols/mqtt/wss/mqtt_wss_example_test.py +++ b/examples/protocols/mqtt/wss/mqtt_wss_example_test.py @@ -8,21 +8,9 @@ import ssl import paho.mqtt.client as mqtt from threading import Thread, Event +from tiny_test_fw import DUT +import ttfw_idf -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 event_client_connected = Event() event_stop_client = Event() @@ -53,7 +41,7 @@ def on_message(client, userdata, msg): message_log += "Received data:" + msg.topic + " " + payload + "\n" -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_mqtt_wss(env, extra_data): broker_url = "" broker_port = 0 @@ -64,12 +52,12 @@ def test_examples_protocol_mqtt_wss(env, extra_data): 3. Test evaluates it received correct qos0 message 4. Test ESP32 client received correct qos0 message """ - dut1 = env.get_dut("mqtt_websocket_secure", "examples/protocols/mqtt/wss", dut_class=ESP32DUT) + dut1 = env.get_dut("mqtt_websocket_secure", "examples/protocols/mqtt/wss", dut_class=ttfw_idf.ESP32DUT) # 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) + ttfw_idf.log_performance("mqtt_websocket_secure_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("mqtt_websocket_secure_size", bin_size // 1024) # Look for host:port in sdkconfig try: value = re.search(r'\:\/\/([^:]+)\:([0-9]+)', dut1.app.get_sdkconfig()["CONFIG_BROKER_URI"]) diff --git a/examples/protocols/websocket/example_test.py b/examples/protocols/websocket/example_test.py index 4ca10b204..8c21c6c03 100644 --- a/examples/protocols/websocket/example_test.py +++ b/examples/protocols/websocket/example_test.py @@ -1,19 +1,10 @@ import re import os -import sys -import IDF -from IDF.IDFDUT import ESP32DUT -# 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 ttfw_idf -@IDF.idf_example_test(env_tag="Example_WIFI", ignore=True) +@ttfw_idf.idf_example_test(env_tag="Example_WIFI", ignore=True) def test_examples_protocol_websocket(env, extra_data): """ steps: | @@ -21,12 +12,12 @@ def test_examples_protocol_websocket(env, extra_data): 2. connect to ws://echo.websocket.org 3. send and receive data """ - dut1 = env.get_dut("websocket", "examples/protocols/websocket", dut_class=ESP32DUT) + dut1 = env.get_dut("websocket", "examples/protocols/websocket", dut_class=ttfw_idf.ESP32DUT) # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "websocket-example.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("websocket_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("websocket_bin_size", bin_size // 1024) + ttfw_idf.log_performance("websocket_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("websocket_bin_size", bin_size // 1024) # start test dut1.start_app() dut1.expect("Waiting for wifi ...") diff --git a/examples/provisioning/ble_prov/ble_prov_test.py b/examples/provisioning/ble_prov/ble_prov_test.py index 760756a43..1e3e8508b 100644 --- a/examples/provisioning/ble_prov/ble_prov_test.py +++ b/examples/provisioning/ble_prov/ble_prov_test.py @@ -17,40 +17,25 @@ from __future__ import print_function import re import os -import sys import time -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 - -try: - import esp_prov -except ImportError: - esp_prov_path = os.getenv("IDF_PATH") + "/tools/esp_prov" - if esp_prov_path and esp_prov_path not in sys.path: - sys.path.insert(0, esp_prov_path) - import esp_prov +import ttfw_idf +import esp_prov # Have esp_prov throw exception esp_prov.config_throw_except = True -@IDF.idf_example_test(env_tag="Example_WIFI_BT") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI_BT") def test_examples_provisioning_ble(env, extra_data): # Acquire DUT - dut1 = env.get_dut("ble_prov", "examples/provisioning/ble_prov", dut_class=ESP32DUT) + dut1 = env.get_dut("ble_prov", "examples/provisioning/ble_prov", dut_class=ttfw_idf.ESP32DUT) # Get binary file binary_file = os.path.join(dut1.app.binary_path, "ble_prov.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("ble_prov_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("ble_prov_bin_size", bin_size // 1024) + ttfw_idf.log_performance("ble_prov_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("ble_prov_bin_size", bin_size // 1024) # Upload binary and start testing dut1.start_app() diff --git a/examples/provisioning/manager/wifi_prov_mgr_test.py b/examples/provisioning/manager/wifi_prov_mgr_test.py index 600bb66d0..f305edd57 100644 --- a/examples/provisioning/manager/wifi_prov_mgr_test.py +++ b/examples/provisioning/manager/wifi_prov_mgr_test.py @@ -17,40 +17,25 @@ from __future__ import print_function import re import os -import sys import time -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 - -try: - import esp_prov -except ImportError: - esp_prov_path = os.getenv("IDF_PATH") + "/tools/esp_prov" - if esp_prov_path and esp_prov_path not in sys.path: - sys.path.insert(0, esp_prov_path) - import esp_prov +import ttfw_idf +import esp_prov # Have esp_prov throw exception esp_prov.config_throw_except = True -@IDF.idf_example_test(env_tag="Example_WIFI_BT") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI_BT") def test_examples_wifi_prov_mgr(env, extra_data): # Acquire DUT - dut1 = env.get_dut("wifi_prov_mgr", "examples/provisioning/manager", dut_class=ESP32DUT) + dut1 = env.get_dut("wifi_prov_mgr", "examples/provisioning/manager", dut_class=ttfw_idf.ESP32DUT) # Get binary file binary_file = os.path.join(dut1.app.binary_path, "wifi_prov_mgr.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("wifi_prov_mgr_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("wifi_prov_mgr_bin_size", bin_size // 1024) + ttfw_idf.log_performance("wifi_prov_mgr_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("wifi_prov_mgr_bin_size", bin_size // 1024) # Upload binary and start testing dut1.start_app() diff --git a/examples/provisioning/softap_prov/softap_prov_test.py b/examples/provisioning/softap_prov/softap_prov_test.py index 73d3a5181..9a6ee6858 100644 --- a/examples/provisioning/softap_prov/softap_prov_test.py +++ b/examples/provisioning/softap_prov/softap_prov_test.py @@ -17,48 +17,26 @@ from __future__ import print_function import re import os -import sys import time -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 - -try: - import esp_prov -except ImportError: - esp_prov_path = os.getenv("IDF_PATH") + "/tools/esp_prov" - if esp_prov_path and esp_prov_path not in sys.path: - sys.path.insert(0, esp_prov_path) - import esp_prov - -try: - import wifi_tools -except ImportError: - wifi_tools_path = os.getenv("IDF_PATH") + "/examples/provisioning/softap_prov/utils" - if wifi_tools_path and wifi_tools_path not in sys.path: - sys.path.insert(0, wifi_tools_path) - import wifi_tools +import ttfw_idf +import esp_prov +import wifi_tools # Have esp_prov throw exception esp_prov.config_throw_except = True -@IDF.idf_example_test(env_tag="Example_WIFI_BT") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI_BT") def test_examples_provisioning_softap(env, extra_data): # Acquire DUT - dut1 = env.get_dut("softap_prov", "examples/provisioning/softap_prov", dut_class=ESP32DUT) + dut1 = env.get_dut("softap_prov", "examples/provisioning/softap_prov", dut_class=ttfw_idf.ESP32DUT) # Get binary file binary_file = os.path.join(dut1.app.binary_path, "softap_prov.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("softap_prov_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("softap_prov_bin_size", bin_size // 1024) + ttfw_idf.log_performance("softap_prov_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("softap_prov_bin_size", bin_size // 1024) # Upload binary and start testing dut1.start_app() diff --git a/examples/security/flash_encryption/example_test.py b/examples/security/flash_encryption/example_test.py index a48fac087..5d99a3782 100644 --- a/examples/security/flash_encryption/example_test.py +++ b/examples/security/flash_encryption/example_test.py @@ -1,15 +1,5 @@ from __future__ import print_function -import os -import sys - -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf # To prepare a test runner for this example: @@ -19,9 +9,9 @@ except ImportError: # espefuse.py --do-not-confirm -p $ESPPORT burn_efuse FLASH_CRYPT_CONFIG 0xf # espefuse.py --do-not-confirm -p $ESPPORT burn_efuse FLASH_CRYPT_CNT 0x1 # espefuse.py --do-not-confirm -p $ESPPORT burn_key flash_encryption key.bin -@IDF.idf_example_test(env_tag='Example_Flash_Encryption') +@ttfw_idf.idf_example_test(env_tag='Example_Flash_Encryption') def test_examples_security_flash_encryption(env, extra_data): - dut = env.get_dut('flash_encryption', 'examples/security/flash_encryption', dut_class=ESP32DUT) + dut = env.get_dut('flash_encryption', 'examples/security/flash_encryption', dut_class=ttfw_idf.ESP32DUT) # start test dut.start_app() lines = [ diff --git a/examples/storage/ext_flash_fatfs/example_test.py b/examples/storage/ext_flash_fatfs/example_test.py index a0696d057..3fa133bb5 100644 --- a/examples/storage/ext_flash_fatfs/example_test.py +++ b/examples/storage/ext_flash_fatfs/example_test.py @@ -1,21 +1,11 @@ from __future__ import print_function -import os -import sys -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 - -from IDF.IDFDUT import ESP32DUT +import ttfw_idf -@IDF.idf_example_test(env_tag='Example_ExtFlash') +@ttfw_idf.idf_example_test(env_tag='Example_ExtFlash') def test_examples_storage_ext_flash_fatfs(env, extra_data): - dut = env.get_dut('ext_flash_fatfs', 'examples/storage/ext_flash_fatfs', dut_class=ESP32DUT) + dut = env.get_dut('ext_flash_fatfs', 'examples/storage/ext_flash_fatfs', dut_class=ttfw_idf.ESP32DUT) dut.start_app() dut.expect('Initialized external Flash') diff --git a/examples/storage/parttool/example_test.py b/examples/storage/parttool/example_test.py index 8b9d37775..c75faf456 100644 --- a/examples/storage/parttool/example_test.py +++ b/examples/storage/parttool/example_test.py @@ -3,19 +3,12 @@ import os import sys import subprocess -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_idf.idf_example_test(env_tag='Example_WIFI') def test_examples_parttool(env, extra_data): - dut = env.get_dut('parttool', 'examples/storage/parttool', dut_class=ESP32DUT) + dut = env.get_dut('parttool', 'examples/storage/parttool', dut_class=ttfw_idf.ESP32DUT) dut.start_app(False) # Verify factory firmware diff --git a/examples/storage/spiffsgen/example_test.py b/examples/storage/spiffsgen/example_test.py index a2ad4304a..0fe004568 100644 --- a/examples/storage/spiffsgen/example_test.py +++ b/examples/storage/spiffsgen/example_test.py @@ -1,22 +1,14 @@ from __future__ import print_function import os -import sys import hashlib -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_idf.idf_example_test(env_tag='Example_WIFI') def test_examples_spiffsgen(env, extra_data): # Test with default build configurations - dut = env.get_dut('spiffsgen', 'examples/storage/spiffsgen', dut_class=ESP32DUT) + dut = env.get_dut('spiffsgen', 'examples/storage/spiffsgen', dut_class=ttfw_idf.ESP32DUT) dut.start_app() base_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'spiffs_image') diff --git a/examples/system/console/example_test.py b/examples/system/console/example_test.py index 23b25ef00..87cc8e392 100644 --- a/examples/system/console/example_test.py +++ b/examples/system/console/example_test.py @@ -1,17 +1,9 @@ from __future__ import print_function -import os -import sys -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 ttfw_idf -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_idf.idf_example_test(env_tag='Example_WIFI') def test_examples_system_console(env, extra_data): dut = env.get_dut('console_example', 'examples/system/console', app_config_name='history') print("Using binary path: {}".format(dut.app.binary_path)) diff --git a/examples/system/cpp_exceptions/example_test.py b/examples/system/cpp_exceptions/example_test.py index 210145836..8340387c9 100644 --- a/examples/system/cpp_exceptions/example_test.py +++ b/examples/system/cpp_exceptions/example_test.py @@ -1,20 +1,11 @@ from __future__ import print_function -import os -import sys -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_idf.idf_example_test(env_tag='Example_WIFI') def test_examples_system_cpp_exceptions(env, extra_data): - dut = env.get_dut('cpp_exceptions_example', 'examples/system/cpp_exceptions', dut_class=ESP32DUT) + dut = env.get_dut('cpp_exceptions_example', 'examples/system/cpp_exceptions', dut_class=ttfw_idf.ESP32DUT) # start test dut.start_app() lines = ['app_main starting', diff --git a/examples/system/cpp_rtti/example_test.py b/examples/system/cpp_rtti/example_test.py index af63c98d4..d384d06c9 100644 --- a/examples/system/cpp_rtti/example_test.py +++ b/examples/system/cpp_rtti/example_test.py @@ -1,23 +1,11 @@ from __future__ import print_function -import os -import sys -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 ttfw_idf -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_idf.idf_example_test(env_tag='Example_WIFI') def test_cpp_rtti_example(env, extra_data): - dut = env.get_dut('cpp_rtti', 'examples/system/cpp_rtti') + dut = env.get_dut('cpp_rtti', 'examples/system/cpp_rtti', dut_class=ttfw_idf.ESP32DUT) dut.start_app() dut.expect('Type name of std::cout is: std::ostream') diff --git a/examples/system/esp_event/default_event_loop/example_test.py b/examples/system/esp_event/default_event_loop/example_test.py index 002c35edd..43a534451 100644 --- a/examples/system/esp_event/default_event_loop/example_test.py +++ b/examples/system/esp_event/default_event_loop/example_test.py @@ -1,18 +1,6 @@ from __future__ import print_function -import os -import sys -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 ttfw_idf # Timer events TIMER_EVENT_LIMIT = 3 @@ -91,7 +79,7 @@ def _test_iteration_events(dut): print("Deleted task event source") -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_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/esp_event/default_event_loop') diff --git a/examples/system/esp_event/user_event_loops/example_test.py b/examples/system/esp_event/user_event_loops/example_test.py index ccd8eeab9..7bb3135bf 100644 --- a/examples/system/esp_event/user_event_loops/example_test.py +++ b/examples/system/esp_event/user_event_loops/example_test.py @@ -1,18 +1,6 @@ from __future__ import print_function -import os -import sys -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 ttfw_idf TASK_ITERATION_LIMIT = 10 @@ -20,9 +8,9 @@ TASK_ITERATION_POSTING = "posting TASK_EVENTS:TASK_ITERATION_EVENT to {}, iterat TASK_ITERATION_HANDLING = "handling TASK_EVENTS:TASK_ITERATION_EVENT from {}, iteration {}" -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_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/esp_event/user_event_loops') + dut = env.get_dut('user_event_loops', 'examples/system/esp_event/user_event_loops', dut_class=ttfw_idf.ESP32DUT) dut.start_app() diff --git a/examples/system/esp_timer/example_test.py b/examples/system/esp_timer/example_test.py index 325088abe..9963bb768 100644 --- a/examples/system/esp_timer/example_test.py +++ b/examples/system/esp_timer/example_test.py @@ -1,20 +1,7 @@ from __future__ import print_function import re -import os -import sys -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf STARTING_TIMERS_REGEX = re.compile(r'Started timers, time since boot: (\d+) us') @@ -38,9 +25,9 @@ LIGHT_SLEEP_TIME = 500000 ONE_SHOT_TIMER_PERIOD = 5000000 -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_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', dut_class=ESP32DUT) + dut = env.get_dut('esp_timer_example', 'examples/system/esp_timer', dut_class=ttfw_idf.ESP32DUT) # start test dut.start_app() groups = dut.expect(STARTING_TIMERS_REGEX, timeout=30) diff --git a/examples/system/freertos/real_time_stats/example_test.py b/examples/system/freertos/real_time_stats/example_test.py index 2d2d2cc0b..d9de7b4b7 100644 --- a/examples/system/freertos/real_time_stats/example_test.py +++ b/examples/system/freertos/real_time_stats/example_test.py @@ -1,27 +1,14 @@ from __future__ import print_function -import os -import sys -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf STATS_TASK_ITERS = 3 STATS_TASK_EXPECT = "Real time stats obtained" -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_idf.idf_example_test(env_tag='Example_WIFI') def test_real_time_stats_example(env, extra_data): - dut = env.get_dut('real_time_stats', 'examples/system/freertos/real_time_stats', dut_class=ESP32DUT) + dut = env.get_dut('real_time_stats', 'examples/system/freertos/real_time_stats', dut_class=ttfw_idf.ESP32DUT) dut.start_app() for iteration in range(0, STATS_TASK_ITERS): diff --git a/examples/system/light_sleep/example_test.py b/examples/system/light_sleep/example_test.py index a29b5dcc8..4c6e42833 100644 --- a/examples/system/light_sleep/example_test.py +++ b/examples/system/light_sleep/example_test.py @@ -1,17 +1,8 @@ from __future__ import print_function import re -import os -import sys import time -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_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') @@ -20,9 +11,9 @@ WAITING_FOR_GPIO_STR = 'Waiting for GPIO0 to go high...' WAKEUP_INTERVAL_MS = 2000 -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_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', dut_class=ESP32DUT) + dut = env.get_dut('light_sleep_example', 'examples/system/light_sleep', dut_class=ttfw_idf.ESP32DUT) dut.start_app() # Ensure DTR and RTS are de-asserted for proper control of GPIO0 diff --git a/examples/system/ota/otatool/example_test.py b/examples/system/ota/otatool/example_test.py index cfbf06450..404f07578 100644 --- a/examples/system/ota/otatool/example_test.py +++ b/examples/system/ota/otatool/example_test.py @@ -3,23 +3,12 @@ import os import sys import subprocess -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 ttfw_idf -@IDF.idf_example_test(env_tag='Example_WIFI') +@ttfw_idf.idf_example_test(env_tag='Example_WIFI') def test_otatool_example(env, extra_data): - dut = env.get_dut('otatool', 'examples/system/ota/otatool', dut_class=ESP32DUT) + dut = env.get_dut('otatool', 'examples/system/ota/otatool', dut_class=ttfw_idf.ESP32DUT) # Verify factory firmware dut.start_app() diff --git a/examples/system/ota/simple_ota_example/example_test.py b/examples/system/ota/simple_ota_example/example_test.py index 0b79b68b5..4ea3045ad 100644 --- a/examples/system/ota/simple_ota_example/example_test.py +++ b/examples/system/ota/simple_ota_example/example_test.py @@ -1,26 +1,13 @@ import re import os -import sys import socket import BaseHTTPServer import SimpleHTTPServer from threading import Thread import ssl -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import DUT +import ttfw_idf server_cert = "-----BEGIN CERTIFICATE-----\n" \ "MIIDXTCCAkWgAwIBAgIJAP4LF7E72HakMA0GCSqGSIb3DQEBCwUAMEUxCzAJBgNV\n"\ @@ -108,7 +95,7 @@ def start_https_server(ota_image_dir, server_ip, server_port): httpd.serve_forever() -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_simple_ota_example(env, extra_data): """ steps: | @@ -116,12 +103,12 @@ def test_examples_protocol_simple_ota_example(env, extra_data): 2. Fetch OTA image over HTTPS 3. Reboot with the new OTA image """ - dut1 = env.get_dut("simple_ota_example", "examples/system/ota/simple_ota_example", dut_class=ESP32DUT) + dut1 = env.get_dut("simple_ota_example", "examples/system/ota/simple_ota_example", dut_class=ttfw_idf.ESP32DUT) # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "simple_ota.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("simple_ota_bin_size", "{}KB".format(bin_size // 1024)) - IDF.check_performance("simple_ota_bin_size", bin_size // 1024) + ttfw_idf.log_performance("simple_ota_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("simple_ota_bin_size", bin_size // 1024) # start test host_ip = get_my_ip() thread1 = Thread(target=start_https_server, args=(dut1.app.binary_path, host_ip, 8000)) diff --git a/examples/wifi/iperf/iperf_test.py b/examples/wifi/iperf/iperf_test.py index b6af1a62a..074cc7a5f 100644 --- a/examples/wifi/iperf/iperf_test.py +++ b/examples/wifi/iperf/iperf_test.py @@ -26,34 +26,13 @@ from builtins import range from builtins import object import re import os -import sys import time import subprocess -try: - import IDF - from IDF.IDFDUT import ESP32DUT -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 +from tiny_test_fw import TinyFW, DUT, Utility +import ttfw_idf +from idf_iperf_test_util import (Attenuator, PowerControl, LineChart, TestReport) -import DUT -import TinyFW -import Utility -from Utility import (Attenuator, PowerControl, LineChart) - -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 @@ -166,8 +145,8 @@ class TestResult(object): throughput = 0.0 if throughput == 0 and rssi > self.ZERO_THROUGHPUT_THRESHOLD: - self.error_list.append("[Error][Fatal][{}][att: {}][rssi: {}]: No throughput data found" - .format(ap_ssid, att, rssi)) + self.error_list.append("[Error][Fatal][{}][att: {}][rssi: {}]: No throughput data found" + .format(ap_ssid, att, rssi)) self._save_result(throughput, ap_ssid, att, rssi, heap_size) @@ -467,7 +446,7 @@ class IperfTestUtility(object): return ret -@IDF.idf_example_test(env_tag="Example_ShieldBox_Basic", category="stress") +@ttfw_idf.idf_example_test(env_tag="Example_ShieldBox_Basic", category="stress") def test_wifi_throughput_with_different_configs(env, extra_data): """ steps: | @@ -492,7 +471,8 @@ def test_wifi_throughput_with_different_configs(env, extra_data): "sdkconfig.ci.{}".format(config_name)) # 2. get DUT and download - dut = env.get_dut("iperf", "examples/wifi/iperf", dut_class=ESP32DUT, app_config_name=config_name) + dut = env.get_dut("iperf", "examples/wifi/iperf", dut_class=ttfw_idf.ESP32DUT, + app_config_name=config_name) dut.start_app() dut.expect("esp32>") @@ -519,12 +499,12 @@ def test_wifi_throughput_with_different_configs(env, extra_data): env.close_dut("iperf") # 5. generate report - report = ThroughputForConfigsReport(os.path.join(env.log_path, "ThroughputForConfigsReport"), - ap_info["ssid"], test_result, sdkconfig_files) + report = TestReport.ThroughputForConfigsReport(os.path.join(env.log_path, "ThroughputForConfigsReport"), + ap_info["ssid"], test_result, sdkconfig_files) report.generate_report() -@IDF.idf_example_test(env_tag="Example_ShieldBox", category="stress") +@ttfw_idf.idf_example_test(env_tag="Example_ShieldBox", category="stress") def test_wifi_throughput_vs_rssi(env, extra_data): """ steps: | @@ -547,7 +527,8 @@ def test_wifi_throughput_vs_rssi(env, extra_data): } # 1. get DUT and download - dut = env.get_dut("iperf", "examples/wifi/iperf", dut_class=ESP32DUT, app_config_name=BEST_PERFORMANCE_CONFIG) + dut = env.get_dut("iperf", "examples/wifi/iperf", dut_class=ttfw_idf.ESP32DUT, + app_config_name=BEST_PERFORMANCE_CONFIG) dut.start_app() dut.expect("esp32>") @@ -573,12 +554,12 @@ def test_wifi_throughput_vs_rssi(env, extra_data): env.close_dut("iperf") # 4. generate report - report = ThroughputVsRssiReport(os.path.join(env.log_path, "ThroughputVsRssiReport"), - test_result) + report = TestReport.ThroughputVsRssiReport(os.path.join(env.log_path, "ThroughputVsRssiReport"), + test_result) report.generate_report() -@IDF.idf_example_test(env_tag="Example_ShieldBox_Basic") +@ttfw_idf.idf_example_test(env_tag="Example_ShieldBox_Basic") def test_wifi_throughput_basic(env, extra_data): """ steps: | @@ -593,7 +574,8 @@ def test_wifi_throughput_basic(env, extra_data): } # 1. get DUT - dut = env.get_dut("iperf", "examples/wifi/iperf", dut_class=ESP32DUT, app_config_name=BEST_PERFORMANCE_CONFIG) + dut = env.get_dut("iperf", "examples/wifi/iperf", dut_class=ttfw_idf.ESP32DUT, + app_config_name=BEST_PERFORMANCE_CONFIG) dut.start_app() dut.expect("esp32>") @@ -615,8 +597,8 @@ def test_wifi_throughput_basic(env, extra_data): # 4. log performance and compare with pass standard performance_items = [] for throughput_type in test_result: - IDF.log_performance("{}_throughput".format(throughput_type), - "{:.02f} Mbps".format(test_result[throughput_type].get_best_throughput())) + ttfw_idf.log_performance("{}_throughput".format(throughput_type), + "{:.02f} Mbps".format(test_result[throughput_type].get_best_throughput())) performance_items.append(["{}_throughput".format(throughput_type), "{:.02f} Mbps".format(test_result[throughput_type].get_best_throughput())]) @@ -624,8 +606,8 @@ def test_wifi_throughput_basic(env, extra_data): TinyFW.JunitReport.update_performance(performance_items) # do check after logging, otherwise test will exit immediately if check fail, some performance can't be logged. for throughput_type in test_result: - IDF.check_performance("{}_throughput".format(throughput_type), - test_result[throughput_type].get_best_throughput()) + ttfw_idf.check_performance("{}_throughput".format(throughput_type), + test_result[throughput_type].get_best_throughput()) env.close_dut("iperf") diff --git a/tools/ble/__init__.py b/tools/ble/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/ci/config/assign-test.yml b/tools/ci/config/assign-test.yml index ffc7c9f27..9846b8d00 100644 --- a/tools/ci/config/assign-test.yml +++ b/tools/ci/config/assign-test.yml @@ -10,7 +10,6 @@ assign_test: - build_ssc - build_esp_idf_tests_cmake variables: - TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw" EXAMPLE_CONFIG_OUTPUT_PATH: "$CI_PROJECT_DIR/examples/test_configs" UNIT_TEST_CASE_FILE: "${CI_PROJECT_DIR}/components/idf_test/unit_test/TestCaseAll.yml" artifacts: @@ -18,6 +17,7 @@ assign_test: - components/idf_test/*/CIConfigs - components/idf_test/*/TC.sqlite - $EXAMPLE_CONFIG_OUTPUT_PATH + - build_examples/artifact_index.json expire_in: 1 week only: variables: @@ -27,9 +27,9 @@ assign_test: - $BOT_LABEL_EXAMPLE_TEST script: # assign example tests - - python $TEST_FW_PATH/CIAssignExampleTest.py $IDF_PATH/examples $CI_TARGET_TEST_CONFIG_FILE $EXAMPLE_CONFIG_OUTPUT_PATH + - python tools/ci/python_packages/ttfw_idf/CIAssignExampleTest.py $IDF_PATH/examples $CI_TARGET_TEST_CONFIG_FILE $EXAMPLE_CONFIG_OUTPUT_PATH # assign unit test cases - - python $TEST_FW_PATH/CIAssignUnitTest.py $UNIT_TEST_CASE_FILE $CI_TARGET_TEST_CONFIG_FILE $IDF_PATH/components/idf_test/unit_test/CIConfigs + - python tools/ci/python_packages/ttfw_idf/CIAssignUnitTest.py $UNIT_TEST_CASE_FILE $CI_TARGET_TEST_CONFIG_FILE $IDF_PATH/components/idf_test/unit_test/CIConfigs # clone test script to assign tests - git clone $TEST_SCRIPT_REPOSITORY - python $CHECKOUT_REF_SCRIPT auto_test_script auto_test_script @@ -57,7 +57,6 @@ update_test_cases: variables: UNIT_TEST_CASE_FILE: "${CI_PROJECT_DIR}/components/idf_test/unit_test/TestCaseAll.yml" BOT_ACCOUNT_CONFIG_FILE: "${CI_PROJECT_DIR}/test-management/Config/Account.local.yml" - TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw" AUTO_TEST_SCRIPT_PATH: "${CI_PROJECT_DIR}/auto_test_script" PYTHON_VER: 3 script: diff --git a/tools/ci/config/target-test.yml b/tools/ci/config/target-test.yml index 47044a1c5..baa30916f 100644 --- a/tools/ci/config/target-test.yml +++ b/tools/ci/config/target-test.yml @@ -30,7 +30,6 @@ reports: junit: $LOG_PATH/*/XUNIT_RESULT.xml variables: - TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw" TEST_CASE_PATH: "$CI_PROJECT_DIR/examples" CONFIG_FILE_PATH: "${CI_PROJECT_DIR}/examples/test_configs" LOG_PATH: "$CI_PROJECT_DIR/TEST_LOGS" @@ -46,7 +45,7 @@ # Get esptool - git submodule sync - git submodule update --init components/esptool_py/esptool - - cd $TEST_FW_PATH + - cd tools/ci/python_packages/tiny_test_fw/bin # run test - python Runner.py $TEST_CASE_PATH -c $CONFIG_FILE -e $ENV_FILE @@ -75,7 +74,6 @@ reports: junit: $LOG_PATH/*/XUNIT_RESULT.xml variables: - TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw" TEST_CASE_PATH: "$CI_PROJECT_DIR/examples" CONFIG_FILE_PATH: "${CI_PROJECT_DIR}/examples/test_configs" LOG_PATH: "$CI_PROJECT_DIR/TEST_LOGS" @@ -87,7 +85,7 @@ # clone test env configs - git clone $TEST_ENV_CONFIG_REPOSITORY - python $CHECKOUT_REF_SCRIPT ci-test-runner-configs ci-test-runner-configs - - cd $TEST_FW_PATH + - cd tools/ci/python_packages/tiny_test_fw/bin # run test - python Runner.py $TEST_CASE_PATH -c $CONFIG_FILE -e $ENV_FILE @@ -170,7 +168,6 @@ test_weekend_mqtt: - $BOT_LABEL_WEEKEND_TEST variables: TEST_CASE_PATH: "$CI_PROJECT_DIR/components/mqtt/weekend_test" - TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw" LOG_PATH: "$CI_PROJECT_DIR/TEST_LOGS" ENV_FILE: "$CI_PROJECT_DIR/components/mqtt/weekend_test/env.yml" CONFIG_FILE_PATH: "$CI_PROJECT_DIR/components/mqtt/weekend_test" @@ -187,7 +184,6 @@ test_weekend_network: - $BOT_LABEL_WEEKEND_TEST variables: TEST_CASE_PATH: "$CI_PROJECT_DIR/components/lwip/weekend_test" - TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw" LOG_PATH: "$CI_PROJECT_DIR/TEST_LOGS" ENV_FILE: "$CI_PROJECT_DIR/components/lwip/weekend_test/env.yml" CONFIG_FILE_PATH: "$CI_PROJECT_DIR/components/lwip/weekend_test" @@ -214,7 +210,7 @@ example_test_002: # clone test env configs - git clone $TEST_ENV_CONFIG_REPOSITORY - python $CHECKOUT_REF_SCRIPT ci-test-runner-configs ci-test-runner-configs - - cd $TEST_FW_PATH + - cd tools/ci/python_packages/tiny_test_fw/bin # run test - python Runner.py $TEST_CASE_PATH -c $CONFIG_FILE -e $ENV_FILE diff --git a/tools/ci/python_packages/idf_http_server_test/__init__.py b/tools/ci/python_packages/idf_http_server_test/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/examples/protocols/http_server/persistent_sockets/scripts/adder.py b/tools/ci/python_packages/idf_http_server_test/adder.py similarity index 99% rename from examples/protocols/http_server/persistent_sockets/scripts/adder.py rename to tools/ci/python_packages/idf_http_server_test/adder.py index a1c9c990f..46eb75a55 100644 --- a/examples/protocols/http_server/persistent_sockets/scripts/adder.py +++ b/tools/ci/python_packages/idf_http_server_test/adder.py @@ -20,7 +20,8 @@ from builtins import str from builtins import range import http.client import argparse -import Utility + +from tiny_test_fw import Utility def start_session(ip, port): diff --git a/examples/protocols/http_server/simple/scripts/client.py b/tools/ci/python_packages/idf_http_server_test/client.py similarity index 96% rename from examples/protocols/http_server/simple/scripts/client.py rename to tools/ci/python_packages/idf_http_server_test/client.py index 5e070b98b..95d195d82 100644 --- a/examples/protocols/http_server/simple/scripts/client.py +++ b/tools/ci/python_packages/idf_http_server_test/client.py @@ -20,19 +20,8 @@ from builtins import str import http.client import argparse -try: - import Utility -except ImportError: - import sys - import os - # This environment variable is expected on the host machine - # > export TEST_FW_PATH=~/esp/esp-idf/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 Utility +from tiny_test_fw import Utility def verbose_print(verbosity, *args): diff --git a/tools/ci/python_packages/idf_http_server_test/test.py b/tools/ci/python_packages/idf_http_server_test/test.py new file mode 100644 index 000000000..3ad5bcc9c --- /dev/null +++ b/tools/ci/python_packages/idf_http_server_test/test.py @@ -0,0 +1,1025 @@ +#!/usr/bin/env python +# +# Copyright 2018 Espressif Systems (Shanghai) PTE LTD +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Utility for testing the web server. Test cases: +# Assume the device supports 'n' simultaneous open sockets +# +# HTTP Server Tests +# +# 0. Firmware Settings: +# - Create a dormant thread whose sole job is to call httpd_stop() when instructed +# - Measure the following before httpd_start() is called: +# - current free memory +# - current free sockets +# - Measure the same whenever httpd_stop is called +# - Register maximum possible URI handlers: should be successful +# - Register one more URI handler: should fail +# - Deregister on URI handler: should be successful +# - Register on more URI handler: should succeed +# - Register separate handlers for /hello, /hello/type_html. Also +# ensure that /hello/type_html is registered BEFORE /hello. (tests +# that largest matching URI is picked properly) +# - Create URI handler /adder. Make sure it uses a custom free_ctx +# structure to free it up + +# 1. Using Standard Python HTTP Client +# - simple GET on /hello (returns Hello World. Ensures that basic +# firmware tests are complete, or returns error) +# - POST on /hello (should fail) +# - PUT on /hello (should fail) +# - simple POST on /echo (returns whatever the POST data) +# - simple PUT on /echo (returns whatever the PUT data) +# - GET on /echo (should fail) +# - simple GET on /hello/type_html (returns Content type as text/html) +# - simple GET on /hello/status_500 (returns HTTP status 500) +# - simple GET on /false_uri (returns HTTP status 404) +# - largest matching URI handler is picked is already verified because +# of /hello and /hello/type_html tests +# +# +# 2. Session Tests +# - Sessions + Pipelining basics: +# - Create max supported sessions +# - On session i, +# - send 3 back-to-back POST requests with data i on /adder +# - read back 3 responses. They should be i, 2i and 3i +# - Tests that +# - pipelining works +# - per-session context is maintained for all supported +# sessions +# - Close all sessions +# +# - Cleanup leftover data: Tests that the web server properly cleans +# up leftover data +# - Create a session +# - POST on /leftover_data with 52 bytes of data (data includes +# \r\n)(the handler only +# reads first 10 bytes and returns them, leaving the rest of the +# bytes unread) +# - GET on /hello (should return 'Hello World') +# - POST on /false_uri with 52 bytes of data (data includes \r\n) +# (should return HTTP 404) +# - GET on /hello (should return 'Hello World') +# +# - Test HTTPd Asynchronous response +# - Create a session +# - GET on /async_data +# - returns 'Hello World!' as a response +# - the handler schedules an async response, which generates a second +# response 'Hello Double World!' +# +# - Spillover test +# - Create max supported sessions with the web server +# - GET /hello on all the sessions (should return Hello World) +# - Create one more session, this should fail +# - GET /hello on all the sessions (should return Hello World) +# +# - Timeout test +# - Create a session and only Send 'GE' on the same (simulates a +# client that left the network halfway through a request) +# - Wait for recv-wait-timeout +# - Server should automatically close the socket + + +# ############ TODO TESTS ############# + +# 3. Stress Tests +# +# - httperf +# - Run the following httperf command: +# httperf --server=10.31.130.126 --wsess=8,50,0.5 --rate 8 --burst-length 2 +# +# - The above implies that the test suite will open +# - 8 simultaneous connections with the server +# - the rate of opening the sessions will be 8 per sec. So in our +# case, a new connection will be opened every 0.2 seconds for 1 second +# - The burst length 2 indicates that 2 requests will be sent +# simultaneously on the same connection in a single go +# - 0.5 seconds is the time between sending out 2 bursts +# - 50 is the total number of requests that will be sent out +# +# - So in the above example, the test suite will open 8 +# connections, each separated by 0.2 seconds. On each connection +# it will send 2 requests in a single burst. The bursts on a +# single connection will be separated by 0.5 seconds. A total of +# 25 bursts (25 x 2 = 50) will be sent out + +# 4. Leak Tests +# - Simple Leak test +# - Simple GET on /hello/restart (returns success, stop web server, measures leaks, restarts webserver) +# - Simple GET on /hello/restart_results (returns the leak results) +# - Leak test with open sockets +# - Open 8 sessions +# - Simple GET on /hello/restart (returns success, stop web server, +# measures leaks, restarts webserver) +# - All sockets should get closed +# - Simple GET on /hello/restart_results (returns the leak results) + + +from __future__ import division +from __future__ import print_function +from builtins import str +from builtins import range +from builtins import object +import threading +import socket +import time +import argparse +import http.client +import sys +import string +import random + +from tiny_test_fw import Utility + +_verbose_ = False + + +class Session(object): + 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 = '' + self.content_type = '' + self.content_len = 0 + + def send_err_check(self, request, data=None): + rval = True + try: + self.client.sendall(request.encode()) + if data: + self.client.sendall(data.encode()) + except socket.error as err: + self.client.close() + Utility.console_log("Socket Error in send :", err) + rval = False + return rval + + def send_get(self, path, headers=None): + 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\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 + if headers: + for field, value in headers.items(): + 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 + if headers: + for field, value in headers.items(): + 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): + try: + state = 'nothing' + resp_read = '' + while True: + char = self.client.recv(1).decode() + if char == '\r' and state == 'nothing': + state = 'first_cr' + elif char == '\n' and state == 'first_cr': + state = 'first_lf' + elif char == '\r' and state == 'first_lf': + state = 'second_cr' + elif char == '\n' and state == 'second_cr': + state = 'second_lf' + else: + state = 'nothing' + resp_read += char + if state == 'second_lf': + break + # Handle first line + line_hdrs = resp_read.splitlines() + line_comp = line_hdrs[0].split() + self.status = line_comp[1] + del line_hdrs[0] + self.encoding = '' + self.content_type = '' + headers = dict() + # Process other headers + for h in range(len(line_hdrs)): + line_comp = line_hdrs[h].split(':') + if line_comp[0] == 'Content-Length': + self.content_len = int(line_comp[1]) + if line_comp[0] == 'Content-Type': + self.content_type = line_comp[1].lstrip() + if line_comp[0] == 'Transfer-Encoding': + self.encoding = line_comp[1].lstrip() + if len(line_comp) == 2: + headers[line_comp[0]] = line_comp[1].lstrip() + return headers + except socket.error as err: + self.client.close() + Utility.console_log("Socket Error in recv :", err) + return None + + def read_resp_data(self): + try: + read_data = '' + if self.encoding != 'chunked': + while len(read_data) != self.content_len: + read_data += self.client.recv(self.content_len).decode() + else: + chunk_data_buf = '' + while (True): + # Read one character into temp buffer + read_ch = self.client.recv(1) + # Check CRLF + if (read_ch == '\r'): + read_ch = self.client.recv(1).decode() + if (read_ch == '\n'): + # If CRLF decode length of chunk + chunk_len = int(chunk_data_buf, 16) + # Keep adding to contents + self.content_len += chunk_len + rem_len = chunk_len + while (rem_len): + new_data = self.client.recv(rem_len) + read_data += new_data + rem_len -= len(new_data) + chunk_data_buf = '' + # Fetch remaining CRLF + if self.client.recv(2) != "\r\n": + # Error in packet + Utility.console_log("Error in chunked data") + return None + if not chunk_len: + # If last chunk + break + continue + chunk_data_buf += '\r' + # If not CRLF continue appending + # character to chunked data buffer + chunk_data_buf += read_ch + return read_data + except socket.error as err: + self.client.close() + Utility.console_log("Socket Error in recv :", err) + return None + + def close(self): + self.client.close() + + +def test_val(text, expected, received): + if expected != received: + Utility.console_log(" Fail!") + Utility.console_log(" [reason] " + text + ":") + Utility.console_log(" expected: " + str(expected)) + Utility.console_log(" received: " + str(received)) + return False + return True + + +class adder_thread (threading.Thread): + def __init__(self, id, dut, port): + threading.Thread.__init__(self) + self.id = id + self.dut = dut + self.depth = 3 + self.session = Session(dut, port) + + def run(self): + self.response = [] + + # Pipeline 3 requests + if (_verbose_): + Utility.console_log(" Thread: Using adder start " + str(self.id)) + + for _ in range(self.depth): + self.session.send_post('/adder', str(self.id)) + time.sleep(2) + + for _ in range(self.depth): + self.session.read_resp_hdrs() + self.response.append(self.session.read_resp_data()) + + def adder_result(self): + if len(self.response) != self.depth: + Utility.console_log("Error : missing response packets") + return False + for i in range(len(self.response)): + if not test_val("Thread" + str(self.id) + " response[" + str(i) + "]", + str(self.id * (i + 1)), str(self.response[i])): + return False + return True + + 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 = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("GET", "/hello") + resp = conn.getresponse() + if not test_val("status_code", 200, resp.status): + conn.close() + return False + if not test_val("data", "Hello World!", resp.read().decode()): + conn.close() + return False + if not test_val("data", "text/html", resp.getheader('Content-Type')): + conn.close() + return False + Utility.console_log("Success") + conn.close() + return True + + +def put_hello(dut, port): + # PUT /hello returns 405' + Utility.console_log("[test] PUT /hello returns 405 =>", end=' ') + conn = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("PUT", "/hello", "Hello") + resp = conn.getresponse() + if not test_val("status_code", 405, resp.status): + conn.close() + return False + Utility.console_log("Success") + conn.close() + return True + + +def post_hello(dut, port): + # POST /hello returns 405' + Utility.console_log("[test] POST /hello returns 405 =>", end=' ') + conn = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("POST", "/hello", "Hello") + resp = conn.getresponse() + if not test_val("status_code", 405, resp.status): + conn.close() + return False + Utility.console_log("Success") + conn.close() + return True + + +def post_echo(dut, port): + # POST /echo echoes data' + Utility.console_log("[test] POST /echo echoes data =>", end=' ') + conn = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("POST", "/echo", "Hello") + resp = conn.getresponse() + if not test_val("status_code", 200, resp.status): + conn.close() + return False + if not test_val("data", "Hello", resp.read().decode()): + conn.close() + return False + Utility.console_log("Success") + conn.close() + return True + + +def put_echo(dut, port): + # PUT /echo echoes data' + Utility.console_log("[test] PUT /echo echoes data =>", end=' ') + conn = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("PUT", "/echo", "Hello") + resp = conn.getresponse() + if not test_val("status_code", 200, resp.status): + conn.close() + return False + if not test_val("data", "Hello", resp.read().decode()): + conn.close() + return False + Utility.console_log("Success") + conn.close() + return True + + +def get_echo(dut, port): + # GET /echo returns 404' + Utility.console_log("[test] GET /echo returns 405 =>", end=' ') + conn = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("GET", "/echo") + resp = conn.getresponse() + if not test_val("status_code", 405, resp.status): + conn.close() + return False + Utility.console_log("Success") + conn.close() + return True + + +def get_test_headers(dut, port): + # GET /test_header returns data of Header2' + Utility.console_log("[test] GET /test_header =>", end=' ') + conn = http.client.HTTPConnection(dut, int(port), timeout=15) + custom_header = {"Header1": "Value1", "Header3": "Value3"} + header2_values = ["", " ", "Value2", " Value2", "Value2 ", " Value2 "] + for val in header2_values: + custom_header["Header2"] = val + conn.request("GET", "/test_header", headers=custom_header) + resp = conn.getresponse() + if not test_val("status_code", 200, resp.status): + conn.close() + return False + hdr_val_start_idx = val.find("Value2") + if hdr_val_start_idx == -1: + if not test_val("header: Header2", "", resp.read().decode()): + conn.close() + return False + else: + if not test_val("header: Header2", val[hdr_val_start_idx:], resp.read().decode()): + conn.close() + return False + resp.read() + Utility.console_log("Success") + 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 = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("GET", "/hello/type_html") + resp = conn.getresponse() + if not test_val("status_code", 200, resp.status): + conn.close() + return False + if not test_val("data", "Hello World!", resp.read().decode()): + conn.close() + return False + if not test_val("data", "text/html", resp.getheader('Content-Type')): + conn.close() + return False + Utility.console_log("Success") + 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 = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("GET", "/hello/status_500") + resp = conn.getresponse() + if not test_val("status_code", 500, resp.status): + conn.close() + return False + Utility.console_log("Success") + 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 = http.client.HTTPConnection(dut, int(port), timeout=15) + conn.request("GET", "/false_uri") + resp = conn.getresponse() + if not test_val("status_code", 404, resp.status): + conn.close() + return False + Utility.console_log("Success") + 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=' ') + t = [] + # Create all sessions + for i in range(max_sessions): + t.append(adder_thread(i, dut, port)) + + for i in range(len(t)): + t[i].start() + + for i in range(len(t)): + t[i].join() + + res = True + for i in range(len(t)): + if not test_val("Thread" + str(i) + " Failed", t[i].adder_result(), True): + res = False + t[i].close() + if (res): + 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("[test] Test HTTPD Work Queue (Async response) =>", end=' ') + s = Session(dut, port) + + s.send_get('/async_data') + s.read_resp_hdrs() + if not test_val("First Response", "Hello World!", s.read_resp_data()): + s.close() + return False + s.read_resp_hdrs() + if not test_val("Second Response", "Hello Double World!", s.read_resp_data()): + s.close() + return False + s.close() + 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=' ') + s = http.client.HTTPConnection(dut + ":" + port, timeout=15) + + s.request("POST", url='/leftover_data', body="abcdefghijklmnopqrstuvwxyz\r\nabcdefghijklmnopqrstuvwxyz") + resp = s.getresponse() + if not test_val("Partial data", "abcdefghij", resp.read().decode()): + s.close() + return False + + s.request("GET", url='/hello') + resp = s.getresponse() + if not test_val("Hello World Data", "Hello World!", resp.read().decode()): + s.close() + return False + + s.request("POST", url='/false_uri', body="abcdefghijklmnopqrstuvwxyz\r\nabcdefghijklmnopqrstuvwxyz") + resp = s.getresponse() + if not test_val("False URI Status", str(404), str(resp.status)): + s.close() + return False + # socket would have been closed by server due to error + s.close() + + s = http.client.HTTPConnection(dut + ":" + port, timeout=15) + s.request("GET", url='/hello') + resp = s.getresponse() + if not test_val("Hello World Data", "Hello World!", resp.read().decode()): + s.close() + return False + + s.close() + 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=' ') + s = [] + _verbose_ = True + for i in range(max_sess + 1): + if (_verbose_): + Utility.console_log("Executing " + str(i)) + try: + a = http.client.HTTPConnection(dut + ":" + port, timeout=15) + a.request("GET", url='/hello') + resp = a.getresponse() + if not test_val("Connection " + str(i), "Hello World!", resp.read().decode()): + a.close() + break + s.append(a) + except Exception: + if (_verbose_): + Utility.console_log("Connection " + str(i) + " rejected") + a.close() + break + + # Close open connections + for a in s: + a.close() + + # Check if number of connections is equal to max_sess + 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) + s.client.sendall(b"GE") + s.read_resp_hdrs() + resp = s.read_resp_data() + if not test_val("Request Timeout", "Server closed this connection", resp): + s.close() + return False + s.close() + 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 + while (retry): + 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))) + 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 + else: + Utility.console_log("Unexpected error") + s.close() + Utility.console_log("Retry...") + continue + resp = resp.read().decode() + result = (resp == random_data) + if not result: + test_val("Data size", str(len(random_data)), str(len(resp))) + s.close() + Utility.console_log("Retry...") + continue + s.close() + Utility.console_log("Success") + return True + Utility.console_log("Failed") + return False + + +def arbitrary_termination_test(dut, port): + Utility.console_log("[test] Arbitrary termination test =>", end=' ') + cases = [ + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nCustom: SomeValue\r\n\r\n", + "code": "200", + "header": "SomeValue" + }, + { + "request": "POST /echo HTTP/1.1\nHost: " + dut + "\r\nCustom: SomeValue\r\n\r\n", + "code": "200", + "header": "SomeValue" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\nCustom: SomeValue\r\n\r\n", + "code": "200", + "header": "SomeValue" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nCustom: SomeValue\n\r\n", + "code": "200", + "header": "SomeValue" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nCustom: SomeValue\r\n\n", + "code": "200", + "header": "SomeValue" + }, + { + "request": "POST /echo HTTP/1.1\nHost: " + dut + "\nCustom: SomeValue\n\n", + "code": "200", + "header": "SomeValue" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: 5\n\r\nABCDE", + "code": "200", + "body": "ABCDE" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: 5\r\n\nABCDE", + "code": "200", + "body": "ABCDE" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: 5\n\nABCDE", + "code": "200", + "body": "ABCDE" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: 5\n\n\rABCD", + "code": "200", + "body": "\rABCD" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\r\nCustom: SomeValue\r\r\n\r\r\n", + "code": "400" + }, + { + "request": "POST /echo HTTP/1.1\r\r\nHost: " + dut + "\r\n\r\n", + "code": "400" + }, + { + "request": "POST /echo HTTP/1.1\r\n\rHost: " + dut + "\r\n\r\n", + "code": "400" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\rCustom: SomeValue\r\n", + "code": "400" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nCustom: Some\rValue\r\n", + "code": "400" + }, + { + "request": "POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nCustom- SomeValue\r\n\r\n", + "code": "400" + } + ] + for case in cases: + s = Session(dut, port) + s.client.sendall((case['request']).encode()) + resp_hdrs = s.read_resp_hdrs() + resp_body = s.read_resp_data() + s.close() + if not test_val("Response Code", case["code"], s.status): + return False + if "header" in case.keys(): + resp_hdr_val = None + if "Custom" in resp_hdrs.keys(): + resp_hdr_val = resp_hdrs["Custom"] + if not test_val("Response Header", case["header"], resp_hdr_val): + return False + if "body" in case.keys(): + if not test_val("Response Body", case["body"], resp_body): + return False + Utility.console_log("Success") + return True + + +def code_500_server_error_test(dut, port): + Utility.console_log("[test] 500 Server Error test =>", end=' ') + s = Session(dut, port) + # Sending a very large content length will cause malloc to fail + content_len = 2**30 + 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() + s.read_resp_data() + if not test_val("Server Error", "500", s.status): + s.close() + return False + s.close() + 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() + 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", "400", s.status): + s.close() + return False + s.close() + 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() + s.read_resp_data() + if not test_val("Server Error", "505", s.status): + s.close() + return False + s.close() + 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() + s.read_resp_data() + if not test_val("Client Error", "400", s.status): + s.close() + return False + s.close() + 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() + s.read_resp_data() + if not test_val("Client Error", "404", s.status): + s.close() + return False + s.close() + 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() + s.read_resp_data() + if not test_val("Client Error", "405", s.status): + s.close() + return False + s.close() + 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() + s.read_resp_data() + if not test_val("Client Error", "408", s.status): + s.close() + return False + s.close() + 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() + 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", "400", s.status): + s.close() + return False + s.close() + 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("/")) + 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() + 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) + if not test_val("Client Error", "404", status): + return False + status = send_getx_uri_len(dut, port, max_uri_len + 1) + if not test_val("Client Error", "414", status): + return False + 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")) + 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]) + time.sleep(1) + s.client.sendall(request[length // 2:]) + hdr = s.read_resp_hdrs() + resp = s.read_resp_data() + s.close() + if hdr and ("Custom" in hdr): + 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) + if not res: + return False + res, status = send_postx_hdr_len(dut, port, max_hdr_len + 1) + if not test_val("Client Error", "431", status): + return False + 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" + 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() + s.read_resp_data() + if not test_val("Client Error", "400", s.status): + s.close() + return False + s.close() + Utility.console_log("Success") + return True + + +if __name__ == '__main__': + # Execution begins here... + # Configuration + # Max number of threads/sessions + max_sessions = 7 + max_uri_len = 512 + max_hdr_len = 512 + + parser = argparse.ArgumentParser(description='Run HTTPD Test') + parser.add_argument('-4','--ipv4', help='IPv4 address') + parser.add_argument('-6','--ipv6', help='IPv6 address') + parser.add_argument('-p','--port', help='Port') + args = vars(parser.parse_args()) + + dut4 = args['ipv4'] + dut6 = args['ipv6'] + port = args['port'] + dut = dut4 + + _verbose_ = True + + Utility.console_log("### Basic HTTP Client Tests") + get_hello(dut, port) + post_hello(dut, port) + put_hello(dut, port) + post_echo(dut, port) + get_echo(dut, port) + put_echo(dut, port) + get_hello_type(dut, port) + get_hello_status(dut, port) + get_false_uri(dut, port) + get_test_headers(dut, port) + + Utility.console_log("### Error code tests") + code_500_server_error_test(dut, port) + code_501_method_not_impl(dut, port) + code_505_version_not_supported(dut, port) + code_400_bad_request(dut, port) + code_404_not_found(dut, port) + code_405_method_not_allowed(dut, port) + code_408_req_timeout(dut, port) + code_414_uri_too_long(dut, port, max_uri_len) + code_431_hdr_too_long(dut, port, max_hdr_len) + test_upgrade_not_supported(dut, port) + + # Not supported yet (Error on chunked request) + # code_411_length_required(dut, port) + + Utility.console_log("### Sessions and Context Tests") + parallel_sessions_adder(dut, port, max_sessions) + leftover_data_test(dut, port) + async_response_test(dut, port) + spillover_session(dut, port, max_sessions) + recv_timeout_test(dut, port) + packet_size_limit_test(dut, port, 50 * 1024) + arbitrary_termination_test(dut, port) + get_hello(dut, port) + + sys.exit() diff --git a/tools/esp_python_packages/tiny_test_fw/Utility/Attenuator.py b/tools/ci/python_packages/idf_iperf_test_util/Attenuator.py similarity index 100% rename from tools/esp_python_packages/tiny_test_fw/Utility/Attenuator.py rename to tools/ci/python_packages/idf_iperf_test_util/Attenuator.py diff --git a/tools/esp_python_packages/tiny_test_fw/Utility/LineChart.py b/tools/ci/python_packages/idf_iperf_test_util/LineChart.py similarity index 100% rename from tools/esp_python_packages/tiny_test_fw/Utility/LineChart.py rename to tools/ci/python_packages/idf_iperf_test_util/LineChart.py diff --git a/tools/esp_python_packages/tiny_test_fw/Utility/PowerControl.py b/tools/ci/python_packages/idf_iperf_test_util/PowerControl.py similarity index 100% rename from tools/esp_python_packages/tiny_test_fw/Utility/PowerControl.py rename to tools/ci/python_packages/idf_iperf_test_util/PowerControl.py diff --git a/examples/wifi/iperf/test_report.py b/tools/ci/python_packages/idf_iperf_test_util/TestReport.py similarity index 100% rename from examples/wifi/iperf/test_report.py rename to tools/ci/python_packages/idf_iperf_test_util/TestReport.py diff --git a/tools/ci/python_packages/idf_iperf_test_util/__init__.py b/tools/ci/python_packages/idf_iperf_test_util/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/ci/python_packages/tiny_test_fw/requirements.txt b/tools/ci/python_packages/tiny_test_fw/requirements.txt index aa6b53b4b..2a742540e 100644 --- a/tools/ci/python_packages/tiny_test_fw/requirements.txt +++ b/tools/ci/python_packages/tiny_test_fw/requirements.txt @@ -2,4 +2,3 @@ pyserial pyyaml junit_xml netifaces -matplotlib diff --git a/examples/provisioning/softap_prov/utils/wifi_tools.py b/tools/ci/python_packages/wifi_tools.py similarity index 100% rename from examples/provisioning/softap_prov/utils/wifi_tools.py rename to tools/ci/python_packages/wifi_tools.py diff --git a/tools/esp_prov/__init__.py b/tools/esp_prov/__init__.py new file mode 100644 index 000000000..cbd5e5b60 --- /dev/null +++ b/tools/esp_prov/__init__.py @@ -0,0 +1 @@ +from esp_prov import * # noqa: export esp_prov module to users diff --git a/tools/unit-test-app/unit_test.py b/tools/unit-test-app/unit_test.py index b30cead1a..c6ed79b74 100755 --- a/tools/unit-test-app/unit_test.py +++ b/tools/unit-test-app/unit_test.py @@ -19,33 +19,12 @@ Test script for unit test case. """ import re -import os -import sys import time import argparse import threading -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) - else: - # or try the copy in IDF - idf_path = os.getenv("IDF_PATH") - tiny_test_path = idf_path + "/tools/tiny-test-fw" - if os.path.exists(tiny_test_path): - sys.path.insert(0, tiny_test_path) - import TinyFW - -import IDF -import Utility -import Env -from DUT import ExpectTimeout -from IDF.IDFApp import UT +from tiny_test_fw import TinyFW, Utility, Env, DUT +import ttfw_idf UT_APP_BOOT_UP_DONE = "Press ENTER to see the list of tests." @@ -180,7 +159,7 @@ def reset_dut(dut): try: dut.expect("0 Tests 0 Failures 0 Ignored", timeout=TEST_HISTORY_CHECK_TIMEOUT) break - except ExpectTimeout: + except DUT.ExpectTimeout: pass else: raise AssertionError("Reset {} ({}) failed!".format(dut.name, dut.port)) @@ -255,14 +234,14 @@ def run_one_normal_case(dut, one_case, junit_test_case): (FINISH_PATTERN, handle_test_finish), (UT_APP_BOOT_UP_DONE, handle_reset_finish), timeout=one_case["timeout"]) - except ExpectTimeout: + except DUT.ExpectTimeout: Utility.console_log("Timeout in expect", color="orange") junit_test_case.add_failure_info("timeout") one_case_finish(False) break -@IDF.idf_unit_test(env_tag="UT_T1_1", junit_report_by_case=True) +@ttfw_idf.idf_unit_test(env_tag="UT_T1_1", junit_report_by_case=True) def run_unit_test_cases(env, extra_data): """ extra_data can be three types of value @@ -401,7 +380,7 @@ class Handler(threading.Thread): time.sleep(1) self.dut.write("\"{}\"".format(self.parent_case_name)) self.dut.expect("Running " + self.parent_case_name + "...") - except ExpectTimeout: + except DUT.ExpectTimeout: Utility.console_log("No case detected!", color="orange") while not self.finish and not self.force_stop.isSet(): try: @@ -411,7 +390,7 @@ class Handler(threading.Thread): (self.SEND_SIGNAL_PATTERN, device_send_action), # send signal pattern (self.FINISH_PATTERN, handle_device_test_finish), # test finish pattern timeout=self.timeout) - except ExpectTimeout: + except DUT.ExpectTimeout: Utility.console_log("Timeout in expect", color="orange") one_device_case_finish(False) break @@ -471,7 +450,7 @@ def run_one_multiple_devices_case(duts, ut_config, env, one_case, app_bin, junit return result -@IDF.idf_unit_test(env_tag="UT_T2_1", junit_report_by_case=True) +@ttfw_idf.idf_unit_test(env_tag="UT_T2_1", junit_report_by_case=True) def run_multiple_devices_cases(env, extra_data): """ extra_data can be two types of value @@ -618,7 +597,7 @@ def run_one_multiple_stage_case(dut, one_case, junit_test_case): (FINISH_PATTERN, handle_test_finish), (UT_APP_BOOT_UP_DONE, handle_next_stage), timeout=one_case["timeout"]) - except ExpectTimeout: + except DUT.ExpectTimeout: Utility.console_log("Timeout in expect", color="orange") one_case_finish(False) break @@ -627,7 +606,7 @@ def run_one_multiple_stage_case(dut, one_case, junit_test_case): break -@IDF.idf_unit_test(env_tag="UT_T1_1", junit_report_by_case=True) +@ttfw_idf.idf_unit_test(env_tag="UT_T1_1", junit_report_by_case=True) def run_multiple_stage_cases(env, extra_data): """ extra_data can be 2 types of value @@ -734,7 +713,7 @@ def detect_update_unit_test_info(env, extra_data, app_bin): for _dic in extra_data: if 'type' not in _dic: raise ValueError("Unit test \"{}\" doesn't exist in the flashed device!".format(_dic.get('name'))) - except ExpectTimeout: + except DUT.ExpectTimeout: Utility.console_log("Timeout during getting the test list", color="red") finally: dut.close() @@ -789,8 +768,8 @@ if __name__ == '__main__': TinyFW.set_default_config(env_config_file=args.env_config_file) env_config = TinyFW.get_default_config() - env_config['app'] = UT - env_config['dut'] = IDF.IDFDUT + env_config['app'] = ttfw_idf.UT + env_config['dut'] = ttfw_idf.IDFDUT env_config['test_suite_name'] = 'unit_test_parsing' test_env = Env.Env(**env_config) detect_update_unit_test_info(test_env, extra_data=list_of_dicts, app_bin=args.app_bin)