217 lines
8.9 KiB
Python
217 lines
8.9 KiB
Python
"""Compile the actual APP NTP tester with scripted network/SDK responses.
|
|
|
|
These are host regression tests, not evidence of device whitelist access.
|
|
"""
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP = ROOT / 'ql-application/threadx/interface/custom_app'
|
|
|
|
with tempfile.TemporaryDirectory(prefix='app-ntp-test-') as directory:
|
|
work = Path(directory)
|
|
headers = {
|
|
'ql_rtos.h': '''#include <stddef.h>
|
|
typedef void *ql_task_t;
|
|
typedef void *ql_sem_t;
|
|
unsigned int ql_rtos_get_systicks(void);
|
|
int ql_rtos_task_create(ql_task_t *, unsigned int, unsigned char, char *, void (*)(void *), void *, ...);
|
|
int ql_rtos_task_delete(ql_task_t);
|
|
int ql_rtos_semaphore_create(ql_sem_t *, unsigned int);
|
|
int ql_rtos_semaphore_wait(ql_sem_t, unsigned int);
|
|
int ql_rtos_semaphore_release(ql_sem_t);
|
|
''',
|
|
'ql_func.h': '#include <stdint.h>\nextern intptr_t (*m_get_api_ptr)(char *);\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;
|
|
int ql_rtc_get_time(ql_rtc_time_t *);
|
|
''',
|
|
'ql_ntp.h': '''enum {QL_NTP_SUCCESS=0};
|
|
int ql_ntp_set_server(char *);
|
|
int ql_ntp_set_cid(int);
|
|
void ql_ntp_sync_ex(void (*)(int));
|
|
''',
|
|
'netdb.h': '''#include <sys/socket.h>
|
|
struct addrinfo {
|
|
int ai_flags,ai_family,ai_socktype,ai_protocol;
|
|
socklen_t ai_addrlen;
|
|
struct sockaddr *ai_addr;
|
|
struct addrinfo *ai_next;
|
|
};
|
|
int getaddrinfo_with_pcid(const char *,const char *,const struct addrinfo *,struct addrinfo **,unsigned char);
|
|
#define freeaddrinfo fake_freeaddrinfo
|
|
void fake_freeaddrinfo(struct addrinfo *);
|
|
''',
|
|
'sockets.h': '''#include <sys/socket.h>
|
|
#include <sys/select.h>
|
|
#include <arpa/inet.h>
|
|
#define socket fake_socket
|
|
#define connect fake_connect
|
|
#define send fake_send
|
|
#define recv fake_recv
|
|
#define close fake_close
|
|
#define select fake_select
|
|
int fake_socket(int,int,int);
|
|
int fake_connect(int,const struct sockaddr *,socklen_t);
|
|
int fake_send(int,const void *,size_t,int);
|
|
int fake_recv(int,void *,size_t,int);
|
|
int fake_close(int);
|
|
int fake_select(int,fd_set *,fd_set *,fd_set *,struct timeval *);
|
|
''',
|
|
}
|
|
for name, content in headers.items():
|
|
(work / name).write_text(content)
|
|
harness = r'''
|
|
#include <assert.h>
|
|
#include <stdlib.h>
|
|
#include "APP_SOURCE"
|
|
static int mode, creates, sends, closes, frees, syncs, signals;
|
|
static unsigned int ticks=100;
|
|
static unsigned char sent[48];
|
|
static time_t app_epoch=100;
|
|
static void (*saved_callback)(int);
|
|
static struct sockaddr_in address;
|
|
static struct addrinfo result;
|
|
unsigned int ql_rtos_get_systicks(void) { return ticks; }
|
|
int getaddrinfo_with_pcid(const char *name,const char *port,
|
|
const struct addrinfo *hints,struct addrinfo **out,unsigned char cid) {
|
|
assert(strcmp(name,"ntp.aliyun.com")==0 && strcmp(port,"123")==0 && cid==1);
|
|
assert(hints->ai_family==AF_INET && hints->ai_socktype==SOCK_DGRAM);
|
|
if (mode==1) return -9;
|
|
address.sin_family=AF_INET; address.sin_port=htons(123);
|
|
address.sin_addr.s_addr=htonl(0xc0000201U);
|
|
result.ai_addr=(struct sockaddr *)&address; result.ai_addrlen=sizeof(address);
|
|
*out=&result; return 0;
|
|
}
|
|
void fake_freeaddrinfo(struct addrinfo *p) {assert(p==&result); ++frees;}
|
|
int fake_socket(int domain,int type,int protocol) {
|
|
assert(domain==AF_INET && type==SOCK_DGRAM && protocol==IPPROTO_UDP);
|
|
return mode==10 ? -1 : 7;
|
|
}
|
|
int fake_connect(int fd,const struct sockaddr *p,socklen_t size) {
|
|
assert(fd==7 && p==(struct sockaddr *)&address && size==sizeof(address));
|
|
return mode==11 ? -1 : 0;
|
|
}
|
|
int fake_send(int fd,const void *data,size_t size,int flags) {
|
|
assert(fd==7 && size==48 && flags==0); ++sends;
|
|
memcpy(sent,data,48); assert(sent[0]==0x23);
|
|
return mode==12 ? -1 : 48;
|
|
}
|
|
int fake_select(int nfds,fd_set *readers,fd_set *writers,fd_set *errors,struct timeval *timeout) {
|
|
assert(nfds==8 && FD_ISSET(7,readers) && writers==NULL && errors==NULL);
|
|
assert(timeout->tv_sec<=5); ticks+=20;
|
|
return mode==2 ? 0 : 1;
|
|
}
|
|
static void valid_reply(unsigned char *p,const unsigned char *request) {
|
|
memset(p,0,48); p[0]=0x24; p[1]=2;
|
|
memcpy(p+24,request+40,8);
|
|
write_u32(p+32,4008988800U); write_u32(p+40,4008988800U);
|
|
}
|
|
int fake_recv(int fd,void *buffer,size_t size,int flags) {
|
|
assert(fd==7 && size>=48 && flags==0);
|
|
valid_reply(buffer,sent); return 48;
|
|
}
|
|
int fake_close(int fd) {assert(fd==7); ++closes; return 0;}
|
|
int ql_rtos_task_create(ql_task_t *task,unsigned int size,unsigned char priority,char *name,
|
|
void (*entry)(void *),void *arg,...) {
|
|
assert(size>=4096 && priority==110 && strcmp(name,"app_ntp_test")==0);
|
|
++creates; *task=(void *)1; entry(arg); return 0;
|
|
}
|
|
int ql_rtos_task_delete(ql_task_t task) {assert(task==NULL); return 0;}
|
|
int ql_rtos_semaphore_create(ql_sem_t *sem,unsigned int count) {
|
|
assert(count==0);
|
|
if (mode==13) return -1;
|
|
*sem=(void *)2; return 0;
|
|
}
|
|
int ql_rtos_semaphore_wait(ql_sem_t sem,unsigned int timeout) {
|
|
assert(sem==(void *)2 && timeout==25000);
|
|
ticks+=100;
|
|
if (mode==5) return -1;
|
|
assert(signals==1); return 0;
|
|
}
|
|
int ql_rtos_semaphore_release(ql_sem_t sem) {assert(sem==(void *)2); ++signals; return 0;}
|
|
static intptr_t lookup(char *name) {
|
|
assert(strncmp(name,"ql_ntp_",7)==0);
|
|
return mode==6 ? 0 : 1;
|
|
}
|
|
intptr_t (*m_get_api_ptr)(char *)=lookup;
|
|
int ql_ntp_set_server(char *name) {assert(strcmp(name,"ntp.aliyun.com")==0); return mode==14 ? 1 : 0;}
|
|
int ql_ntp_set_cid(int cid) {assert(cid==1); return mode==15 ? 1 : 0;}
|
|
void ql_ntp_sync_ex(void (*callback)(int)) {
|
|
++syncs; saved_callback=callback;
|
|
if (mode==5) return;
|
|
if (mode!=3) app_epoch=1800000000;
|
|
if (mode==7) app_epoch+=28800;
|
|
callback(mode==4 ? 1 : 0);
|
|
}
|
|
time_t time(time_t *out) {(void)out; return app_epoch;}
|
|
int ql_rtc_get_time(ql_rtc_time_t *calendar) {
|
|
calendar->tm_year=2027; return mode==8 ? -1 : 0;
|
|
}
|
|
static void check_parser(void) {
|
|
unsigned char request[48]={0},reply[48]; unsigned int epoch=0;
|
|
request[40]=9; valid_reply(reply,request);
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==0 && epoch==1800000000);
|
|
assert(parse_ntp_reply(reply,47,request,&epoch)==-1);
|
|
reply[0]|=0xc0; assert(parse_ntp_reply(reply,48,request,&epoch)==-2);
|
|
valid_reply(reply,request); reply[1]=0;
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==-2);
|
|
reply[1]=16; assert(parse_ntp_reply(reply,48,request,&epoch)==-2);
|
|
valid_reply(reply,request); reply[0]=0x23;
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==-2);
|
|
valid_reply(reply,request); reply[24]^=1;
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==-3);
|
|
valid_reply(reply,request); memset(reply+32,0,8);
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==-4);
|
|
valid_reply(reply,request); memset(reply+40,0,8);
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==-4);
|
|
valid_reply(reply,request); write_u32(reply+40,2208988800U+100);
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==-5);
|
|
valid_reply(reply,request); write_u32(reply+40,(unsigned int)(NTP_EPOCH_DELTA+2100000000ULL));
|
|
assert(parse_ntp_reply(reply,48,request,&epoch)==0 && epoch==2100000000U);
|
|
}
|
|
int main(int argc,char **argv) {
|
|
assert(argc==2); mode=atoi(argv[1]); check_parser();
|
|
app_ntp_test_start(); app_ntp_test_start(); assert(creates==1);
|
|
assert(frees==(mode==1 ? 0 : 1));
|
|
assert(closes==((mode==1 || mode==10) ? 0 : 1));
|
|
if (mode==1 || mode==2 || mode==6 || mode>=10) assert(syncs==0);
|
|
else assert(syncs==1);
|
|
if (mode==5) {saved_callback(0); assert(signals==1);}
|
|
return 0;
|
|
}
|
|
'''.replace('APP_SOURCE', str(APP / 'app_ntp_test.c'))
|
|
(work / 'test.c').write_text(harness)
|
|
compiler = shutil.which('clang') or shutil.which('cc')
|
|
if not compiler:
|
|
raise SystemExit('A native C compiler is required')
|
|
subprocess.run([compiler, '-std=c99', '-Wall', '-Wextra', '-Werror',
|
|
'-I', str(work), '-I', str(APP / 'inc'),
|
|
str(work / 'test.c'), '-o', str(work / 'test')], check=True)
|
|
cases = {
|
|
0: 'PASS udp_response=verified app_utc=verified',
|
|
1: 'FAIL stage=dns',
|
|
2: 'FAIL stage=receive_wait',
|
|
3: 'FAIL stage=app_clock_verification',
|
|
4: 'FAIL stage=app_clock_verification',
|
|
5: 'FAIL stage=sdk_callback_timeout',
|
|
6: 'FAIL stage=sdk_api_unavailable',
|
|
7: 'FAIL stage=app_clock_verification',
|
|
8: 'FAIL stage=app_clock_verification',
|
|
10: 'FAIL stage=socket',
|
|
11: 'FAIL stage=udp_connect',
|
|
12: 'FAIL stage=send',
|
|
13: 'FAIL stage=semaphore',
|
|
14: 'FAIL stage=sdk_server',
|
|
15: 'FAIL stage=sdk_cid',
|
|
}
|
|
for mode, expected in cases.items():
|
|
log = subprocess.check_output([str(work / 'test'), str(mode)], text=True)
|
|
assert expected in log, log
|
|
if mode != 0:
|
|
assert '[APP-NTP-TEST] PASS ' not in log, log
|
|
print('PASS: packet validation and 15 network/SDK/clock scenarios; no device traffic sent')
|