From 5a51af523c4350fdb276d218a032034cc7c315af Mon Sep 17 00:00:00 2001 From: Harshit Malpani Date: Tue, 10 May 2022 12:16:56 +0530 Subject: [PATCH] https examples pytest migration --- examples/protocols/https_mbedtls/README.md | 2 + .../protocols/https_mbedtls/example_test.py | 47 ---- .../https_mbedtls/pytest_https_mbedtls.py | 42 +++ examples/protocols/https_request/README.md | 2 + .../protocols/https_request/example_test.py | 251 ------------------ .../https_request/pytest_https_request.py | 234 ++++++++++++++++ .../protocols/https_server/simple/README.md | 2 + ..._test.py => pytest_https_server_simple.py} | 124 +++++---- .../https_server/wss_server/README.md | 2 + ...ple_test.py => pytest_https_wss_server.py} | 75 +++--- .../protocols/https_x509_bundle/README.md | 2 + .../https_x509_bundle/example_test.py | 40 --- .../pytest_https_x509_bundle.py | 47 ++++ 13 files changed, 434 insertions(+), 436 deletions(-) delete mode 100644 examples/protocols/https_mbedtls/example_test.py create mode 100644 examples/protocols/https_mbedtls/pytest_https_mbedtls.py delete mode 100644 examples/protocols/https_request/example_test.py create mode 100644 examples/protocols/https_request/pytest_https_request.py rename examples/protocols/https_server/simple/{example_test.py => pytest_https_server_simple.py} (68%) rename examples/protocols/https_server/wss_server/{wss_server_example_test.py => pytest_https_wss_server.py} (63%) delete mode 100644 examples/protocols/https_x509_bundle/example_test.py create mode 100644 examples/protocols/https_x509_bundle/pytest_https_x509_bundle.py diff --git a/examples/protocols/https_mbedtls/README.md b/examples/protocols/https_mbedtls/README.md index 52b636b326..b86aa17468 100644 --- a/examples/protocols/https_mbedtls/README.md +++ b/examples/protocols/https_mbedtls/README.md @@ -1,3 +1,5 @@ +| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 | +| ----------------- | ----- | -------- | -------- | -------- | # HTTP server with TLS support using mbedTLS (See the README.md file in the upper level 'examples' directory for more information about examples.) diff --git a/examples/protocols/https_mbedtls/example_test.py b/examples/protocols/https_mbedtls/example_test.py deleted file mode 100644 index 5c200c295a..0000000000 --- a/examples/protocols/https_mbedtls/example_test.py +++ /dev/null @@ -1,47 +0,0 @@ -#!/usr/bin/env python -# -# SPDX-FileCopyrightText: 2021 Espressif Systems (Shanghai) CO LTD -# SPDX-License-Identifier: Apache-2.0 - -import os -import re - -import tiny_test_fw -import ttfw_idf -from tiny_test_fw import Utility - - -@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') -def test_examples_protocol_https_mbedtls(env, extra_data): # type: (tiny_test_fw.Env.Env, None) -> None # pylint: disable=unused-argument - """ - steps: | - 1. join AP - 2. connect to www.howsmyssl.com:443 - 3. send http request - """ - app_name = 'https_mbedtls' - dut1 = env.get_dut(app_name, 'examples/protocols/https_mbedtls', dut_class=ttfw_idf.ESP32DUT) - # check and log bin size - binary_file = os.path.join(dut1.app.binary_path, 'https_mbedtls.bin') - bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_mbedtls_bin_size', '{}KB'.format(bin_size // 1024)) - # start test - dut1.start_app() - dut1.expect('Connected.', timeout=30) - Utility.console_log('TCP connection established with the server\n performing SSL/TLS handshake') - dut1.expect('Performing the SSL/TLS handshake...') - dut1.expect('Certificate verified.') - Utility.console_log('SSL/TLS handshake successful') - dut1.expect('Writing HTTP request...') - dut1.expect('Reading HTTP response...') - dut1.expect(re.compile(r'Completed (\d) requests')) - - # Read free heap size - res = dut1.expect(ttfw_idf.MINIMUM_FREE_HEAP_SIZE_RE) - if not res: - raise ValueError('Maximum heap size info not found') - ttfw_idf.print_heap_size(app_name, dut1.app.config_name, dut1.TARGET, res[0]) - - -if __name__ == '__main__': - test_examples_protocol_https_mbedtls() # pylint: disable=no-value-for-parameter diff --git a/examples/protocols/https_mbedtls/pytest_https_mbedtls.py b/examples/protocols/https_mbedtls/pytest_https_mbedtls.py new file mode 100644 index 0000000000..648470ecef --- /dev/null +++ b/examples/protocols/https_mbedtls/pytest_https_mbedtls.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python +# +# SPDX-FileCopyrightText: 2021-2022 Espressif Systems (Shanghai) CO LTD +# SPDX-License-Identifier: Apache-2.0 + +import logging +import os + +import pytest +from pytest_embedded import Dut + + +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.ethernet +def test_examples_protocol_https_mbedtls(dut: Dut) -> None: + """ + steps: | + 1. join AP + 2. connect to www.howsmyssl.com:443 + 3. send http request + """ + # check and log bin size + binary_file = os.path.join(dut.app.binary_path, 'https_mbedtls.bin') + bin_size = os.path.getsize(binary_file) + logging.info('https_mbedtls_bin_size : {}KB'.format(bin_size // 1024)) + # start test + dut.expect('Connected.', timeout=30) + logging.info('TCP connection established with the server\n performing SSL/TLS handshake') + dut.expect('Performing the SSL/TLS handshake...') + dut.expect('Certificate verified.') + logging.info('SSL/TLS handshake successful') + dut.expect('Writing HTTP request...') + dut.expect('Reading HTTP response...') + dut.expect(r'Completed (\d) requests') + + # Read free heap size + res = dut.expect(r'Minimum free heap size: (\d+)')[1].decode() + if not res: + raise ValueError('Maximum heap size info not found') diff --git a/examples/protocols/https_request/README.md b/examples/protocols/https_request/README.md index 05fed4de5a..821f6cbe1d 100644 --- a/examples/protocols/https_request/README.md +++ b/examples/protocols/https_request/README.md @@ -1,3 +1,5 @@ +| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 | +| ----------------- | ----- | -------- | -------- | -------- | # HTTPS Request Example Uses APIs from `esp-tls` component to make a very simple HTTPS request over a secure connection, including verifying the server TLS certificate. diff --git a/examples/protocols/https_request/example_test.py b/examples/protocols/https_request/example_test.py deleted file mode 100644 index 222b9839ad..0000000000 --- a/examples/protocols/https_request/example_test.py +++ /dev/null @@ -1,251 +0,0 @@ -import http.server -import multiprocessing -import os -import re -import socket -import ssl - -import ttfw_idf -from RangeHTTPServer import RangeRequestHandler -from tiny_test_fw import DUT, Utility - - -def get_my_ip(): - s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s1.connect(('8.8.8.8', 80)) - my_ip = s1.getsockname()[0] - s1.close() - return my_ip - - -def get_server_status(host_ip, port): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - server_status = sock.connect_ex((host_ip, port)) - sock.close() - if server_status == 0: - return True - return False - - -def https_request_handler(): - """ - Returns a request handler class that handles broken pipe exception - """ - class RequestHandler(RangeRequestHandler): - protocol_version = 'HTTP/1.1' - - def finish(self): - try: - if not self.wfile.closed: - self.wfile.flush() - self.wfile.close() - except socket.error: - pass - self.rfile.close() - - def handle(self): - try: - RangeRequestHandler.handle(self) - except socket.error: - pass - - def do_GET(self): - self.close_connection = True - self.send_response(200) - self.end_headers() - - return RequestHandler - - -def start_https_server(server_file, key_file, server_ip, server_port): - - requestHandler = https_request_handler() - httpd = http.server.HTTPServer((server_ip, server_port), requestHandler) - - httpd.socket = ssl.wrap_socket(httpd.socket, keyfile=key_file, - certfile=server_file, server_side=True) - httpd.serve_forever() - - -@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') -def test_examples_protocol_https_request_cli_session_tickets(env, extra_data): - Utility.console_log("Testing for \"esp_tls client session tickets\"") - - dut1 = env.get_dut('https_request_ses_tkt', 'examples/protocols/https_request', dut_class=ttfw_idf.ESP32DUT, app_config_name='cli_ses_tkt') - Utility.console_log('[app_config_name] - {}'.format(dut1.app.config_name)) - # check and log bin size - binary_file = os.path.join(dut1.app.binary_path, 'https_request.bin') - bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_request_bin_size', '{}KB'.format(bin_size // 1024)) - # start test - host_ip = get_my_ip() - server_port = 8070 - server_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_cert.pem') - key_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_key.pem') - if (get_server_status(host_ip, server_port) is False): - thread1 = multiprocessing.Process(target=start_https_server, args=(server_file, key_file, host_ip, server_port)) - thread1.daemon = True - thread1.start() - Utility.console_log('The server started on {}:{}'.format(host_ip, server_port)) - dut1.start_app() - - dut1.expect('Loaded app from partition at offset', timeout=30) - try: - ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=60) - print('Connected to AP with IP: {}'.format(ip_address)) - except DUT.ExpectTimeout: - raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - - dut1.expect('Start https_request example', timeout=30) - - print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port))) - - dut1.write('https://' + host_ip + ':' + str(server_port)) - Utility.console_log("Testing for \"https_request using saved session\"") - - # Check for connection using already saved client session - try: - dut1.expect(re.compile('https_request to local server'), timeout=30) - dut1.expect_all('Connection established...', - 'Reading HTTP response...', - 'HTTP/1.1 200 OK', - re.compile('connection closed')) - except Exception: - Utility.console_log("Failed to connect to local https server\"") - raise - - try: - dut1.expect(re.compile('https_request using saved client session'), timeout=20) - dut1.expect_all('Connection established...', - 'Reading HTTP response...', - 'HTTP/1.1 200 OK', - re.compile('connection closed')) - except Exception: - Utility.console_log("Failed the test for \"https_request using saved client session\"") - raise - - Utility.console_log("Passed the test for \"https_request using saved client session\"") - thread1.terminate() - env.close_dut('https_request_ses_tkt') - - -@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') -def test_examples_protocol_https_request_dynamic_buffers(env, extra_data): - # Check for connection using crt bundle with mbedtls dynamic resource enabled - dut1 = env.get_dut('https_request_ssldyn', 'examples/protocols/https_request', dut_class=ttfw_idf.ESP32DUT, app_config_name='ssldyn') - # check and log bin size - Utility.console_log('[app_config_name] - {}'.format(dut1.app.config_name)) - binary_file = os.path.join(dut1.app.binary_path, 'https_request.bin') - bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_request_bin_size', '{}KB'.format(bin_size // 1024)) - # start test - dut1.start_app() - - dut1.expect('Loaded app from partition at offset', timeout=30) - try: - ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=60) - print('Connected to AP with IP: {}'.format(ip_address)) - except DUT.ExpectTimeout: - raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - - # only check if one connection is established - Utility.console_log("Testing for \"https_request using crt bundle\" with mbedtls dynamic resource enabled") - try: - dut1.expect(re.compile('https_request using crt bundle'), timeout=30) - dut1.expect_all('Connection established...', - 'Reading HTTP response...', - 'HTTP/1.1 200 OK', - re.compile('connection closed')) - except Exception: - Utility.console_log("Failed the test for \"https_request using crt bundle\" when mbedtls dynamic resource was enabled") - raise - Utility.console_log("Passed the test for \"https_request using crt bundle\" when mbedtls dynamic resource was enabled") - - # Read free heap size - res = dut1.expect(ttfw_idf.MINIMUM_FREE_HEAP_SIZE_RE,timeout=20) - if not res: - raise ValueError('Maximum heap size info not found') - ttfw_idf.print_heap_size('https_request', dut1.app.config_name, dut1.TARGET, res[0]) - - env.close_dut('https_request_ssldyn') - - -@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1') -def test_examples_protocol_https_request(env, extra_data): - - """ - steps: | - 1. join AP - 2. establish TLS connection to www.howsmyssl.com:443 with multiple - certificate verification options - 3. send http request - """ - dut1 = env.get_dut('https_request', 'examples/protocols/https_request', dut_class=ttfw_idf.ESP32DUT) - # check and log bin size - Utility.console_log('[app_config_name] - {}'.format(dut1.app.config_name)) - binary_file = os.path.join(dut1.app.binary_path, 'https_request.bin') - bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_request_bin_size', '{}KB'.format(bin_size // 1024)) - # start tes - Utility.console_log('Starting https_request simple test app') - dut1.start_app() - - dut1.expect('Loaded app from partition at offset', timeout=30) - try: - ip_address = dut1.expect(re.compile(r' (sta|eth) ip: ([^,]+),'), timeout=60) - print('Connected to AP with IP: {}'.format(ip_address)) - except DUT.ExpectTimeout: - raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') - - # Check for connection using crt bundle - Utility.console_log("Testing for \"https_request using crt bundle\"") - try: - dut1.expect(re.compile('https_request using crt bundle'), timeout=30) - dut1.expect_all('Certificate validated', - 'Connection established...', - 'Reading HTTP response...', - 'HTTP/1.1 200 OK', - re.compile('connection closed')) - except Exception: - Utility.console_log("Failed the test for \"https_request using crt bundle\"") - raise - Utility.console_log("Passed the test for \"https_request using crt bundle\"") - - # Read free heap size - res = dut1.expect(ttfw_idf.MINIMUM_FREE_HEAP_SIZE_RE,timeout=20) - if not res: - raise ValueError('Maximum heap size info not found') - ttfw_idf.print_heap_size('https_request', dut1.app.config_name, dut1.TARGET, res[0]) - - # Check for connection using cacert_buf - Utility.console_log("Testing for \"https_request using cacert_buf\"") - try: - dut1.expect(re.compile('https_request using cacert_buf'), timeout=20) - dut1.expect_all('Connection established...', - 'Reading HTTP response...', - 'HTTP/1.1 200 OK', - re.compile('connection closed')) - except Exception: - Utility.console_log("Passed the test for \"https_request using cacert_buf\"") - raise - Utility.console_log("Passed the test for \"https_request using cacert_buf\"") - - # Check for connection using global ca_store - Utility.console_log("Testing for \"https_request using global ca_store\"") - try: - dut1.expect(re.compile('https_request using global ca_store'), timeout=20) - dut1.expect_all('Connection established...', - 'Reading HTTP response...', - 'HTTP/1.1 200 OK', - re.compile('connection closed')) - except Exception: - Utility.console_log("Failed the test for \"https_request using global ca_store\"") - raise - Utility.console_log("Passed the test for \"https_request using global ca_store\"") - env.close_dut('https_request') - - -if __name__ == '__main__': - test_examples_protocol_https_request() - test_examples_protocol_https_request_cli_session_tickets() - test_examples_protocol_https_request_dynamic_buffers() diff --git a/examples/protocols/https_request/pytest_https_request.py b/examples/protocols/https_request/pytest_https_request.py new file mode 100644 index 0000000000..6f3bc42b17 --- /dev/null +++ b/examples/protocols/https_request/pytest_https_request.py @@ -0,0 +1,234 @@ +# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD +# SPDX-License-Identifier: Unlicense OR CC0-1.0 +import http.server +import logging +import multiprocessing +import os +import socket +import ssl +from typing import Callable + +import pexpect +import pytest +from pytest_embedded import Dut +from RangeHTTPServer import RangeRequestHandler + + +def get_my_ip() -> str: + s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s1.connect(('8.8.8.8', 80)) + my_ip = '' + my_ip = s1.getsockname()[0] + s1.close() + return my_ip + + +def get_server_status(host_ip: str, port: int) -> bool: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_status = sock.connect_ex((host_ip, port)) + sock.close() + if server_status == 0: + return True + return False + + +def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]: + """ + Returns a request handler class that handles broken pipe exception + """ + class RequestHandler(RangeRequestHandler): + protocol_version = 'HTTP/1.1' + + def finish(self) -> None: + try: + if not self.wfile.closed: + self.wfile.flush() + self.wfile.close() + except socket.error: + pass + self.rfile.close() + + def handle(self) -> None: + try: + RangeRequestHandler.handle(self) + except socket.error: + pass + + def do_GET(self) -> None: + self.close_connection = True + self.send_response(200) + self.end_headers() + + return RequestHandler + + +def start_https_server(server_file: str, key_file: str, server_ip: str, server_port: int) -> None: + + requestHandler = https_request_handler() + httpd = http.server.HTTPServer((server_ip, server_port), requestHandler) + + httpd.socket = ssl.wrap_socket(httpd.socket, keyfile=key_file, + certfile=server_file, server_side=True) + httpd.serve_forever() + + +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.ethernet +@pytest.mark.parametrize('config', ['cli_ses_tkt',], indirect=True) +def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None: + logging.info("Testing for \"esp_tls client session tickets\"") + + # check and log bin size + binary_file = os.path.join(dut.app.binary_path, 'https_request.bin') + bin_size = os.path.getsize(binary_file) + logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024)) + # start test + host_ip = get_my_ip() + server_port = 8070 + server_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_cert.pem') + key_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_key.pem') + if (get_server_status(host_ip, server_port) is False): + thread1 = multiprocessing.Process(target=start_https_server, args=(server_file, key_file, host_ip, server_port)) + thread1.daemon = True + thread1.start() + logging.info('The server started on {}:{}'.format(host_ip, server_port)) + + dut.expect('Loaded app from partition at offset', timeout=30) + try: + ip_address = dut.expect(r' (sta|eth) ip: (\d+\.\d+\.\d+\.\d+)', timeout=60)[2].decode() + print('Connected to AP with IP: {}'.format(ip_address)) + except pexpect.exceptions.TIMEOUT: + raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + + dut.expect('Start https_request example', timeout=30) + + print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port))) + + dut.write('https://' + host_ip + ':' + str(server_port)) + logging.info("Testing for \"https_request using saved session\"") + + # Check for connection using already saved client session + try: + dut.expect('https_request to local server', timeout=30) + dut.expect(['Connection established...', + 'Reading HTTP response...', + 'HTTP/1.1 200 OK', + 'connection closed'], expect_all=True) + except Exception: + logging.info("Failed to connect to local https server\"") + raise + + try: + dut.expect('https_request using saved client session', timeout=20) + dut.expect(['Connection established...', + 'Reading HTTP response...', + 'HTTP/1.1 200 OK', + 'connection closed'], expect_all=True) + except Exception: + logging.info("Failed the test for \"https_request using saved client session\"") + raise + + logging.info("Passed the test for \"https_request using saved client session\"") + thread1.terminate() + + +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.ethernet +@pytest.mark.parametrize('config', ['ssldyn',], indirect=True) +def test_examples_protocol_https_request_dynamic_buffers(dut: Dut) -> None: + # Check for connection using crt bundle with mbedtls dynamic resource enabled + # check and log bin size + binary_file = os.path.join(dut.app.binary_path, 'https_request.bin') + bin_size = os.path.getsize(binary_file) + logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024)) + + dut.expect('Loaded app from partition at offset', timeout=30) + try: + ip_address = dut.expect(r' (sta|eth) ip: (\d+\.\d+\.\d+\.\d+)', timeout=60)[2].decode() + print('Connected to AP with IP: {}'.format(ip_address)) + except pexpect.exceptions.TIMEOUT: + raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + + # only check if one connection is established + logging.info("Testing for \"https_request using crt bundle\" with mbedtls dynamic resource enabled") + try: + dut.expect('https_request using crt bundle', timeout=30) + dut.expect(['Connection established...', + 'Reading HTTP response...', + 'HTTP/1.1 200 OK', + 'connection closed'], expect_all=True) + except Exception: + logging.info("Failed the test for \"https_request using crt bundle\" when mbedtls dynamic resource was enabled") + raise + logging.info("Passed the test for \"https_request using crt bundle\" when mbedtls dynamic resource was enabled") + + +@pytest.mark.supported_targets +@pytest.mark.ethernet +def test_examples_protocol_https_request(dut: Dut) -> None: + + """ + steps: | + 1. join AP + 2. establish TLS connection to www.howsmyssl.com:443 with multiple + certificate verification options + 3. send http request + """ + # check and log bin size + binary_file = os.path.join(dut.app.binary_path, 'https_request.bin') + bin_size = os.path.getsize(binary_file) + logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024)) + logging.info('Starting https_request simple test app') + + dut.expect('Loaded app from partition at offset', timeout=30) + try: + ip_address = dut.expect(r' (sta|eth) ip: (\d+\.\d+\.\d+\.\d+)', timeout=60)[2].decode() + print('Connected to AP with IP: {}'.format(ip_address)) + except pexpect.exceptions.TIMEOUT: + raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') + + # Check for connection using crt bundle + logging.info("Testing for \"https_request using crt bundle\"") + try: + dut.expect('https_request using crt bundle', timeout=30) + dut.expect(['Certificate validated', + 'Connection established...', + 'Reading HTTP response...', + 'HTTP/1.1 200 OK', + 'connection closed'], expect_all=True) + except Exception: + logging.info("Failed the test for \"https_request using crt bundle\"") + raise + logging.info("Passed the test for \"https_request using crt bundle\"") + + # Check for connection using cacert_buf + logging.info("Testing for \"https_request using cacert_buf\"") + try: + dut.expect('https_request using cacert_buf', timeout=20) + dut.expect(['Connection established...', + 'Reading HTTP response...', + 'HTTP/1.1 200 OK', + 'connection closed'], expect_all=True) + except Exception: + logging.info("Passed the test for \"https_request using cacert_buf\"") + raise + logging.info("Passed the test for \"https_request using cacert_buf\"") + + # Check for connection using global ca_store + logging.info("Testing for \"https_request using global ca_store\"") + try: + dut.expect('https_request using global ca_store', timeout=20) + dut.expect(['Connection established...', + 'Reading HTTP response...', + 'HTTP/1.1 200 OK', + 'connection closed'], expect_all=True) + except Exception: + logging.info("Failed the test for \"https_request using global ca_store\"") + raise + logging.info("Passed the test for \"https_request using global ca_store\"") diff --git a/examples/protocols/https_server/simple/README.md b/examples/protocols/https_server/simple/README.md index fce84c0aac..5c3ab84f0a 100644 --- a/examples/protocols/https_server/simple/README.md +++ b/examples/protocols/https_server/simple/README.md @@ -1,3 +1,5 @@ +| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 | +| ----------------- | ----- | -------- | -------- | -------- | # HTTP server with SSL support using OpenSSL This example creates a SSL server that returns a simple HTML page when you visit its root URL. diff --git a/examples/protocols/https_server/simple/example_test.py b/examples/protocols/https_server/simple/pytest_https_server_simple.py similarity index 68% rename from examples/protocols/https_server/simple/example_test.py rename to examples/protocols/https_server/simple/pytest_https_server_simple.py index 1f4b509dda..d970391e01 100644 --- a/examples/protocols/https_server/simple/example_test.py +++ b/examples/protocols/https_server/simple/pytest_https_server_simple.py @@ -4,13 +4,12 @@ # SPDX-License-Identifier: Apache-2.0 import http.client +import logging import os -import re import ssl -import tiny_test_fw -import ttfw_idf -from tiny_test_fw import Utility +import pytest +from pytest_embedded import Dut server_cert_pem = '-----BEGIN CERTIFICATE-----\n'\ 'MIIDKzCCAhOgAwIBAgIUBxM3WJf2bP12kAfqhmhhjZWv0ukwDQYJKoZIhvcNAQEL\n'\ @@ -90,33 +89,36 @@ client_key_pem = '-----BEGIN PRIVATE KEY-----\n' \ success_response = '

