63 lines
3.1 KiB
Python
63 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Production idle lifecycle/sweep + installed IDF completion path, real host sockets.
|
|
|
|
Deterministic owner/queue/TLS doubles, not a TLS server or hardware timing test.
|
|
No dependency installation, device operation, or SDK modification.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
ROOT = HERE.parents[1]
|
|
IDF = Path(os.environ.get('IDF_PATH', str(Path.home() / '.platformio/packages/framework-espidf')))
|
|
|
|
|
|
def stripped(path):
|
|
return '\n'.join(line for line in path.read_text().splitlines()
|
|
if not line.startswith(('#include', '#pragma once')))
|
|
|
|
|
|
def function(source, name):
|
|
match = re.search(r'^(?:static )?(?:void|esp_err_t) ' + name + r'\(.*?^\}', source, re.M | re.S)
|
|
assert match, name
|
|
return match.group() + '\n'
|
|
|
|
|
|
sess = (IDF / 'components/esp_http_server/src/httpd_sess.c').read_text()
|
|
parse = (IDF / 'components/esp_http_server/src/httpd_parse.c').read_text()
|
|
main = (IDF / 'components/esp_http_server/src/httpd_main.c').read_text()
|
|
ssl = (IDF / 'components/esp_https_server/src/https_server.c').read_text()
|
|
completion = function(sess, 'httpd_sess_process')
|
|
assert completion.index('httpd_req_new') < completion.index('httpd_req_delete') < completion.index('session->lru_counter = ++hd->lru_counter')
|
|
assert 'Only listen for new connections if server has capacity' in main
|
|
assert main.index('/* Case0:') < main.index('/* Case1:') < main.index('/* Case2:')
|
|
assert 'httpd_sess_delete(hd, sock_db);' in sess # queued reusable-pointer close is not safe here
|
|
assert ssl.index('httpd_sess_set_pending_override') < ssl.index('HTTPD_SSL_USER_CB_SESS_CREATE')
|
|
adapter = (ROOT / 'src/web_httpd_adapter.c').read_text()
|
|
idle = (ROOT / 'src/web_httpd_idle.c').read_text()
|
|
assert 'ESP_IDF_VERSION_VAL(5, 5, 0)' in adapter
|
|
for forbidden in ('httpd_sess_trigger_close', 'web_diagnostics', 'xTaskCreate', 'malloc(', 'calloc(', 'ESP_LOG'):
|
|
assert forbidden not in idle, forbidden
|
|
sweep = function(adapter, 'web_httpd_idle_sweep')
|
|
assert 'httpd_sess_trigger_close' not in sweep
|
|
assert 'web_httpd_idle_tls(arg);\n web_diagnostics_tls(arg);' in (ROOT / 'src/web_server.c').read_text()
|
|
assert 'setInterval(pollStatus, 5000)' in (ROOT / 'src/web_ui.c').read_text()
|
|
|
|
with tempfile.TemporaryDirectory(prefix='web-httpd-idle-') as directory:
|
|
directory = Path(directory)
|
|
unit = directory / 'test.c'
|
|
unit.write_text((HERE / 'fakes.h').read_text() + '\n' +
|
|
stripped(ROOT / 'src/web_httpd_adapter.h') + '\n' +
|
|
stripped(ROOT / 'src/web_httpd_idle.h') + '\n' +
|
|
sweep + '\n' + stripped(ROOT / 'src/web_httpd_idle.c') + '\n' +
|
|
function(parse, 'httpd_req_delete') + '\n' + completion + '\n' +
|
|
(HERE / 'test.c').read_text())
|
|
executable = directory / 'test'
|
|
subprocess.run([os.environ.get('CC', 'cc'), '-std=c11', '-Wall', '-Wextra', '-Werror',
|
|
'-g', str(unit), '-o', str(executable)], check=True, timeout=30)
|
|
subprocess.run([str(executable)], check=True, timeout=20)
|
|
print('PASS installed SDK completion/owner-order and production integration guards')
|