alex/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2025-08-09 20:41:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

https examples pytest migration

Browse Source
This commit is contained in:
Harshit Malpani
2022-05-10 12:16:56 +05:30
parent 9177e9fb37
commit 5a51af523c
13 changed files with 434 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
examples/protocols/https_server/simple/README.md
View File

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

124
examples/protocols/https_server/simple/example_test.py → examples/protocols/https_server/simple/pytest_https_server_simple.py
View File

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

2
examples/protocols/https_server/wss_server/README.md
View File

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

75
examples/protocols/https_server/wss_server/wss_server_example_test.py → examples/protocols/https_server/wss_server/pytest_https_wss_server.py
View File

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