118 lines
5.0 KiB
Python
118 lines
5.0 KiB
Python
"""Run the real diagnostic code with mocked modem APIs (no network traffic)."""
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
base = root / 'ql-application/threadx/interface/custom_app'
|
|
sdk = root / 'ql-application/threadx/common/include'
|
|
harness = r'''
|
|
#include <assert.h>
|
|
#include <setjmp.h>
|
|
#include "net_diag.c"
|
|
static int creates, fail_create, mode, hold, pdp = 1;
|
|
static unsigned long clock_ms;
|
|
static int sample_steps, step_limit = 1;
|
|
static jmp_buf loop_exit;
|
|
static ql_ping_ctx_t mock_ctx;
|
|
static ql_ping_cb_t callback;
|
|
uint64_t hal_os_up_time_ms(void) { return clock_ms; }
|
|
uint8_t get_sleep_mode(void) { return hold; }
|
|
int get_mqtt_connect_status(void) { return 1; }
|
|
unsigned char get_xiaoice_connected_status(void) { return 1; }
|
|
int fota_start;
|
|
u8 power_down_status;
|
|
QlOSStatus ql_rtos_task_create(ql_task_t *task, u32 stack, u8 priority, char *name,
|
|
void (*entry)(void *), void *arg, ...) {
|
|
(void)name; (void)entry; (void)arg;
|
|
assert(stack == 4096 && priority == 120);
|
|
++creates;
|
|
if (fail_create == creates) return -9;
|
|
*task = (void *)1; return 0;
|
|
}
|
|
void ql_rtos_task_sleep_ms(u32 ms) {
|
|
clock_ms += ms;
|
|
if (step_limit > 1) {
|
|
++sample_steps;
|
|
if (sample_steps == 1) pdp = 0;
|
|
if (sample_steps == 2) pdp = 1;
|
|
if (sample_steps < step_limit) return;
|
|
}
|
|
longjmp(loop_exit, 1);
|
|
}
|
|
QL_NW_ERROR_CODE ql_nw_get_signal_strength(QL_NW_SIGNAL_STRENGTH_INFO_T *s) {
|
|
s->LTE_SignalStrength.rsrp = -97; return QL_NW_SUCCESS;
|
|
}
|
|
QL_NW_ERROR_CODE ql_nw_get_reg_status(QL_NW_REG_STATUS_INFO_T *r) {
|
|
r->data_reg.state = 1; return QL_NW_SUCCESS;
|
|
}
|
|
int ql_get_data_call_info(int profile, int version, struct ql_data_call_info *d) {
|
|
assert(profile == 1 && version == 0); d->v4.state = pdp; return 0;
|
|
}
|
|
QL_NW_ERROR_CODE ql_nw_get_sinr(int *s) { *s = 13; return QL_NW_SUCCESS; }
|
|
QL_NW_ERROR_CODE ql_nw_get_cell_info(QL_NW_CELL_INFO_T *c) {
|
|
c->lte_info_valid = 1; c->lte_info_num = 1;
|
|
c->lte_info[0].cid = 123; c->lte_info[0].pci = 42; return QL_NW_SUCCESS;
|
|
}
|
|
ql_ping_ctx_t *ql_ping_init(ql_ping_cfg_t cfg, ql_ping_cb_t cb) {
|
|
assert(cfg.timeout == 2000 && cfg.pingNum == 1 && cfg.dataSize == 32 && cfg.pdpCid == 1);
|
|
callback = cb; return mode == 3 ? NULL : &mock_ctx;
|
|
}
|
|
int ql_ping(ql_ping_ctx_t *ctx) {
|
|
ql_ping_stat_t s;
|
|
assert(ctx == &mock_ctx); memset(&s, 0, sizeof(s));
|
|
if (mode == 2) return -2;
|
|
s.sentNum = 1; s.rcvdNum = mode == 0; s.rtt = 123;
|
|
callback(QL_PING_FINISHED, mode == 1 ? -1 : 0, &s); return 0;
|
|
}
|
|
void ql_ping_deinit(ql_ping_ctx_t *ctx) { assert(ctx == &mock_ctx); }
|
|
int main(void) {
|
|
fail_create = 2;
|
|
net_diag_start(); assert(!enabled && sample_task && !probe_task);
|
|
fail_create = 0;
|
|
net_diag_start(); assert(enabled && creates == 3);
|
|
net_diag_start(); assert(creates == 3);
|
|
mode = 0; probe_once(0); assert(attempts[0] == 1 && received[0] == 1 && !streak[0]);
|
|
mode = 1; probe_once(0); assert(attempts[0] == 2 && received[0] == 1 && streak[0] == 1);
|
|
mode = 2; probe_once(0); assert(attempts[0] == 2 && errors[0] == 1);
|
|
mode = 3; probe_once(0); assert(attempts[0] == 2 && errors[0] == 2);
|
|
mode = 0; probe_once(0); assert(!streak[0]);
|
|
step_limit = 3;
|
|
if (!setjmp(loop_exit)) sample_worker(NULL);
|
|
step_limit = 1;
|
|
hold = 1;
|
|
if (!setjmp(loop_exit)) probe_worker(NULL);
|
|
assert(attempts[1] == 0);
|
|
puts("PASS: creation failure/retry, idempotent activation, probe counters, sampling, sleep pause");
|
|
return 0;
|
|
}
|
|
'''
|
|
with tempfile.TemporaryDirectory(prefix='banban-netdiag-') as directory:
|
|
tmp = Path(directory)
|
|
for name in ['ql_type.h', 'ql_rtos.h', 'ql_nw.h', 'ql_data_call.h', 'ql_ping.h']:
|
|
content = (sdk / name).read_bytes()
|
|
if name == 'ql_type.h':
|
|
# The ARM SDK defines 32-bit size_t unconditionally. Host tests use libc's.
|
|
content, count = re.subn(rb'typedef unsigned int\s+size_t;', b'/* host size_t */', content)
|
|
assert count == 1
|
|
(tmp / name).write_bytes(content)
|
|
(tmp / 'test.c').write_text(harness)
|
|
command = [shutil.which('clang') or 'cc', '-std=c99', '-Werror=implicit-function-declaration',
|
|
'-Wno-typedef-redefinition']
|
|
for include in [tmp, base, base / 'inc', base / 'lib/hal', sdk, sdk / 'lwipv4v6']:
|
|
command.extend(['-I', str(include)])
|
|
subprocess.run(command + [str(tmp / 'test.c'), '-o', str(tmp / 'test')], check=True)
|
|
result = subprocess.run([str(tmp / 'test')], check=True, capture_output=True, text=True)
|
|
print(result.stdout, end='')
|
|
assert '本次网络不可用观测时长=2000 ms' in result.stdout
|
|
assert '未回应率=50%' in result.stdout
|
|
|
|
key = (base / 'hal_key.c').read_text()
|
|
alarm = key[key.index('if(key_press_alarm_counter!=0'):key.index('ql_rtos_task_sleep_ms(10);')]
|
|
assert 'net_diag_start();' in alarm
|
|
for forbidden in ['build_alarm(', 'build_leave_to_parent(', 'stop_player_func(', 'app_play_tips_file(']:
|
|
assert forbidden not in alarm, forbidden
|
|
print('PASS: diagnostic key has no former alarm/message/audio actions')
|