106 lines
5.1 KiB
Python
106 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Actual canonical stop/reboot handlers with deterministic side-effect doubles."""
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
def function(path, name):
|
|
source = path.read_text()
|
|
start = source.index('static int ' + name + '(')
|
|
return source[start:source.index('\n}', start) + 2]
|
|
|
|
header = '\n'.join(line for line in (ROOT / 'src/admin_ssh_console.h').read_text().splitlines()
|
|
if not line.startswith(('#include', '#pragma once')))
|
|
prelude = r'''
|
|
#include <assert.h>
|
|
#include <stdbool.h>
|
|
#include <stddef.h>
|
|
#include <stdint.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
typedef int esp_err_t;
|
|
enum { ESP_OK=0, ESP_FAIL=-1, ESP_ERR_TIMEOUT=7 };
|
|
typedef struct { int unused; } user_principal_t;
|
|
'''
|
|
fakes = r'''
|
|
static bool remote, web;
|
|
static unsigned stops, reboots, scheduled, waits, rotations;
|
|
static esp_err_t schedule_result, stop_result;
|
|
static admin_ssh_deferred_action_type_t last_action;
|
|
bool admin_ssh_console_dispatch_is_remote(void) { return remote; }
|
|
bool admin_ssh_console_dispatch_is_web(void) { return remote && web; }
|
|
esp_err_t admin_ssh_console_dispatch_defer(admin_ssh_deferred_action_type_t action, uint32_t argument) {
|
|
assert(remote && !argument); ++scheduled; last_action=action; return schedule_result;
|
|
}
|
|
static const char *esp_err_to_name(esp_err_t error) { (void)error; return "fake"; }
|
|
static esp_err_t web_server_stop(void) { ++stops; return stop_result; }
|
|
static esp_err_t web_server_start(void) { assert(false); return ESP_FAIL; }
|
|
static esp_err_t web_server_clear_counters(void) { assert(false); return ESP_FAIL; }
|
|
static esp_err_t web_serial_transport_clear_counters(void) { assert(false); return ESP_FAIL; }
|
|
static void esp_restart(void) { ++reboots; }
|
|
static void vTaskDelay(unsigned delay) { assert(delay==100); ++waits; }
|
|
#define pdMS_TO_TICKS(ms) (ms)
|
|
static void print_usage(void) { assert(false); }
|
|
static int web_diagnostics_command(const char *action) { assert(!strcmp(action, "show")); return 0; }
|
|
static int show_status(void) { assert(false); return 1; }
|
|
static int show_counters(void) { assert(false); return 1; }
|
|
static int show_credentials(void) { assert(false); return 1; }
|
|
static int show_certificate(void) { assert(false); return 1; }
|
|
static int rotate_credentials(void) { assert(false); return 1; }
|
|
static int rotate_certificate(void) { ++rotations; return 0; }
|
|
static int reset_material(void) { assert(false); return 1; }
|
|
static bool force_is_present(int argc, char **argv, int expected) {
|
|
return argc == expected && !strcmp(argv[expected - 1], "--force");
|
|
}
|
|
'''
|
|
tests = r'''
|
|
int main(void) {
|
|
char *diagnostics[]={"web", "diagnostics", "show"};
|
|
assert(command_web(3, diagnostics)==0 && !stops && !scheduled);
|
|
char *stop[]={"web", "stop"};
|
|
remote=web=true;
|
|
assert(command_web(2,stop)==0 && scheduled==1 && !stops && last_action==ADMIN_CONSOLE_DEFER_WEB_STOP);
|
|
schedule_result=ESP_ERR_TIMEOUT;
|
|
assert(command_web(2,stop)==1 && scheduled==2 && !stops);
|
|
schedule_result=ESP_OK;
|
|
web=false; /* SSH preserves its synchronous HTTPS path. */
|
|
assert(command_web(2,stop)==0 && stops==1 && scheduled==2);
|
|
remote=false;
|
|
assert(command_web(2,stop)==0 && stops==2 && scheduled==2);
|
|
stop_result=ESP_FAIL;
|
|
assert(command_web(2,stop)==1 && stops==3);
|
|
remote=true;
|
|
assert(command_reboot(1,NULL)==0 && scheduled==3 && !reboots && last_action==ADMIN_SSH_DEFER_REBOOT);
|
|
web=true;
|
|
assert(command_reboot(1,NULL)==0 && scheduled==4 && !reboots && last_action==ADMIN_SSH_DEFER_REBOOT);
|
|
schedule_result=ESP_FAIL;
|
|
assert(command_reboot(1,NULL)==1 && scheduled==5 && !reboots);
|
|
assert(command_reboot(2,NULL)==1 && scheduled==5 && !reboots);
|
|
remote=false;
|
|
assert(command_reboot(1,NULL)==0 && reboots==1 && waits==1 && scheduled==5);
|
|
char *rotate[]={"web", "certificate", "rotate", "--force", "extra"};
|
|
remote=web=true; schedule_result=ESP_OK;
|
|
assert(command_web(4,rotate)==0 && scheduled==6 && !rotations &&
|
|
last_action==ADMIN_CONSOLE_DEFER_WEB_CERTIFICATE_ROTATE);
|
|
schedule_result=ESP_ERR_TIMEOUT;
|
|
assert(command_web(4,rotate)==1 && scheduled==7 && !rotations);
|
|
assert(command_web(3,rotate)==1 && scheduled==7 && !rotations);
|
|
assert(command_web(5,rotate)==1 && scheduled==7 && !rotations);
|
|
web=false;
|
|
assert(command_web(4,rotate)==0 && rotations==1 && scheduled==7);
|
|
remote=false;
|
|
assert(command_web(4,rotate)==0 && rotations==2 && scheduled==7);
|
|
puts("PASS: canonical WEB stop/certificate deferred, exact force required, SSH/UART unchanged, reboot and queue failure isolation");
|
|
}
|
|
'''
|
|
with tempfile.TemporaryDirectory(prefix='console-lifecycle-') as directory:
|
|
tmp = Path(directory)
|
|
(tmp / 'test.c').write_text(prelude + header + fakes +
|
|
function(ROOT / 'src/web_console.c', 'command_web') +
|
|
function(ROOT / 'src/system_console.c', 'command_reboot') + tests)
|
|
subprocess.run(['cc', '-std=c11', '-Wall', '-Wextra', '-Werror',
|
|
str(tmp / 'test.c'), '-o', str(tmp / 'test')], check=True, timeout=30)
|
|
subprocess.run([str(tmp / 'test')], check=True, timeout=10)
|