diff --git a/components/lwip/weekend_test/net_suite_test.py b/components/lwip/weekend_test/net_suite_test.py index 59e763b44c..41d95c8d66 100644 --- a/components/lwip/weekend_test/net_suite_test.py +++ b/components/lwip/weekend_test/net_suite_test.py @@ -1,27 +1,13 @@ import re import os -import sys import socket from threading import Thread, Event import subprocess import time from shutil import copyfile -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 -import Utility +from tiny_test_fw import Utility, DUT +import ttfw_idf stop_sock_listener = Event() stop_io_listener = Event() @@ -73,7 +59,7 @@ def sock_listener(dut1): sock = None -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def lwip_test_suite(env, extra_data): global stop_io_listener global stop_sock_listener @@ -84,12 +70,12 @@ def lwip_test_suite(env, extra_data): 3. Execute ttcn3 test suite 4. Collect result from ttcn3 """ - dut1 = env.get_dut("net_suite", "examples/system/network_tests", dut_class=ESP32DUT) + dut1 = env.get_dut("net_suite", "examples/system/network_tests", dut_class=ttfw_idf.ESP32DUT) # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "net_suite.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("net_suite", "{}KB".format(bin_size // 1024)) - IDF.check_performance("net_suite", bin_size // 1024) + ttfw_idf.log_performance("net_suite", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("net_suite", bin_size // 1024) dut1.start_app() thread1 = Thread(target=sock_listener, args=(dut1, )) thread2 = Thread(target=io_listener, args=(dut1, )) diff --git a/components/mqtt/weekend_test/mqtt_publish_test.py b/components/mqtt/weekend_test/mqtt_publish_test.py index 55e04074cf..32ae3389f2 100644 --- a/components/mqtt/weekend_test/mqtt_publish_test.py +++ b/components/mqtt/weekend_test/mqtt_publish_test.py @@ -11,20 +11,8 @@ import time import string import random -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() @@ -53,6 +41,8 @@ def mqtt_client_task(client): def get_host_port_from_dut(dut1, config_option): value = re.search(r'\:\/\/([^:]+)\:([0-9]+)', dut1.app.get_sdkconfig()[config_option]) + if value is None: + return None, None return value.group(1), int(value.group(2)) @@ -124,7 +114,7 @@ def test_single_config(dut, transport, qos, repeat, published): event_stop_client.clear() -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_weekend_mqtt_publish(env, extra_data): # Using broker url dictionary for different transport global broker_host @@ -138,13 +128,12 @@ def test_weekend_mqtt_publish(env, extra_data): 3. Test evaluates python client received correct qos0 message 4. Test ESP32 client received correct qos0 message """ - dut1 = env.get_dut("mqtt_publish", "examples/protocols/mqtt/publish_test", dut_class=ESP32DUT) + dut1 = env.get_dut("mqtt_publish", "examples/protocols/mqtt/publish_test") # check and log bin size binary_file = os.path.join(dut1.app.binary_path, "mqtt_publish.bin") bin_size = os.path.getsize(binary_file) - IDF.log_performance("mqtt_publish_bin_size", "{}KB" - .format(bin_size // 1024)) - IDF.check_performance("mqtt_publish_size", bin_size // 1024) + ttfw_idf.log_performance("mqtt_publish_bin_size", "{}KB".format(bin_size // 1024)) + ttfw_idf.check_performance("mqtt_publish_size", bin_size // 1024) # Look for host:port in sdkconfig try: # python client subscribes to the topic to which esp client publishes and vice versa @@ -159,13 +148,16 @@ def test_weekend_mqtt_publish(env, extra_data): raise dut1.start_app() try: - ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30) + ip_address = dut1.expect(re.compile(r" IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"), timeout=30) print("Connected to AP with IP: {}".format(ip_address)) except DUT.ExpectTimeout: print('ENV_TEST_FAILURE: Cannot connect to AP') raise for qos in [0, 1, 2]: for transport in ["tcp", "ssl", "ws", "wss"]: + if broker_host[transport] is None: + print('Skipping transport: {}...'.format(transport)) + continue # simple test with empty message test_single_config(dut1, transport, qos, 0, 5) # decide on broker what level of test will pass (local broker works the best) @@ -189,4 +181,4 @@ def test_weekend_mqtt_publish(env, extra_data): if __name__ == '__main__': - test_weekend_mqtt_publish() + test_weekend_mqtt_publish(dut=ttfw_idf.ESP32QEMUDUT if sys.argv[1:] == ['qemu'] else ttfw_idf.ESP32DUT) diff --git a/components/mqtt/weekend_test/test_weekend_mqtt_qemu.yml b/components/mqtt/weekend_test/test_weekend_mqtt_qemu.yml new file mode 100644 index 0000000000..e4a72ecd03 --- /dev/null +++ b/components/mqtt/weekend_test/test_weekend_mqtt_qemu.yml @@ -0,0 +1,7 @@ +CaseConfig: +- name: test_weekend_mqtt_publish + overwrite: + dut: + class: ESP32QEMUDUT + package: ttfw_idf + diff --git a/examples/protocols/mqtt/publish_test/sdkconfig.qemu b/examples/protocols/mqtt/publish_test/sdkconfig.qemu new file mode 100644 index 0000000000..e0d289d74e --- /dev/null +++ b/examples/protocols/mqtt/publish_test/sdkconfig.qemu @@ -0,0 +1,19 @@ +CONFIG_IDF_TARGET_ESP32=y +CONFIG_EXAMPLE_USE_OPENETH=y +CONFIG_ETH_USE_OPENETH=y +CONFIG_ETH_OPENETH_DMA_RX_BUFFER_NUM=4 +CONFIG_ETH_OPENETH_DMA_TX_BUFFER_NUM=1 +CONFIG_EXAMPLE_CONNECT_ETHERNET=y +CONFIG_LOG_DEFAULT_LEVEL_DEBUG=y +CONFIG_ESPTOOLPY_FLASHMODE_DOUT=y +CONFIG_MBEDTLS_ASYMMETRIC_CONTENT_LEN=y +CONFIG_MBEDTLS_SSL_IN_CONTENT_LEN=16384 +CONFIG_MBEDTLS_SSL_OUT_CONTENT_LEN=16384 +CONFIG_EXAMPLE_BROKER_SSL_URI="mqtts://${EXAMPLE_MQTT_BROKER_SSL}" +CONFIG_EXAMPLE_BROKER_TCP_URI="mqtt://${EXAMPLE_MQTT_BROKER_TCP}" +CONFIG_EXAMPLE_BROKER_WS_URI="ws://${EXAMPLE_MQTT_BROKER_WS}/ws" +CONFIG_EXAMPLE_BROKER_WSS_URI="wss://${EXAMPLE_MQTT_BROKER_WSS}/ws" +CONFIG_EXAMPLE_BROKER_CERTIFICATE_OVERRIDE="${EXAMPLE_MQTT_BROKER_CERTIFICATE}" +CONFIG_MBEDTLS_HARDWARE_AES=n +CONFIG_MBEDTLS_HARDWARE_MPI=n +CONFIG_MBEDTLS_HARDWARE_SHA=n diff --git a/tools/ci/python_packages/tiny_test_fw/DUT.py b/tools/ci/python_packages/tiny_test_fw/DUT.py index 71007d85b0..c5139c628d 100644 --- a/tools/ci/python_packages/tiny_test_fw/DUT.py +++ b/tools/ci/python_packages/tiny_test_fw/DUT.py @@ -766,7 +766,7 @@ class SerialDUT(BaseDUT): return formatted_data def _port_open(self): - self.port_inst = serial.Serial(self.port, **self.serial_configs) + self.port_inst = serial.serial_for_url(self.port, **self.serial_configs) def _port_close(self): self.port_inst.close() diff --git a/tools/ci/python_packages/tiny_test_fw/bin/example.py b/tools/ci/python_packages/tiny_test_fw/bin/example.py index c4348bf346..238d3561cc 100644 --- a/tools/ci/python_packages/tiny_test_fw/bin/example.py +++ b/tools/ci/python_packages/tiny_test_fw/bin/example.py @@ -14,24 +14,12 @@ """ example of writing test with TinyTestFW """ import re -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 +import ttfw_idf +from tiny_test_fw import TinyFW -@IDF.idf_example_test(env_tag="Example_WIFI") +@ttfw_idf.idf_example_test(env_tag="Example_WIFI") def test_examples_protocol_https_request(env, extra_data): """ steps: | @@ -39,7 +27,7 @@ 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) dut1.start_app() dut1.expect(re.compile(r"Connecting to www.howsmyssl.com:443"), timeout=30) dut1.expect("Performing the SSL/TLS handshake") @@ -51,5 +39,5 @@ def test_examples_protocol_https_request(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_examples_protocol_https_request() diff --git a/tools/ci/python_packages/ttfw_idf/IDFDUT.py b/tools/ci/python_packages/ttfw_idf/IDFDUT.py index c5e0608d5b..bd47648537 100644 --- a/tools/ci/python_packages/ttfw_idf/IDFDUT.py +++ b/tools/ci/python_packages/ttfw_idf/IDFDUT.py @@ -20,6 +20,8 @@ import re import functools import tempfile import subprocess +import time +import pexpect # python2 and python3 queue package name is different try: @@ -418,6 +420,7 @@ class IDFDUT(DUT.SerialDUT): class ESP32DUT(IDFDUT): TARGET = "esp32" TOOLCHAIN_PREFIX = "xtensa-esp32-elf-" + @classmethod def _get_rom(cls): return esptool.ESP32ROM @@ -426,6 +429,7 @@ class ESP32DUT(IDFDUT): class ESP32S2DUT(IDFDUT): TARGET = "esp32s2beta" TOOLCHAIN_PREFIX = "xtensa-esp32s2-elf-" + @classmethod def _get_rom(cls): return esptool.ESP32S2ROM @@ -434,13 +438,105 @@ class ESP32S2DUT(IDFDUT): class ESP8266DUT(IDFDUT): TARGET = "esp8266" TOOLCHAIN_PREFIX = "xtensa-lx106-elf-" + @classmethod def _get_rom(cls): return esptool.ESP8266ROM def get_target_by_rom_class(cls): - for c in [ESP32DUT, ESP32S2DUT, ESP8266DUT]: + for c in [ESP32DUT, ESP32S2DUT, ESP8266DUT, IDFQEMUDUT]: if c._get_rom() == cls: return c.TARGET return None + + +class IDFQEMUDUT(IDFDUT): + TARGET = None + TOOLCHAIN_PREFIX = None + ERASE_NVS = True + DEFAULT_EXPECT_TIMEOUT = 30 # longer timeout, since app startup takes more time in QEMU (due to slow SHA emulation) + QEMU_SERIAL_PORT = 3334 + + def __init__(self, name, port, log_file, app, allow_dut_exception=False, **kwargs): + self.flash_image = tempfile.NamedTemporaryFile('rb+', suffix=".bin", prefix="qemu_flash_img") + self.app = app + self.flash_size = 4 * 1024 * 1024 + self._write_flash_img() + + args = [ + "qemu-system-xtensa", + "-nographic", + "-machine", self.TARGET, + "-drive", "file={},if=mtd,format=raw".format(self.flash_image.name), + "-nic", "user,model=open_eth", + "-serial", "tcp::{},server,nowait".format(self.QEMU_SERIAL_PORT), + "-S", + "-global driver=timer.esp32.timg,property=wdt_disable,value=true"] + # TODO(IDF-1242): generate a temporary efuse binary, pass it to QEMU + + if "QEMU_BIOS_PATH" in os.environ: + args += ["-L", os.environ["QEMU_BIOS_PATH"]] + + self.qemu = pexpect.spawn(" ".join(args), timeout=self.DEFAULT_EXPECT_TIMEOUT) + self.qemu.expect_exact(b"(qemu)") + super(IDFQEMUDUT, self).__init__(name, port, log_file, app, allow_dut_exception=allow_dut_exception, **kwargs) + + def _write_flash_img(self): + self.flash_image.seek(0) + self.flash_image.write(b'\x00' * self.flash_size) + for offs, path in self.app.flash_files: + with open(path, "rb") as flash_file: + contents = flash_file.read() + self.flash_image.seek(offs) + self.flash_image.write(contents) + self.flash_image.flush() + + @classmethod + def _get_rom(cls): + return esptool.ESP32ROM + + @classmethod + def get_mac(cls, app, port): + # TODO(IDF-1242): get this from QEMU/efuse binary + return "11:22:33:44:55:66" + + @classmethod + def confirm_dut(cls, port, **kwargs): + return True, cls.TARGET + + def start_app(self, erase_nvs=ERASE_NVS): + # TODO: implement erase_nvs + # since the flash image is generated every time in the constructor, maybe this isn't needed... + self.qemu.sendline(b"cont\n") + self.qemu.expect_exact(b"(qemu)") + + def reset(self): + self.qemu.sendline(b"system_reset\n") + self.qemu.expect_exact(b"(qemu)") + + def erase_partition(self, partition): + raise NotImplementedError("method not erase_partition not implemented") + + def dump_flush(self, output_file, **kwargs): + raise NotImplementedError("method not dump_flush not implemented") + + @classmethod + def list_available_ports(cls): + return ["socket://localhost:{}".format(cls.QEMU_SERIAL_PORT)] + + def close(self): + super(IDFQEMUDUT, self).close() + self.qemu.sendline(b"q\n") + self.qemu.expect_exact(b"(qemu)") + for _ in range(self.DEFAULT_EXPECT_TIMEOUT): + if not self.qemu.isalive(): + break + time.sleep(1) + else: + self.qemu.terminate(force=True) + + +class ESP32QEMUDUT(IDFQEMUDUT): + TARGET = "esp32" + TOOLCHAIN_PREFIX = "xtensa-esp32-elf-" diff --git a/tools/ci/python_packages/ttfw_idf/__init__.py b/tools/ci/python_packages/ttfw_idf/__init__.py index 8c72047d5f..8d7f4df639 100644 --- a/tools/ci/python_packages/ttfw_idf/__init__.py +++ b/tools/ci/python_packages/ttfw_idf/__init__.py @@ -16,7 +16,7 @@ import re from tiny_test_fw import TinyFW, Utility from IDFApp import IDFApp, Example, LoadableElfExample, UT # noqa: export all Apps for users -from IDFDUT import IDFDUT, ESP32DUT, ESP32S2DUT, ESP8266DUT # noqa: export DUTs for users +from IDFDUT import IDFDUT, ESP32DUT, ESP32S2DUT, ESP8266DUT, ESP32QEMUDUT # noqa: export DUTs for users def format_case_id(chip, case_name): diff --git a/tools/ci/python_packages/ttfw_idf/requirements.txt b/tools/ci/python_packages/ttfw_idf/requirements.txt new file mode 100644 index 0000000000..112ee0bc80 --- /dev/null +++ b/tools/ci/python_packages/ttfw_idf/requirements.txt @@ -0,0 +1,3 @@ +-r ../tiny_test_fw/requirements.txt +pexpect +python-gitlab