alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2025-12-15 19:34:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: update example and unit tests with new import roles:

Browse Source 
tiny_test_fw is a python package now. import it using normal way.
This commit is contained in:
He Yin Ling
2019-11-27 11:58:07 +08:00
parent 4d45932c5e
commit c906e2afee
61 changed files with 1283 additions and 864 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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,))

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

@@ -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

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

@@ -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

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

@@ -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

23
examples/protocols/esp_http_client/esp_http_client_test.py
View File

@@ -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)

31
examples/protocols/http_server/advanced_tests/http_server_advanced_test.py
View File

@@ -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")

30
examples/protocols/http_server/persistent_sockets/http_server_persistence_test.py
View File

@@ -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")

105
examples/protocols/http_server/persistent_sockets/scripts/adder.py
View File

@@ -1,105 +0,0 @@
#!/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.
from __future__ import print_function
from __future__ import unicode_literals
from builtins import str
from builtins import range
import http.client
import argparse
import Utility
def start_session(ip, port):
return http.client.HTTPConnection(ip, int(port), timeout=15)
def end_session(conn):
conn.close()
def getreq(conn, path, verbose=False):
conn.request("GET", path)
resp = conn.getresponse()
data = resp.read()
if verbose:
Utility.console_log("GET : " + path)
Utility.console_log("Status : " + resp.status)
Utility.console_log("Reason : " + resp.reason)
Utility.console_log("Data length : " + str(len(data)))
Utility.console_log("Data content : " + data)
return data
def postreq(conn, path, data, verbose=False):
conn.request("POST", path, data)
resp = conn.getresponse()
data = resp.read()
if verbose:
Utility.console_log("POST : " + data)
Utility.console_log("Status : " + resp.status)
Utility.console_log("Reason : " + resp.reason)
Utility.console_log("Data length : " + str(len(data)))
Utility.console_log("Data content : " + data)
return data
def putreq(conn, path, body, verbose=False):
conn.request("PUT", path, body)
resp = conn.getresponse()
data = resp.read()
if verbose:
Utility.console_log("PUT : " + path, body)
Utility.console_log("Status : " + resp.status)
Utility.console_log("Reason : " + resp.reason)
Utility.console_log("Data length : " + str(len(data)))
Utility.console_log("Data content : " + data)
return data
if __name__ == '__main__':
# Configure argument parser
parser = argparse.ArgumentParser(description='Run HTTPd Test')
parser.add_argument('IP', metavar='IP', type=str, help='Server IP')
parser.add_argument('port', metavar='port', type=str, help='Server port')
parser.add_argument('N', metavar='integer', type=int, help='Integer to sum upto')
args = vars(parser.parse_args())
# Get arguments
ip = args['IP']
port = args['port']
N = args['N']
# Establish HTTP connection
Utility.console_log("Connecting to => " + ip + ":" + port)
conn = start_session(ip, port)
# Reset adder context to specified value(0)
# -- Not needed as new connection will always
# -- have zero value of the accumulator
Utility.console_log("Reset the accumulator to 0")
putreq(conn, "/adder", str(0))
# Sum numbers from 1 to specified value(N)
Utility.console_log("Summing numbers from 1 to " + str(N))
for i in range(1, N + 1):
postreq(conn, "/adder", str(i))
# Fetch the result
Utility.console_log("Result :" + getreq(conn, "/adder"))
# Close HTTP connection
end_session(conn)

28
examples/protocols/http_server/simple/http_server_simple_test.py
View File

@@ -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")

279
examples/protocols/http_server/simple/scripts/client.py
View File

