/* lib9p_util/tests/test_nut.c - Tests for NUT file generation * Keep this file in-sync with test_nut.py. * * Copyright (C) 2025 Luke T. Shumaker * SPDX-License-Identifier: AGPL-3.0-or-later */ #include #include #include #include #define error __error #include #undef error #include #include #include #define LOG_NAME TEST #include #include #include "nut.h" uint32_t nut_crc32(void *dat, size_t len); /******************************************************************************/ #define MIN(a, b) ( ((a) < (b)) ? (a) : (b) ) #define test_assert(expr) do { \ if (!(expr)) { \ log_errorln("test failure: ", __FILE__, ":", __LINE__, ":", __func__, ": ", #expr); \ exit(1); \ } \ } while (0) /* bytearray **********************************************/ typedef struct { void *dat; size_t len; size_t cap; } bytearray; void bytearray_extend(bytearray *buf, const void *dat, size_t len) { if (buf->cap < buf->len + len) { buf->cap = LM_ROUND_UP(buf->len + len, 1024); buf->dat = realloc(buf->dat, buf->cap); assert(buf->dat); } memcpy(&((char *)buf->dat)[buf->len], dat, len); buf->len += len; } LO_IMPLEMENTATION_STATIC(io_writer, bytearray, bytearray); static size_t_and_error bytearray_writev(bytearray *self, const struct wr_iovec *iov, int iovcnt) { assert(self); assert(iovcnt == 1); assert(iov); bytearray_extend(self, iov[0].iov_write_from, iov[0].iov_len); return ERROR_AND(size_t, iov[0].iov_len, ERROR_NULL); } /* generate_py ********************************************/ char *argv0; bytearray generate_py(void) { int p[2]; if (pipe(p)) __error(1, errno, "pipe"); switch (fork()) { case -1: __error(1, errno, "fork"); break; case 0: if (dup2(p[1], 1) < 0) __error(1, errno, "dup2"); if (close(p[0])) __error(1, errno, "close"); char *py_suffix = "../../../lib9p_util/nut_gen/test_nut.py"; char *slash = strrchr(argv0, '/'); char *py_path0; if (slash) { py_path0 = malloc(slash-argv0 + strlen(py_suffix) + 2); memcpy(py_path0, argv0, slash-argv0); py_path0[slash-argv0] = '/'; strcpy(&py_path0[slash-argv0+1], py_suffix); } else { py_path0 = py_suffix; } char *py_path = realpath(py_path0, NULL); if (!py_path) __error(1, errno, "realpath"); execl(py_path, py_path, NULL); __error(1, errno, "exec"); } if (close(p[1])) __error(1, errno, "close"); bytearray ret = {}; for (;;) { char *buf[4096]; ssize_t r = read(p[0], buf, sizeof(buf)); if (r < 0) __error(1, errno, "read"); if (r == 0) break; bytearray_extend(&ret, buf, r); } if (close(p[0])) __error(1, errno, "close"); return ret; } /* testclock **********************************************/ struct testclock { uint64_t now_ns; }; LO_IMPLEMENTATION_STATIC(alarmclock, struct testclock, testclock); static uint64_t testclock_get_time_ns(struct testclock *self) { return self->now_ns; } static void testclock_add_trigger(struct testclock *, struct alarmclock_trigger *, uint64_t, void (*)(void *), void *) { assert_notreached("should not call add_trigger()"); } static void testclock_del_trigger(struct testclock *, struct alarmclock_trigger *) { assert_notreached("should not call del_trigger()"); } /* state **************************************************/ struct state { struct nut_file file; lo_interface lib9p_srv_fio fio; struct testclock clock; }; void state_init(struct state *self, void *framebuffer) { memset(self, 0, sizeof(*self)); self->file.framebuffer = framebuffer; lib9p_srv_fio_or_error r = nut_file_fopen(&self->file, &(struct lib9p_srv_ctx){}, true, false, false); test_assert(!r.is_err); self->fio = r.lib9p_srv_fio; bootclock = LO_BOX(alarmclock, &self->clock); } void _state_copy(struct state *src, bytearray *dst, size_t len) { size_t goal = dst->len + len; while (dst->len < goal) { error err = LO_CALL(src->fio, pread, &(struct lib9p_srv_ctx){}, LO_BOX(io_writer, dst), dst->len, MIN(4096, goal - dst->len)); test_assert(ERROR_IS_NULL(err)); } } void state_append_intro(struct state *self, bytearray *buf) { _state_copy(self, buf, NUT_FILE_INTRO_LEN); } void state_append_frame(struct state *self, bytearray *buf, uint64_t time_ns) { self->clock.now_ns = time_ns; _state_copy(self, buf, NUT_FRAME_INTRO_LEN + (NUT_FB_W*NUT_FB_H)); } /******************************************************************************/ void test_crc32(void) { test_assert(nut_crc32("hello", 5) == 0xA1DE215E); } #define font_w 8 #define font_h 4 [[gnu::nonstring]] const char font[10][font_w*font_h] = { " ## " " # # " " # # " " ## ", " # " " ## " " # " " ### ", " ## " " # # " " # " " #### ", " ## " " # " " # " " ## ", " ## " " # # " " #### " " # ", " #### " " ### " " # " " ### ", " # " " ### " " # # " " ## ", " #### " " # " " # " " # ", " ## " " # # " " #### " " ## ", " ## " " # # " " ### " " # ", }; static_assert(LM_ARRAY_LEN(font) == 10); static_assert(sizeof(font[0]) == 32); bytearray generate_c(void) { bytearray buf = {}; uint8_t framebuffer[NUT_FB_W*NUT_FB_H] = {}; struct state state; state_init(&state, framebuffer); state_append_intro(&state, &buf); #define SCALE 10 for (size_t i = 0; i < 10; i++) { memset(framebuffer, 0b00011000, sizeof(framebuffer)); for (size_t y = 0; y < font_h*SCALE; y++) { for (size_t x = 0; x < font_w*SCALE; x++) { char chr = font[i][(y/SCALE)*font_w + (x/SCALE)]; uint8_t color = (chr == ' ') ? 0b11000000 : 0b00000011; framebuffer[(y*NUT_FB_W) + x] = color; } } state_append_frame(&state, &buf, (uint64_t)i * UINT64_C(1000000000)); } LO_CALL(state.fio, iofree); return buf; } void test_full_file(void) { bytearray content_c = generate_c(); bytearray content_py = generate_py(); test_assert(content_c.len == content_py.len); test_assert(memcmp(content_c.dat, content_py.dat, content_py.len) == 0); free(content_c.dat); free(content_py.dat); } int main(int argc, char *argv[]) { if (argc >= 2) { bytearray content = generate_c(); for (size_t done = 0; done < content.len;) { ssize_t r = write(1, &((char *)content.dat)[done], content.len-done); if (r < 0) __error(1, errno, "write"); done += r; } free(content.dat); return 0; } argv0 = argv[0]; log_infoln("test_crc32"); test_crc32(); log_infoln("test_full_file"); test_full_file(); log_infoln("passed!"); return 0; }