Hello Secure World!

' -@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols') -def test_examples_protocol_https_server_simple(env, extra_data): # type: (tiny_test_fw.Env.Env, None) -> None # pylint: disable=unused-argument +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.wifi +def test_examples_protocol_https_server_simple(dut: Dut) -> None: """ steps: | 1. join AP 2. connect to www.howsmyssl.com:443 3. send http request """ - dut1 = env.get_dut('https_server_simple', 'examples/protocols/https_server/simple', dut_class=ttfw_idf.ESP32DUT) # check and log bin size - binary_file = os.path.join(dut1.app.binary_path, 'https_server.bin') + binary_file = os.path.join(dut.app.binary_path, 'https_server.bin') bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_server_simple_bin_size', '{}KB'.format(bin_size // 1024)) + logging.info('https_server_simple_bin_size : {}KB'.format(bin_size // 1024)) # start test - dut1.start_app() # Parse IP address and port of the server - dut1.expect(re.compile(r'Starting server')) - got_port = dut1.expect(re.compile(r'Server listening on port (\d+)'), timeout=30)[0] - Utility.console_log('Waiting to connect with AP') + dut.expect(r'Starting server') + got_port = int(dut.expect(r'Server listening on port (\d+)', timeout=30)[1].decode()) + logging.info('Waiting to connect with AP') + + got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode() - got_ip = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0] # Expected logs - Utility.console_log('Got IP : ' + got_ip) - Utility.console_log('Got Port : ' + got_port) + logging.info('Got IP : {}'.format(got_ip)) + logging.info('Got Port : {}'.format(got_port)) - Utility.console_log('Performing GET request over an SSL connection with the server') + logging.info('Performing GET request over an SSL connection with the server') CLIENT_CERT_FILE = 'client_cert.pem' CLIENT_KEY_FILE = 'client_key.pem' @@ -133,51 +135,61 @@ def test_examples_protocol_https_server_simple(env, extra_data): # type: (tiny_ ssl_context.load_cert_chain(certfile=CLIENT_CERT_FILE, keyfile=CLIENT_KEY_FILE) conn = http.client.HTTPSConnection(got_ip, got_port, context=ssl_context) - Utility.console_log('Performing SSL handshake with the server') + logging.info('Performing SSL handshake with the server') conn.request('GET','/') resp = conn.getresponse() - dut1.expect('performing session handshake') + dut.expect('performing session handshake') got_resp = resp.read().decode('utf-8') if got_resp != success_response: - Utility.console_log('Response obtained does not match with correct response') + logging.info('Response obtained does not match with correct response') raise RuntimeError('Failed to test SSL connection') - current_cipher = dut1.expect(re.compile(r'Current Ciphersuite(.*)'), timeout=5)[0] - Utility.console_log('Current Ciphersuite' + current_cipher) + current_cipher = dut.expect(r'Current Ciphersuite(.*)', timeout=5)[0] + logging.info('Current Ciphersuite {}'.format(current_cipher)) # Close the connection conn.close() - Utility.console_log('Checking user callback: Obtaining client certificate...') + logging.info('Checking user callback: Obtaining client certificate...') - serial_number = dut1.expect(re.compile(r'serial number(.*)'), timeout=5)[0] - issuer_name = dut1.expect(re.compile(r'issuer name(.*)'), timeout=5)[0] - expiry = dut1.expect(re.compile(r'expires on(.*)'), timeout=5)[0] + serial_number = dut.expect(r'serial number(.*)', timeout=5)[0] + issuer_name = dut.expect(r'issuer name(.*)', timeout=5)[0] + expiry = dut.expect(r'expires on ((.*)\d{4}\-(0?[1-9]|1[012])\-(0?[1-9]|[12][0-9]|3[01])*)', timeout=5)[1].decode() - Utility.console_log('Serial No.' + serial_number) - Utility.console_log('Issuer Name' + issuer_name) - Utility.console_log('Expires on' + expiry) + logging.info('Serial No. {}'.format(serial_number)) + logging.info('Issuer Name {}'.format(issuer_name)) + logging.info('Expires on {}'.format(expiry)) - Utility.console_log('Correct response obtained') - Utility.console_log('SSL connection test successful\nClosing the connection') + logging.info('Correct response obtained') + logging.info('SSL connection test successful\nClosing the connection') + +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.wifi +@pytest.mark.parametrize('config', ['dynamic_buffer',], indirect=True) +def test_examples_protocol_https_server_simple_dynamic_buffers(dut: Dut) -> None: # Test with mbedTLS dynamic buffer feature - dut1 = env.get_dut('https_server_simple', 'examples/protocols/https_server/simple', dut_class=ttfw_idf.ESP32DUT, app_config_name='dynamic_buffer') # start test - dut1.start_app() # Parse IP address and port of the server - dut1.expect(re.compile(r'Starting server')) - got_port = dut1.expect(re.compile(r'Server listening on port (\d+)'), timeout=30)[0] - Utility.console_log('Waiting to connect with AP') + dut.expect(r'Starting server') + got_port = int(dut.expect(r'Server listening on port (\d+)', timeout=30)[1].decode()) + logging.info('Waiting to connect with AP') + + got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode() - got_ip = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=30)[0] # Expected logs - Utility.console_log('Got IP : ' + got_ip) - Utility.console_log('Got Port : ' + got_port) + logging.info('Got IP : {}'.format(got_ip)) + logging.info('Got Port : {}'.format(got_port)) - Utility.console_log('Performing GET request over an SSL connection with the server') + logging.info('Performing GET request over an SSL connection with the server') + + CLIENT_CERT_FILE = 'client_cert.pem' + CLIENT_KEY_FILE = 'client_key.pem' ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) ssl_context.verify_mode = ssl.CERT_REQUIRED @@ -190,34 +202,30 @@ def test_examples_protocol_https_server_simple(env, extra_data): # type: (tiny_ os.remove(CLIENT_KEY_FILE) conn = http.client.HTTPSConnection(got_ip, got_port, context=ssl_context) - Utility.console_log('Performing SSL handshake with the server') + logging.info('Performing SSL handshake with the server') conn.request('GET','/') resp = conn.getresponse() - dut1.expect('performing session handshake') + dut.expect('performing session handshake') got_resp = resp.read().decode('utf-8') if got_resp != success_response: - Utility.console_log('Response obtained does not match with correct response') + logging.info('Response obtained does not match with correct response') raise RuntimeError('Failed to test SSL connection') - current_cipher = dut1.expect(re.compile(r'Current Ciphersuite(.*)'), timeout=5)[0] - Utility.console_log('Current Ciphersuite' + current_cipher) + current_cipher = dut.expect(r'Current Ciphersuite(.*)', timeout=5)[0] + logging.info('Current Ciphersuite {}'.format(current_cipher)) # Close the connection conn.close() - Utility.console_log('Checking user callback: Obtaining client certificate...') + logging.info('Checking user callback: Obtaining client certificate...') - serial_number = dut1.expect(re.compile(r'serial number(.*)'), timeout=5)[0] - issuer_name = dut1.expect(re.compile(r'issuer name(.*)'), timeout=5)[0] - expiry = dut1.expect(re.compile(r'expires on(.*)'), timeout=5)[0] + serial_number = dut.expect(r'serial number(.*)', timeout=5)[0] + issuer_name = dut.expect(r'issuer name(.*)', timeout=5)[0] + expiry = dut.expect(r'expires on ((.*)\d{4}\-(0?[1-9]|1[012])\-(0?[1-9]|[12][0-9]|3[01])*)', timeout=5)[1].decode() - Utility.console_log('Serial No.' + serial_number) - Utility.console_log('Issuer Name' + issuer_name) - Utility.console_log('Expires on' + expiry) + logging.info('Serial No. : {}'.format(serial_number)) + logging.info('Issuer Name : {}'.format(issuer_name)) + logging.info('Expires on : {}'.format(expiry)) - Utility.console_log('Correct response obtained') - Utility.console_log('SSL connection test successful\nClosing the connection') - - -if __name__ == '__main__': - test_examples_protocol_https_server_simple() # pylint: disable=no-value-for-parameter + logging.info('Correct response obtained') + logging.info('SSL connection test successful\nClosing the connection') diff --git a/examples/protocols/https_server/wss_server/README.md b/examples/protocols/https_server/wss_server/README.md index 781d8520e7..3d9c015561 100644 --- a/examples/protocols/https_server/wss_server/README.md +++ b/examples/protocols/https_server/wss_server/README.md @@ -1,3 +1,5 @@ +| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 | +| ----------------- | ----- | -------- | -------- | -------- | # HTTP Websocket server with SSL support This example creates a SSL server and employs a simple Websocket request handler. It demonstrates handling multiple clients from the server including: diff --git a/examples/protocols/https_server/wss_server/wss_server_example_test.py b/examples/protocols/https_server/wss_server/pytest_https_wss_server.py similarity index 63% rename from examples/protocols/https_server/wss_server/wss_server_example_test.py rename to examples/protocols/https_server/wss_server/pytest_https_wss_server.py index 2b4f9e994e..07e1cd52dd 100644 --- a/examples/protocols/https_server/wss_server/wss_server_example_test.py +++ b/examples/protocols/https_server/wss_server/pytest_https_wss_server.py @@ -5,17 +5,16 @@ from __future__ import division, print_function, unicode_literals +import logging import os -import re import threading import time from types import TracebackType from typing import Any, Optional -import tiny_test_fw -import ttfw_idf +import pytest import websocket -from tiny_test_fw import Utility +from pytest_embedded import Dut OPCODE_TEXT = 0x1 OPCODE_BIN = 0x2 @@ -74,12 +73,12 @@ class wss_client_thread(threading.Thread): if opcode == OPCODE_TEXT: if data == CORRECT_ASYNC_DATA: self.async_response = True - Utility.console_log('Thread {} obtained correct async message'.format(self.name)) + logging.info('Thread {} obtained correct async message'.format(self.name)) # Keep sending pong to update the keepalive in the server if (time.time() - self.start_time) > 20: break except Exception as e: - Utility.console_log('Failed to connect to the client and read async data') + logging.info('Failed to connect to the client and read async data') self.exc = e # type: ignore if self.async_response is not True: self.exc = RuntimeError('Failed to obtain correct async data') # type: ignore @@ -98,7 +97,7 @@ def test_multiple_client_keep_alive_and_async_response(ip, port, ca_file): # ty thread.start() threads.append(thread) except OSError: - Utility.console_log('Error: unable to start thread') + logging.info('Error: unable to start thread') # keep delay of 5 seconds between two connections to avoid handshake timeout time.sleep(5) @@ -106,76 +105,72 @@ def test_multiple_client_keep_alive_and_async_response(ip, port, ca_file): # ty t.join() -@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols') -def test_examples_protocol_https_wss_server(env, extra_data): # type: (tiny_test_fw.Env.Env, None) -> None # pylint: disable=unused-argument - - # Acquire DUT - dut1 = env.get_dut('https_server', 'examples/protocols/https_server/wss_server', dut_class=ttfw_idf.ESP32DUT) +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.wifi +def test_examples_protocol_https_wss_server(dut: Dut) -> None: # Get binary file - binary_file = os.path.join(dut1.app.binary_path, 'wss_server.bin') + binary_file = os.path.join(dut.app.binary_path, 'wss_server.bin') bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_wss_server_bin_size', '{}KB'.format(bin_size // 1024)) + logging.info('https_wss_server_bin_size : {}KB'.format(bin_size // 1024)) - # Upload binary and start testing - Utility.console_log('Starting wss_server test app') - dut1.start_app() + logging.info('Starting wss_server test app') # Parse IP address of STA - got_port = dut1.expect(re.compile(r'Server listening on port (\d+)'), timeout=60)[0] - Utility.console_log('Waiting to connect with AP') - got_ip = dut1.expect(re.compile(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)'), timeout=60)[0] + got_port = int(dut.expect(r'Server listening on port (\d+)', timeout=30)[1].decode()) + logging.info('Waiting to connect with AP') - Utility.console_log('Got IP : ' + got_ip) - Utility.console_log('Got Port : ' + got_port) + got_ip = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)', timeout=30)[1].decode() + + logging.info('Got IP : {}'.format(got_ip)) + logging.info('Got Port : {}'.format(got_port)) ca_file = os.path.join(os.path.dirname(__file__), 'main', 'certs', 'servercert.pem') # Start ws server test with WsClient(got_ip, int(got_port), ca_file) as ws: # Check for echo DATA = 'Espressif' - dut1.expect('performing session handshake') - client_fd = dut1.expect(re.compile(r'New client connected (\d+)'), timeout=20)[0] + dut.expect('performing session handshake') + client_fd = int(dut.expect(r'New client connected (\d+)', timeout=30)[1].decode()) ws.write(data=DATA, opcode=OPCODE_TEXT) - dut1.expect(re.compile(r'Received packet with message: {}'.format(DATA))) + dut.expect(r'Received packet with message: {}'.format(DATA)) opcode, data = ws.read() data = data.decode('UTF-8') if data != DATA: raise RuntimeError('Failed to receive the correct echo response') - Utility.console_log('Correct echo response obtained from the wss server') + logging.info('Correct echo response obtained from the wss server') # Test for keepalive - Utility.console_log('Testing for keep alive (approx time = 20s)') + logging.info('Testing for keep alive (approx time = 20s)') start_time = time.time() while True: try: opcode, data = ws.read() if opcode == OPCODE_PING: ws.write(data='Espressif', opcode=OPCODE_PONG) - Utility.console_log('Received PING, replying PONG (to update the keepalive)') + logging.info('Received PING, replying PONG (to update the keepalive)') # Keep sending pong to update the keepalive in the server if (time.time() - start_time) > 20: break except Exception: - Utility.console_log('Failed the test for keep alive,\nthe client got abruptly disconnected') + logging.info('Failed the test for keep alive,\nthe client got abruptly disconnected') raise # keepalive timeout is 10 seconds so do not respond for (10 + 1) senconds - Utility.console_log('Testing if client is disconnected if it does not respond for 10s i.e. keep_alive timeout (approx time = 11s)') + logging.info('Testing if client is disconnected if it does not respond for 10s i.e. keep_alive timeout (approx time = 11s)') try: - dut1.expect('Client not alive, closing fd {}'.format(client_fd), timeout=20) - dut1.expect('Client disconnected {}'.format(client_fd)) + dut.expect('Client not alive, closing fd {}'.format(client_fd), timeout=20) + dut.expect('Client disconnected {}'.format(client_fd)) except Exception: - Utility.console_log('ENV_ERROR:Failed the test for keep alive,\nthe connection was not closed after timeout') + logging.info('ENV_ERROR:Failed the test for keep alive,\nthe connection was not closed after timeout') time.sleep(11) - Utility.console_log('Passed the test for keep alive') + logging.info('Passed the test for keep alive') # Test keep alive and async response for multiple simultaneous client connections - Utility.console_log('Testing for multiple simultaneous client connections (approx time = 30s)') + logging.info('Testing for multiple simultaneous client connections (approx time = 30s)') test_multiple_client_keep_alive_and_async_response(got_ip, int(got_port), ca_file) - Utility.console_log('Passed the test for multiple simultaneous client connections') - - -if __name__ == '__main__': - test_examples_protocol_https_wss_server() # pylint: disable=no-value-for-parameter + logging.info('Passed the test for multiple simultaneous client connections') diff --git a/examples/protocols/https_x509_bundle/README.md b/examples/protocols/https_x509_bundle/README.md index a6fb6b819c..e1077477db 100644 --- a/examples/protocols/https_x509_bundle/README.md +++ b/examples/protocols/https_x509_bundle/README.md @@ -1,3 +1,5 @@ +| Supported Targets | ESP32 | ESP32-S2 | ESP32-S3 | ESP32-C3 | +| ----------------- | ----- | -------- | -------- | -------- | # HTTPS x509 Bundle Example This example shows how to use the ESP certificate bundle utility to embed a bundle of x509 certificates and use them to diff --git a/examples/protocols/https_x509_bundle/example_test.py b/examples/protocols/https_x509_bundle/example_test.py deleted file mode 100644 index 49823aeb09..0000000000 --- a/examples/protocols/https_x509_bundle/example_test.py +++ /dev/null @@ -1,40 +0,0 @@ -import os -import re - -import ttfw_idf - - -@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols', ignore=True) -def test_examples_protocol_https_x509_bundle(env, extra_data): - """ - steps: | - 1. join AP - 2. connect to multiple URLs - 3. send http request - """ - dut1 = env.get_dut('https_x509_bundle', 'examples/protocols/https_x509_bundle') - # check and log bin size - binary_file = os.path.join(dut1.app.binary_path, 'https_x509_bundle.bin') - bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_x509_bundle_bin_size', '{}KB'.format(bin_size // 1024)) - # start test - dut1.start_app() - num_URLS = dut1.expect(re.compile(r'Connecting to (\d+) URLs'), timeout=30) - dut1.expect(re.compile(r'Connection established to ([\s\S]*)'), timeout=30) - dut1.expect('Completed {} connections'.format(num_URLS[0]), timeout=60) - - # test mbedtls dynamic resource - dut1 = env.get_dut('https_x509_bundle', 'examples/protocols/https_x509_bundle', app_config_name='ssldyn') - # check and log bin size - binary_file = os.path.join(dut1.app.binary_path, 'https_x509_bundle.bin') - bin_size = os.path.getsize(binary_file) - ttfw_idf.log_performance('https_x509_bundle_bin_size', '{}KB'.format(bin_size // 1024)) - # start test - dut1.start_app() - num_URLS = dut1.expect(re.compile(r'Connecting to (\d+) URLs'), timeout=30) - dut1.expect(re.compile(r'Connection established to ([\s\S]*)'), timeout=30) - dut1.expect('Completed {} connections'.format(num_URLS[0]), timeout=60) - - -if __name__ == '__main__': - test_examples_protocol_https_x509_bundle() diff --git a/examples/protocols/https_x509_bundle/pytest_https_x509_bundle.py b/examples/protocols/https_x509_bundle/pytest_https_x509_bundle.py new file mode 100644 index 0000000000..385b10bf15 --- /dev/null +++ b/examples/protocols/https_x509_bundle/pytest_https_x509_bundle.py @@ -0,0 +1,47 @@ +# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD +# SPDX-License-Identifier: Unlicense OR CC0-1.0 +import logging +import os + +import pytest +from pytest_embedded import Dut + + +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.wifi +def test_examples_protocol_https_x509_bundle(dut: Dut) -> None: + """ + steps: | + 1. join AP + 2. connect to multiple URLs + 3. send http request + """ + # check and log bin size + binary_file = os.path.join(dut.app.binary_path, 'https_x509_bundle.bin') + bin_size = os.path.getsize(binary_file) + logging.info('https_x509_bundle_bin_size : {}KB'.format(bin_size // 1024)) + # start test + num_URLS = int(dut.expect(r'Connecting to (\d+) URLs', timeout=30)[1].decode()) + dut.expect(r'Connection established to ([\s\S]*)', timeout=30) + dut.expect('Completed {} connections'.format(num_URLS), timeout=60) + + +@pytest.mark.esp32 +@pytest.mark.esp32c3 +@pytest.mark.esp32s2 +@pytest.mark.esp32s3 +@pytest.mark.wifi +@pytest.mark.parametrize('config', ['ssldyn',], indirect=True) +def test_examples_protocol_https_x509_bundle_dynamic_buffer(dut: Dut) -> None: + # test mbedtls dynamic resource + # check and log bin size + binary_file = os.path.join(dut.app.binary_path, 'https_x509_bundle.bin') + bin_size = os.path.getsize(binary_file) + logging.info('https_x509_bundle_bin_size : {}KB'.format(bin_size // 1024)) + # start test + num_URLS = int(dut.expect(r'Connecting to (\d+) URLs', timeout=30)[1].decode()) + dut.expect(r'Connection established to ([\s\S]*)', timeout=30) + dut.expect('Completed {} connections'.format(num_URLS), timeout=60)