chenjintao
db6109172c
Track microphone levels, short reads and PCM buffer drops. Report audio queue admission separately from TLS writes, expose server replies and ACKs before filtering, and capture per-round counters before timeouts. Add diagnostic regression coverage and support POSIX hosts in WSS tests.
85 lines
3.8 KiB
Python
85 lines
3.8 KiB
Python
"""Compile actual SDK-independent function bodies with scripted TLS/RTOS stubs."""
|
|
from pathlib import Path
|
|
import subprocess
|
|
import os
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
lib = root / 'ql-application/threadx/interface/custom_app/lib/librws/src'
|
|
private = (lib / 'rws_socketpriv.c').read_bytes().decode('latin1')
|
|
public = (lib / 'rws_socketpub.c').read_bytes().decode('latin1')
|
|
send = private[private.index('static int rws_tls_send_all('):].split('#endif', 1)[0]
|
|
cancel = public[public.index('void rws_socket_disconnect_and_release('):].split('rws_bool rws_socket_send_text(', 1)[0]
|
|
harness = r'''
|
|
#include <assert.h>
|
|
#include <stddef.h>
|
|
#include <stdio.h>
|
|
#define MBEDTLS_ERR_SSL_WANT_READ -2
|
|
#define MBEDTLS_ERR_SSL_WANT_WRITE -3
|
|
#define COMMAND_NONE 0
|
|
#define COMMAND_DISCONNECT 6
|
|
#define COMMAND_END 9999
|
|
typedef struct {int ssl_ctx;} tls;
|
|
typedef struct {tls *ssl; int work_mutex, is_connected, worker_started, work_thread, command;} _rws_socket;
|
|
typedef _rws_socket *rws_socket;
|
|
static int replies[16], calls, reply_count, expected_offset, freed, locks;
|
|
static unsigned int ticks;
|
|
static const unsigned char bytes[] = "abcdef";
|
|
static unsigned int ql_rtos_get_systicks(void) { return ticks; }
|
|
static void rws_thread_sleep(int ms) { ticks += (unsigned int)ms; }
|
|
static int mbedtls_ssl_write(int *ctx, const unsigned char *data, size_t size) {
|
|
int ret;
|
|
(void)ctx;
|
|
assert(data == bytes + expected_offset);
|
|
assert(size == 6 - (size_t)expected_offset);
|
|
ret = calls < reply_count ? replies[calls] : MBEDTLS_ERR_SSL_WANT_WRITE;
|
|
++calls;
|
|
if (ret > 0) expected_offset += ret;
|
|
return ret;
|
|
}
|
|
static void rws_mutex_lock(int mutex) {(void)mutex; ++locks;}
|
|
static void rws_mutex_unlock(int mutex) {(void)mutex; --locks;}
|
|
static void rws_socket_delete(_rws_socket *s) {(void)s; assert(locks == 0); ++freed;}
|
|
'''
|
|
harness += send + cancel
|
|
harness += r'''
|
|
int main(void) {
|
|
tls ctx = {0};
|
|
_rws_socket s = {0};
|
|
s.ssl = &ctx;
|
|
replies[0] = -2; replies[1] = 2; replies[2] = -3; replies[3] = 4; reply_count = 4;
|
|
assert(rws_tls_send_all(&s, bytes, 6) == 6 && calls == 4 && expected_offset == 6);
|
|
calls = 0; expected_offset = 0; replies[0] = 2; replies[1] = -99; reply_count = 2;
|
|
assert(rws_tls_send_all(&s, bytes, 6) == -1 && calls == 2);
|
|
calls = 0; expected_offset = 0; replies[0] = 0; reply_count = 1;
|
|
assert(rws_tls_send_all(&s, bytes, 6) == -1 && calls == 1);
|
|
calls = 0; expected_offset = 0; reply_count = 0; ticks = 0;
|
|
assert(rws_tls_send_all(&s, bytes, 6) == -1 && ticks >= 1000 && calls < 1000);
|
|
calls = 0; ticks = (unsigned int)-500;
|
|
assert(rws_tls_send_all(&s, bytes, 6) == -1 && calls < 1000);
|
|
s.ssl = NULL; calls = 0;
|
|
assert(rws_tls_send_all(&s, bytes, 6) == -1 && calls == 0);
|
|
s.is_connected = 1; rws_socket_disconnect_and_release(&s);
|
|
assert(s.command == COMMAND_DISCONNECT && locks == 0 && !freed);
|
|
s.is_connected = 0; s.worker_started = 1; s.command = 1;
|
|
rws_socket_disconnect_and_release(&s);
|
|
assert(s.command == COMMAND_END && locks == 0 && !freed);
|
|
s.worker_started = 0; s.command = COMMAND_NONE;
|
|
rws_socket_disconnect_and_release(&s);
|
|
assert(freed == 1 && locks == 0);
|
|
rws_socket_disconnect_and_release(NULL);
|
|
puts("PASS: TLS partial writes, WANT retry, fatal/EOF, timeout, tick wrap, null context, worker cancellation");
|
|
return 0;
|
|
}
|
|
'''
|
|
output = root / 'artifacts/wss'
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
cfile = output / 'test_wss_io.c'
|
|
cfile.write_text(harness, encoding='utf-8')
|
|
exe = output / 'test_wss_io.exe'
|
|
if os.name == "nt":
|
|
subprocess.run(['cl', '/nologo', '/W4', '/WX', '/TC', str(cfile),
|
|
'/Fo' + str(output / 'test_wss_io.obj'), '/Fe' + str(exe)], check=True)
|
|
else:
|
|
subprocess.run(['cc', '-std=c99', '-Wall', '-Wextra', '-Werror', str(cfile), '-o', str(exe)], check=True)
|
|
subprocess.run([str(exe)], check=True)
|