mirror of
https://github.com/espressif/esp-idf.git
synced 2024-10-05 20:47:46 -04:00
Merge branch 'feature/https_example_pytest_migration' into 'master'
https example pytest migration See merge request espressif/esp-idf!18054
This commit is contained in:
Mahavir Jain
commit
7640971f14
@ -218,6 +218,7 @@ before_script:
|
||||
"pytest-embedded-qemu~=$PYTEST_EMBEDDED_VERSION"
|
||||
pytest-rerunfailures
|
||||
scapy
|
||||
websocket-client
|
||||
-r tools/esp_prov/requirements.txt
|
||||
- export EXTRA_CFLAGS=${PEDANTIC_CFLAGS}
|
||||
- export EXTRA_CXXFLAGS=${PEDANTIC_CXXFLAGS}
|
||||
|
||||
@ -41,8 +41,9 @@ build_pytest_examples_esp32:
|
||||
extends:
|
||||
- .build_pytest_template
|
||||
- .rules:build:example_test-esp32
|
||||
parallel: 2
|
||||
script:
|
||||
- run_cmd python tools/ci/build_pytest_apps.py examples --target esp32 --size-info $SIZE_INFO_LOCATION -vv
|
||||
- run_cmd python tools/ci/build_pytest_apps.py examples --target esp32 --size-info $SIZE_INFO_LOCATION -vv --parallel-count $CI_NODE_TOTAL --parallel-index $CI_NODE_INDEX
|
||||
|
||||
build_pytest_examples_esp32s2:
|
||||
extends:
|
||||
|
||||
@ -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.)
|
||||
|
||||
@ -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
|
||||
42
examples/protocols/https_mbedtls/pytest_https_mbedtls.py
Normal file
42
examples/protocols/https_mbedtls/pytest_https_mbedtls.py
Normal file
@ -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')
|
||||
@ -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.
|
||||
|
||||
@ -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()
|
||||
234
examples/protocols/https_request/pytest_https_request.py
Normal file
234
examples/protocols/https_request/pytest_https_request.py
Normal file
@ -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\"")
|
||||
@ -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.
|
||||
|
||||
@ -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 = '<h1>Hello Secure World!</h1>'
|
||||
|
||||
|
||||
@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')
|
||||
@ -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:
|
||||
|
||||
@ -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')
|
||||
@ -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
|
||||
|
||||
@ -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()
|
||||
@ -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)
|
||||
Loading…
Reference in New Issue
Block a user