"""Host behavioral checks for the actual diagnostic modules; no device required.""" from pathlib import Path import re import shutil import subprocess import tempfile ROOT = Path(__file__).resolve().parents[1] APP = ROOT / 'ql-application/threadx/interface/custom_app' KERNEL = ROOT / 'ql-kernel/threadx/api_map/talkingq_time_diag.c' def api_hash(name): value = 0 for byte in name.encode(): term = (value * 5527 + byte * 7) & 0xffffffff value = (term ^ ((term & 65535) ** 2)) & 0xffffffff return value assert api_hash('ql_user_api_map_test') == 0x60BCEA5A abi = (ROOT / 'ql-config/config/talkingq_time_diag.h').read_text() hash_value = int(re.search(r'API_HASH 0x([0-9A-F]+)', abi)[1], 16) assert api_hash('talkingq_time_diag_snapshot_v1') == hash_value table = (ROOT / 'ql-kernel/threadx/api_map/ql_user_api_map.c').read_text() assert f'{hash_value:08x}' not in table.lower() with tempfile.TemporaryDirectory(prefix='time-diag-') as directory: work = Path(directory) headers = { 'ql_func.h': '#include \nextern intptr_t (*m_get_api_ptr)(const char *);\n', 'ql_rtos.h': 'unsigned int ql_rtos_get_systicks(void);\n', 'ql_dev.h': 'int ql_dev_get_firmware_version(char *, unsigned int);\n', 'build_macro.h': '#define CUSTOM_APP_BUILD_VER "test"\n#define CUSTOM_APP_BUILD_TIME "test"\n#define CUSTOM_APP_BUILD_REV "test"\n', 'ql_rtc.h': 'typedef struct { int tm_sec,tm_min,tm_hour,tm_mday,tm_mon,tm_year,tm_wday; } ql_rtc_time_t;\nint ql_rtc_get_time(ql_rtc_time_t *);\n', 'ql_type.h': '', 'quec_cust_feature.h': '', 'pmic.h': 'typedef int PMIC_RTC_OFFSET_STORAGE;\nunsigned int PM812_RTC_GET_32B_SYS_OFFSET(void), PM812_RTC_GET_32B_APP_OFFSET(void), PM812_RTC_GET_32B_COUNTER(void);\n', 'pmic_rtc.h': 'int pmic_rtc_INIT_FLAG_GET(void), pmic_userdata_get_app_rtc_ever_configed(void), PMIC_RTC_GetTimezone(void);\n', 'quec_rtc.h': 'int GetTimeZone(void), quec_get_rtc_source(void);\n', } for name, content in headers.items(): (work / name).write_text(content) (work / 'harness.c').write_text(r''' #include #include #include #include #include #include "talkingq_time_diag.h" #include "time_diag.h" #include "ql_rtc.h" unsigned char m_rtc_app_sys_keeping_synced; int m_pmic_rtc_offset_storage_type; char ntp_sync_cnt = 3, ntp_dns_gain = 1; static int initialized, raw_reads, api_mode, sample_index, write_pointer; static time_t samples[2]; unsigned int ql_rtos_get_systicks(void) { return 123; } int pmic_rtc_INIT_FLAG_GET(void) { return initialized; } int pmic_userdata_get_app_rtc_ever_configed(void) { return 1; } unsigned int PM812_RTC_GET_32B_COUNTER(void) { ++raw_reads; return 0xfffffff0U; } unsigned int PM812_RTC_GET_32B_SYS_OFFSET(void) { ++raw_reads; return 100; } unsigned int PM812_RTC_GET_32B_APP_OFFSET(void) { ++raw_reads; return 50; } int PMIC_RTC_GetTimezone(void) { return 8; } int GetTimeZone(void) { return 480; } int quec_get_rtc_source(void) { return 1; } int get_ntp_time_type(void) { return 0; } int GetNTPStatus(void) { return 1; } int talkingq_ntp_diag_ready(void) { return 1; } int quec_get_rtc_status(void) { return 1; } int quec_get_nw_status(void) { return 2; } char *get_ntp_server_ip(void) { return "192.0.2.1"; } int ql_dev_get_firmware_version(char *out, unsigned int size) { snprintf(out, size, "test-firmware"); return 0; } int ql_rtc_get_time(ql_rtc_time_t *out) { (void)out; return -1; } time_t time(time_t *out) { time_t value; assert(sample_index < 2); value = samples[sample_index++]; if (out && write_pointer) *out = value; return value; } static int bad_snapshot(talkingq_time_diag_snapshot *out, unsigned int size) { (void)size; out->abi_version = 99; return 0; } static intptr_t lookup(const char *name) { assert(strcmp(name, TALKINGQ_TIME_DIAG_API_NAME) == 0); if (api_mode == 0) return 0; if (api_mode == 1) return (intptr_t)bad_snapshot; return (intptr_t)talkingq_time_diag_snapshot_v1; } intptr_t (*m_get_api_ptr)(const char *) = lookup; int main(void) { talkingq_time_diag_snapshot s; assert(sizeof(s) == 108); assert(talkingq_time_diag_snapshot_v1(NULL, sizeof(s)) == -1); assert(talkingq_time_diag_snapshot_v1(&s, sizeof(s)-1) == -1); assert(talkingq_time_diag_snapshot_v1(&s, sizeof(s)) == 0); assert(!s.rtc_initialized && raw_reads == 0); initialized = 1; assert(talkingq_time_diag_snapshot_v1(&s, sizeof(s)) == 0); assert(raw_reads == 3 && s.sys_offset == 100 && s.app_offset == 50); assert(s.counter + s.sys_offset == 84 && s.counter + s.app_offset == 34); assert(s.app_follows_sys == 0 && m_rtc_app_sys_keeping_synced == 0); assert(s.network_tz_minutes == 480 && s.rtc_tz_hours == 8); #ifdef NTP assert(s.ntp_ready == 1 && s.ntp_status == 1); #else assert(s.ntp_ready == -1 && s.ntp_status == -1); #endif #ifdef QUEC_GNSS_NTP assert(s.gnss_cid == 2 && strcmp(s.ntp_server, "192.0.2.1") == 0); #else assert(s.gnss_cid == -1 && s.ntp_server[0] == 0); #endif talkingq_time_diag_boot(); /* old kernel */ api_mode = 1; talkingq_time_diag_log("bad-abi"); api_mode = 2; talkingq_time_diag_log("divergent"); samples[0] = 100; samples[1] = 1800000000; assert(talkingq_time_diag_wss_time() == 100); samples[0] = 1800000000; samples[1] = 100; sample_index = 0; write_pointer = 1; assert(talkingq_time_diag_wss_time() == 1800000000); assert(m_rtc_app_sys_keeping_synced == 0); talkingq_time_diag_trace("kernel-check"); puts("PASS"); return 0; } ''') compiler = shutil.which('clang') or shutil.which('cc') if not compiler: raise SystemExit('A native C compiler is required') for features in ([], ['-DNTP'], ['-DNTP', '-DQUEC_GNSS_NTP']): command = [compiler, '-std=c99', '-Wall', '-Wextra', '-Werror', '-I', str(work), '-I', str(APP / 'inc'), '-I', str(ROOT / 'ql-config/config'), *features, str(APP / 'time_diag.c'), str(KERNEL), str(work / 'harness.c'), '-o', str(work / 'test')] subprocess.run(command, check=True) log = subprocess.check_output([str(work / 'test')], text=True) for expected in ('kernel_diag=unavailable', 'kernel_diag=abi_mismatch', 'sys_raw=84 app_raw=34', 'time_ptr_value=1', 'time_ptr_value=100', 'valid=0', 'valid=1', 'PASS'): assert expected in log, expected print('PASS:', ' '.join(features) or 'optional features off')