mirror of
https://github.com/espressif/esp-idf.git
synced 2026-06-03 03:36:34 +03:00
fix(https_request): write host time to NVS to eliminate SNTP dependency in CI
Replace erase_nvs + SNTP time sync with direct NVS timestamp injection
from the pytest host. This eliminates CI flakiness caused by NTP servers
being unreachable from the CI lab network.
Changes:
- Add write_time_to_nvs() helper that generates an NVS partition image
with the current host timestamp and flashes it to the DUT before each
test. The firmware reads this via the existing update_time_from_nvs()
path and skips SNTP entirely.
- Remove @pytest.mark.parametrize('erase_nvs', ['y']) from all 4
Ethernet-based tests since NVS is now written with valid data.
This commit is contained in:
hrushikesh.bhosale
@@ -4,10 +4,13 @@ import http.server
|
||||
import logging
|
||||
import multiprocessing
|
||||
import os
|
||||
import socket
|
||||
import ssl
|
||||
from typing import Callable
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
|
||||
import esptool
|
||||
import pexpect
|
||||
import pytest
|
||||
from common_test_methods import get_host_ip4_by_dest_ip
|
||||
@@ -29,14 +32,14 @@ def https_request_handler() -> Callable[..., http.server.BaseHTTPRequestHandler]
|
||||
if not self.wfile.closed:
|
||||
self.wfile.flush()
|
||||
self.wfile.close()
|
||||
except socket.error:
|
||||
except OSError:
|
||||
pass
|
||||
self.rfile.close()
|
||||
|
||||
def handle(self) -> None:
|
||||
try:
|
||||
RangeRequestHandler.handle(self)
|
||||
except socket.error:
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def do_GET(self) -> None:
|
||||
@@ -58,6 +61,50 @@ def start_https_server(server_file: str, key_file: str, server_ip: str, server_p
|
||||
httpd.serve_forever()
|
||||
|
||||
|
||||
def write_time_to_nvs(dut: Dut) -> None:
|
||||
"""Write current host timestamp to the DUT's NVS partition.
|
||||
|
||||
This eliminates the need for SNTP time synchronization in CI,
|
||||
where NTP servers may be unreachable. The firmware reads the
|
||||
timestamp from NVS key 'storage/timestamp' and uses it to set
|
||||
the system time for TLS certificate validation.
|
||||
"""
|
||||
nvs_offset = dut.app.partition_table['nvs']['offset']
|
||||
nvs_size = dut.app.partition_table['nvs']['size']
|
||||
|
||||
csv_file = os.path.join(tempfile.gettempdir(), 'nvs_time.csv')
|
||||
bin_file = os.path.join(tempfile.gettempdir(), 'nvs_time.bin')
|
||||
|
||||
with open(csv_file, 'w') as f:
|
||||
f.write('key,type,encoding,value\n')
|
||||
f.write('storage,namespace,,\n')
|
||||
f.write(f'timestamp,data,i64,{int(time.time())}\n')
|
||||
|
||||
nvs_gen = os.path.join(
|
||||
os.environ['IDF_PATH'],
|
||||
'components',
|
||||
'nvs_flash',
|
||||
'nvs_partition_generator',
|
||||
'nvs_partition_gen.py',
|
||||
)
|
||||
subprocess.check_call(
|
||||
['python3', nvs_gen, 'generate', csv_file, bin_file, hex(nvs_size)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
with dut.serial.disable_redirect_thread():
|
||||
dut.serial.esp.connect()
|
||||
esptool.main(
|
||||
['write-flash', '--no-compress', hex(nvs_offset), bin_file],
|
||||
esp=dut.serial.esp,
|
||||
)
|
||||
settings = dut.serial.proc.get_settings()
|
||||
dut.serial.esp.hard_reset()
|
||||
dut.serial.proc.apply_settings(settings)
|
||||
|
||||
logging.info('Wrote host timestamp (%d) to NVS at offset %s', int(time.time()), hex(nvs_offset))
|
||||
|
||||
|
||||
@pytest.mark.ethernet
|
||||
@pytest.mark.parametrize(
|
||||
'config',
|
||||
@@ -66,15 +113,15 @@ def start_https_server(server_file: str, key_file: str, server_ip: str, server_p
|
||||
],
|
||||
indirect=True,
|
||||
)
|
||||
@pytest.mark.parametrize('erase_nvs', ['y'], indirect=True)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None:
|
||||
write_time_to_nvs(dut)
|
||||
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))
|
||||
logging.info(f'https_request_bin_size : {bin_size // 1024}KB')
|
||||
# start https server
|
||||
server_port = 8070
|
||||
server_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_cert.pem')
|
||||
@@ -82,13 +129,13 @@ def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None:
|
||||
thread1 = multiprocessing.Process(target=start_https_server, args=(server_file, key_file, '0.0.0.0', server_port))
|
||||
thread1.daemon = True
|
||||
thread1.start()
|
||||
logging.info('The server started on localhost:{}'.format(server_port))
|
||||
logging.info(f'The server started on localhost:{server_port}')
|
||||
try:
|
||||
# start test
|
||||
dut.expect('Loaded app from partition at offset', timeout=30)
|
||||
try:
|
||||
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=60)[1].decode()
|
||||
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
|
||||
print(f'Connected to AP/Ethernet with IP: {ip_address}')
|
||||
except pexpect.exceptions.TIMEOUT:
|
||||
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
|
||||
host_ip = get_host_ip4_by_dest_ip(ip_address)
|
||||
@@ -132,19 +179,19 @@ def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None:
|
||||
],
|
||||
indirect=True,
|
||||
)
|
||||
@pytest.mark.parametrize('erase_nvs', ['y'], indirect=True)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_examples_protocol_https_request_dynamic_buffers(dut: Dut) -> None:
|
||||
write_time_to_nvs(dut)
|
||||
# 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))
|
||||
logging.info(f'https_request_bin_size : {bin_size // 1024}KB')
|
||||
|
||||
dut.expect('Loaded app from partition at offset', timeout=30)
|
||||
try:
|
||||
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=60)[1].decode()
|
||||
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
|
||||
print(f'Connected to AP/Ethernet with IP: {ip_address}')
|
||||
except pexpect.exceptions.TIMEOUT:
|
||||
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
|
||||
|
||||
@@ -163,7 +210,6 @@ def test_examples_protocol_https_request_dynamic_buffers(dut: Dut) -> None:
|
||||
|
||||
|
||||
@pytest.mark.ethernet
|
||||
@pytest.mark.parametrize('erase_nvs', ['y'], indirect=True)
|
||||
@idf_parametrize('target', ['esp32'], indirect=['target'])
|
||||
def test_examples_protocol_https_request(dut: Dut) -> None:
|
||||
"""
|
||||
@@ -173,16 +219,17 @@ def test_examples_protocol_https_request(dut: Dut) -> None:
|
||||
certificate verification options
|
||||
3. send http request
|
||||
"""
|
||||
write_time_to_nvs(dut)
|
||||
# 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(f'https_request_bin_size : {bin_size // 1024}KB')
|
||||
logging.info('Starting https_request simple test app')
|
||||
|
||||
dut.expect('Loaded app from partition at offset', timeout=30)
|
||||
try:
|
||||
ip_address = dut.expect(r'IPv4 address: (\d+\.\d+\.\d+\.\d+)[^\d]', timeout=60)[1].decode()
|
||||
print('Connected to AP/Ethernet with IP: {}'.format(ip_address))
|
||||
print(f'Connected to AP/Ethernet with IP: {ip_address}')
|
||||
except pexpect.exceptions.TIMEOUT:
|
||||
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP/Ethernet')
|
||||
|
||||
|
||||
Reference in New Issue
Block a user