feat(factory): support USB CDC alongside UART CLI
This commit is contained in:
chenjintao
@@ -0,0 +1,218 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Compile actual console driver and CLI framing/handlers with UART/CDC stubs."""
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def extract(source, name):
|
||||
match = re.search(r'^(?:static )?[\w *]+\b' + name + r'\([^;]*?\)\s*\{', source, re.M)
|
||||
assert match, name
|
||||
end, depth = match.end(), 1
|
||||
while depth:
|
||||
depth += (source[end] == '{') - (source[end] == '}')
|
||||
end += 1
|
||||
return source[match.start():end]
|
||||
|
||||
|
||||
STUBS = r'''
|
||||
#include <assert.h>
|
||||
#include <stdbool.h>
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <ctype.h>
|
||||
#define CONFIG_VFS_SUPPORT_IO 1
|
||||
#define ESP_OK 0
|
||||
#define ESP_FAIL 1
|
||||
#define ESP_ERR_TIMEOUT 2
|
||||
#define UART_NUM_0 0
|
||||
#define portMUX_INITIALIZER_UNLOCKED 0
|
||||
#define portENTER_CRITICAL(p) ((void)(p))
|
||||
#define portEXIT_CRITICAL(p) ((void)(p))
|
||||
#define pdTRUE 1
|
||||
#define pdMS_TO_TICKS(n) (n)
|
||||
#define ESP_LOGI(tag, ...) do { (void)(tag); if(0) printf(__VA_ARGS__); } while(0)
|
||||
#define ESP_LOGW ESP_LOGI
|
||||
#define ESP_LOGE ESP_LOGI
|
||||
#define FACTORY_CLI_MAX_LINE_LEN 512
|
||||
#define FACTORY_CLI_MAX_ARGC 8
|
||||
#define FACTORY_CLI_DEVICE_ID_MAX_LEN 64
|
||||
#define FACTORY_CLI_SERIAL_MAX_LEN 64
|
||||
#define FACTORY_CLI_INPUT_BURST_BYTES 64
|
||||
#define FACTORY_CLI_PARTIAL_TIMEOUT_MS 2000
|
||||
#define FACTORY_CLI_JSON_MAX_LEN 640
|
||||
#define DOMAIN_EXPRESSION_CHOUQIAN 1
|
||||
#define DOMAIN_EXPRESSION_TRIGGER_FACTORY_LCD_TEST 2
|
||||
typedef int esp_err_t;
|
||||
typedef uint32_t TickType_t;
|
||||
typedef int StaticSemaphore_t;
|
||||
typedef int *SemaphoreHandle_t;
|
||||
typedef int portMUX_TYPE;
|
||||
typedef struct {uint32_t rx_buffer_size,tx_buffer_size;} usb_serial_jtag_driver_config_t;
|
||||
typedef struct {int id,trigger;bool chouqian_result_index_valid;size_t chouqian_result_index;} domain_expression_request_t;
|
||||
static TickType_t ticks;
|
||||
static bool uart_ready,usb_ready,lock_fail,usb_block;
|
||||
static int uart_error,usb_error,uart_installs,usb_installs;
|
||||
static const char *uart_rx="", *usb_rx="";
|
||||
static char uart_tx[16384],usb_tx[16384];
|
||||
static size_t uart_tx_len,usb_tx_len;
|
||||
static int saves,notifies;
|
||||
static char saved_id[64],saved_sn[64];
|
||||
static TickType_t xTaskGetTickCount(void) {return ticks;}
|
||||
#if CONFIG_TQ_FACTORY_CLI_USB_CDC_ENABLE
|
||||
static void vTaskDelay(TickType_t delay) {ticks+=delay;}
|
||||
#endif
|
||||
static SemaphoreHandle_t xSemaphoreCreateMutexStatic(StaticSemaphore_t *p) {return p;}
|
||||
static int xSemaphoreTake(SemaphoreHandle_t p, TickType_t wait) {assert(p && wait<=100);return !lock_fail;}
|
||||
static int xSemaphoreGive(SemaphoreHandle_t p) {assert(p);return 1;}
|
||||
static const char *esp_err_to_name(int err) {(void)err;return "fake";}
|
||||
static bool uart_is_driver_installed(int port) {assert(port==0);return uart_ready;}
|
||||
static int uart_driver_install(int port,int rx,int tx,int q,void *handle,int flags) {
|
||||
assert(port==0 && rx==2048 && tx==0 && q==0 && !handle && !flags);
|
||||
++uart_installs;uart_ready=!uart_error;return uart_error;
|
||||
}
|
||||
static int uart_read_bytes(int port,void *buf,size_t len,TickType_t wait) {
|
||||
assert(port==0 && len>0);(void)wait;
|
||||
if(!*uart_rx)return 0; *(char *)buf=*uart_rx++;return 1;
|
||||
}
|
||||
static int uart_write_bytes(int port,const void *buf,size_t len) {
|
||||
assert(port==0 && uart_tx_len+len<sizeof(uart_tx));
|
||||
memcpy(uart_tx+uart_tx_len,buf,len);uart_tx_len+=len;uart_tx[uart_tx_len]=0;return (int)len;
|
||||
}
|
||||
#if CONFIG_TQ_FACTORY_CLI_USB_CDC_ENABLE
|
||||
static void usb_serial_jtag_vfs_use_driver(void) {}
|
||||
static bool usb_serial_jtag_is_driver_installed(void) {return usb_ready;}
|
||||
static int usb_serial_jtag_driver_install(usb_serial_jtag_driver_config_t *p) {
|
||||
assert(p->rx_buffer_size==2048 && p->tx_buffer_size==1024);
|
||||
++usb_installs;usb_ready=!usb_error;return usb_error;
|
||||
}
|
||||
static int usb_serial_jtag_read_bytes(void *buf,uint32_t len,TickType_t wait) {
|
||||
assert(len>0);(void)wait;
|
||||
if(!*usb_rx)return 0;*(char *)buf=*usb_rx++;return 1;
|
||||
}
|
||||
static int usb_serial_jtag_write_bytes(const void *buf,size_t len,TickType_t wait) {
|
||||
assert(wait==0 && len<=64);if(usb_block)return 0;
|
||||
assert(usb_tx_len+len<sizeof(usb_tx));memcpy(usb_tx+usb_tx_len,buf,len);
|
||||
usb_tx_len+=len;usb_tx[usb_tx_len]=0;return (int)len;
|
||||
}
|
||||
#endif
|
||||
|
||||
#define strlcpy test_strlcpy
|
||||
static size_t test_strlcpy(char *dst,const char *src,size_t cap) {
|
||||
size_t n=strlen(src);if(cap){size_t c=n<cap-1?n:cap-1;memcpy(dst,src,c);dst[c]=0;}return n;
|
||||
}
|
||||
static void domain_runtime_read_device_identity(char *id,size_t ilen,char *sn,size_t slen) {
|
||||
strlcpy(id,saved_id,ilen);strlcpy(sn,saved_sn,slen);
|
||||
}
|
||||
static int domain_runtime_write_device_identity(const char *id,const char *sn) {
|
||||
++saves;strlcpy(saved_id,id,sizeof(saved_id));strlcpy(saved_sn,sn,sizeof(saved_sn));return ESP_OK;
|
||||
}
|
||||
static void control_plane_provisioning_configuration_changed(void) {++notifies;}
|
||||
static void control_plane_multimodal_device_auth_invalidate(int reason) {(void)reason;}
|
||||
static int domain_runtime_request_expression(const domain_expression_request_t *p) {(void)p;return ESP_OK;}
|
||||
static void handle_get_wifi(void) {}
|
||||
static void handle_set_wifi(size_t n,char **a) {(void)n;(void)a;}
|
||||
static void handle_reset_wifi(void) {}
|
||||
static void handle_get_auth_material(void) {}
|
||||
static void handle_sign_challenge(const char *a,const char *b) {(void)a;(void)b;}
|
||||
static void handle_prepare_rotate(uint32_t v) {(void)v;}
|
||||
static void handle_lcd_test(void) {}
|
||||
'''
|
||||
|
||||
BRIDGE = r'''
|
||||
static int domain_runtime_console_read_bytes(uint8_t *buf,size_t len,int ms) {return platform_console_read_bytes(buf,len,ms);}
|
||||
static int domain_runtime_console_read_cdc_bytes(uint8_t *buf,size_t len,int ms) {return platform_console_read_cdc_bytes(buf,len,ms);}
|
||||
static void domain_runtime_console_write_line(const char *p) {platform_console_write_line(p);}
|
||||
static void domain_runtime_console_write_cdc_line(const char *p) {platform_console_write_cdc_line(p);}
|
||||
'''
|
||||
|
||||
TESTS = r'''
|
||||
static void pump(void) {
|
||||
for(int i=0;i<100;++i) {factory_cli_poll_input(&s_uart_session,false);factory_cli_poll_input(&s_cdc_session,true);}
|
||||
}
|
||||
static void clear_outputs(void) {uart_tx_len=usb_tx_len=0;uart_tx[0]=usb_tx[0]=0;}
|
||||
int main(void) {
|
||||
assert(platform_console_install()==ESP_OK);
|
||||
assert(platform_console_install()==ESP_OK && uart_installs==1);
|
||||
#if CONFIG_TQ_FACTORY_CLI_USB_CDC_ENABLE
|
||||
assert(usb_installs==1);
|
||||
usb_ready=false;usb_error=ESP_FAIL;
|
||||
assert(platform_console_install()==ESP_OK && uart_ready); // UART fallback
|
||||
usb_error=0;uart_ready=false;uart_error=ESP_FAIL;
|
||||
assert(platform_console_install()==ESP_OK && usb_ready); // CDC fallback
|
||||
uart_error=0;assert(platform_console_install()==ESP_OK);
|
||||
usb_ready=uart_ready=false;usb_error=uart_error=ESP_FAIL;
|
||||
assert(platform_console_install()!=ESP_OK);
|
||||
usb_error=uart_error=0;assert(platform_console_install()==ESP_OK);
|
||||
#endif
|
||||
lock_fail=true;assert(platform_console_install()==ESP_ERR_TIMEOUT);lock_fail=false;
|
||||
uint8_t ch;
|
||||
assert(platform_console_read_bytes(NULL,1,0)<0);
|
||||
assert(platform_console_read_bytes(&ch,0,0)<0);
|
||||
assert(platform_console_read_bytes(&ch,1,-1)<0);
|
||||
uart_rx="get_device_info\r\n";pump();
|
||||
assert(strstr(uart_tx,"device_id") && strstr(uart_tx,"OK\r\n"));
|
||||
assert(usb_tx_len==0);
|
||||
#if CONFIG_TQ_FACTORY_CLI_USB_CDC_ENABLE
|
||||
clear_outputs();usb_rx="get_device_info\n";pump();
|
||||
assert(strstr(usb_tx,"device_id") && strstr(usb_tx,"OK\r\n") && uart_tx_len==0);
|
||||
assert(platform_console_read_cdc_bytes(NULL,1,0)<0);
|
||||
assert(platform_console_read_cdc_bytes(&ch,1,-1)<0);
|
||||
// Partial lines do not cross ports, and pending identities are isolated too.
|
||||
clear_outputs();uart_rx="set_device_id uart-id";pump();
|
||||
usb_rx="set_serial_number cdc-sn\nsave_config\n";pump();
|
||||
assert(saves==0 && strstr(usb_tx,"ERROR"));
|
||||
uart_rx="\nset_serial_number uart-sn\nsave_config\n";pump();
|
||||
assert(saves==1 && notifies==1 && !strcmp(saved_id,"uart-id") && !strcmp(saved_sn,"uart-sn"));
|
||||
// A saturated/disconnected USB writer has a fixed budget; UART remains usable.
|
||||
clear_outputs();usb_block=true;TickType_t start=ticks;
|
||||
platform_console_write_cdc_line("reply");
|
||||
assert(ticks-start==250 && !usb_tx_len);
|
||||
usb_block=false;uart_rx="get_device_info\n";pump();assert(strstr(uart_tx,"OK"));
|
||||
#else
|
||||
assert(platform_console_read_cdc_bytes(&ch,1,0)<0);
|
||||
platform_console_write_cdc_line("disabled");assert(!usb_tx_len);
|
||||
#endif
|
||||
// Oversize input must discard the whole line, including any valid-looking tail.
|
||||
char oversized[700];memset(oversized,'x',600);strcpy(oversized+600,"save_config\n");
|
||||
int before=saves;uart_rx=oversized;pump();assert(saves==before);
|
||||
// A delayed partial command is discarded, not joined to a later command.
|
||||
uart_rx="save_";pump();ticks+=2000;pump();uart_rx="config\n";pump();assert(saves==before);
|
||||
clear_outputs();uart_rx="get_device_info\n";pump();assert(strstr(uart_tx,"OK"));
|
||||
puts("PASS: actual driver/CLI init, UART/CDC routing, session isolation, bounded USB writes, overflow and partial-line recovery");
|
||||
}
|
||||
'''
|
||||
|
||||
|
||||
def main():
|
||||
driver = (ROOT / 'components/platform/src/platform_console_driver.c').read_text()
|
||||
driver = re.sub(r'^#include .*$', '', driver, flags=re.M)
|
||||
cli = (ROOT / 'components/control_plane/src/runtime/control_plane_factory_cli.c').read_text()
|
||||
session = re.search(r'typedef struct \{[^}]+\} factory_cli_session_t;', cli).group()
|
||||
state = session + '\nstatic factory_cli_session_t s_uart_session, s_cdc_session;\n' \
|
||||
'static factory_cli_session_t *s_current_session=&s_uart_session;\nstatic bool s_reply_over_cdc;\n'
|
||||
names = ['is_text_empty', 'trim_text', 'split_args', 'cli_write_line', 'cli_write_ok',
|
||||
'cli_write_error', 'handle_get_device_info', 'handle_save_config',
|
||||
'handle_command_line', 'factory_cli_poll_input']
|
||||
actual = '\n'.join(extract(cli, n) for n in names)
|
||||
with tempfile.TemporaryDirectory(prefix='factory-cdc-') as tmp:
|
||||
path = Path(tmp)
|
||||
cfile = path / 'test.c'
|
||||
cfile.write_text(STUBS + driver + BRIDGE + state + actual + TESTS)
|
||||
for enabled in (0, 1):
|
||||
binary = path / f'test-{enabled}'
|
||||
subprocess.run(['clang', '-std=c11', '-Wall', '-Wextra', '-Werror',
|
||||
'-Wno-unused-variable', '-fsanitize=address,undefined',
|
||||
f'-DCONFIG_TQ_FACTORY_CLI_USB_CDC_ENABLE={enabled}',
|
||||
str(cfile), '-o', str(binary)], check=True)
|
||||
subprocess.run([str(binary)], check=True)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user