@@ -1,279 +0,0 @@
#!/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.
from __future__ import print_function
from __future__ import unicode_literals
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
def verbose_print(verbosity, *args):
if (verbosity):
Utility.console_log(''.join(str(elems) for elems in args))
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
def test_get_handler(ip, port, verbosity=False):
verbose_print(verbosity, "======== GET HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
uri = "/hello?query1=value1&query2=value2&query3=value3"
# GET hello response
test_headers = {"Test-Header-1":"Test-Value-1", "Test-Header-2":"Test-Value-2"}
verbose_print(verbosity, "Sending GET to URI : ", uri)
verbose_print(verbosity, "Sending additional headers : ")
for k, v in test_headers.items():
verbose_print(verbosity, "\t", k, ": ", v)
sess.request("GET", url=uri, headers=test_headers)
resp = sess.getresponse()
resp_hdrs = resp.getheaders()
resp_data = resp.read().decode()
# Close HTTP connection
sess.close()
if not (
test_val("Status code mismatch", 200, resp.status) and
test_val("Response mismatch", "Custom-Value-1", resp.getheader("Custom-Header-1")) and
test_val("Response mismatch", "Custom-Value-2", resp.getheader("Custom-Header-2")) and
test_val("Response mismatch", "Hello World!", resp_data)
):
return False
verbose_print(verbosity, "vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv")
verbose_print(verbosity, "Server response to GET /hello")
verbose_print(verbosity, "Response Headers : ")
for k, v in resp_hdrs:
verbose_print(verbosity, "\t", k, ": ", v)
verbose_print(verbosity, "Response Data : " + resp_data)
verbose_print(verbosity, "========================================\n")
return True
def test_post_handler(ip, port, msg, verbosity=False):
verbose_print(verbosity, "======== POST HANDLER TEST ============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
# POST message to /echo and get back response
sess.request("POST", url="/echo", body=msg)
resp = sess.getresponse()
resp_data = resp.read().decode()
verbose_print(verbosity, "Server response to POST /echo (" + msg + ")")
verbose_print(verbosity, "vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv")
verbose_print(verbosity, resp_data)
verbose_print(verbosity, "========================================\n")
# Close HTTP connection
sess.close()
return test_val("Response mismatch", msg, resp_data)
def test_put_handler(ip, port, verbosity=False):
verbose_print(verbosity, "======== PUT HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
# PUT message to /ctrl to disable /hello and /echo URI handlers
# and set 404 error handler to custom http_404_error_handler()
verbose_print(verbosity, "Disabling /hello and /echo handlers")
sess.request("PUT", url="/ctrl", body="0")
resp = sess.getresponse()
resp.read()
try:
# Send HTTP request to /hello URI
sess.request("GET", url="/hello")
resp = sess.getresponse()
resp_data = resp.read().decode()
# 404 Error must be returned from server as URI /hello is no longer available.
# But the custom error handler http_404_error_handler() will not close the
# session if the requested URI is /hello
if not test_val("Status code mismatch", 404, resp.status):
raise AssertionError
# Compare error response string with expectation
verbose_print(verbosity, "Response on GET /hello : " + resp_data)
if not test_val("Response mismatch", "/hello URI is not available", resp_data):
raise AssertionError
# Using same session for sending an HTTP request to /echo, as it is expected
# that the custom error handler http_404_error_handler() would not have closed
# the session
sess.request("POST", url="/echo", body="Some content")
resp = sess.getresponse()
resp_data = resp.read().decode()
# 404 Error must be returned from server as URI /hello is no longer available.
# The custom error handler http_404_error_handler() will close the session
# this time as the requested URI is /echo
if not test_val("Status code mismatch", 404, resp.status):
raise AssertionError
# Compare error response string with expectation
verbose_print(verbosity, "Response on POST /echo : " + resp_data)
if not test_val("Response mismatch", "/echo URI is not available", resp_data):
raise AssertionError
try:
# Using same session should fail as by now the session would have closed
sess.request("POST", url="/hello", body="Some content")
resp = sess.getresponse()
resp.read().decode()
# If control reaches this point then the socket was not closed.
# This is not expected
verbose_print(verbosity, "Socket not closed by server")
raise AssertionError
except http.client.HTTPException:
# Catch socket error as we tried to communicate with an already closed socket
pass
except http.client.HTTPException:
verbose_print(verbosity, "Socket closed by server")
return False
except AssertionError:
return False
finally:
# Close HTTP connection
sess.close()
verbose_print(verbosity, "Enabling /hello handler")
# Create new connection
sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
# PUT message to /ctrl to enable /hello URI handler
# and restore 404 error handler to default
sess.request("PUT", url="/ctrl", body="1")
resp = sess.getresponse()
resp.read()
# Close HTTP connection
sess.close()
# Create new connection
sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
try:
# Sending HTTP request to /hello should work now
sess.request("GET", url="/hello")
resp = sess.getresponse()
resp_data = resp.read().decode()
if not test_val("Status code mismatch", 200, resp.status):
raise AssertionError
verbose_print(verbosity, "Response on GET /hello : " + resp_data)
if not test_val("Response mismatch", "Hello World!", resp_data):
raise AssertionError
# 404 Error handler should have been restored to default
sess.request("GET", url="/invalid")
resp = sess.getresponse()
resp_data = resp.read().decode()
if not test_val("Status code mismatch", 404, resp.status):
raise AssertionError
verbose_print(verbosity, "Response on GET /invalid : " + resp_data)
if not test_val("Response mismatch", "This URI does not exist", resp_data):
raise AssertionError
except http.client.HTTPException:
verbose_print(verbosity, "Socket closed by server")
return False
except AssertionError:
return False
finally:
# Close HTTP connection
sess.close()
return True
def test_custom_uri_query(ip, port, query, verbosity=False):
verbose_print(verbosity, "======== GET HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
sess = http.client.HTTPConnection(ip + ":" + port, timeout=15)
uri = "/hello?" + query
# GET hello response
verbose_print(verbosity, "Sending GET to URI : ", uri)
sess.request("GET", url=uri, headers={})
resp = sess.getresponse()
resp_data = resp.read().decode()
verbose_print(verbosity, "vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv")
verbose_print(verbosity, "Server response to GET /hello")
verbose_print(verbosity, "Response Data : " + resp_data)
verbose_print(verbosity, "========================================\n")
# Close HTTP connection
sess.close()
return "Hello World!" == resp_data
if __name__ == '__main__':
# Configure argument parser
parser = argparse.ArgumentParser(description='Run HTTPd Test')
parser.add_argument('IP', metavar='IP', type=str, help='Server IP')
parser.add_argument('port', metavar='port', type=str, help='Server port')
parser.add_argument('msg', metavar='message', type=str, help='Message to be sent to server')
args = vars(parser.parse_args())
# Get arguments
ip = args['IP']
port = args['port']
msg = args['msg']
if not (
test_get_handler(ip, port, True) and
test_put_handler(ip, port, True) and
test_post_handler(ip, port, msg, True)
):
Utility.console_log("Failed!")

22
examples/protocols/https_request/example_test.py
View File

@@ -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)

27
examples/protocols/mdns/mdns_example_test.py
View File

@@ -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)

17
examples/protocols/modbus/serial/example_test.py
View File

@@ -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)

27
examples/protocols/mqtt/ssl/mqtt_ssl_example_test.py
View File

@@ -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"])

25
examples/protocols/mqtt/tcp/mqtt_tcp_example_test.py
View File

@@ -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))

25
examples/protocols/mqtt/ws/mqtt_ws_example_test.py
View File

@@ -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"])

24
examples/protocols/mqtt/wss/mqtt_wss_example_test.py
View File

@@ -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"])

19
examples/protocols/websocket/example_test.py
View File

@@ -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 ...")