/* Границы FEC-пачек, все пары потерь, ненулевой offset начала, malformed wire и восстановление после GAP. */ #include #include #include "../lib/debug_config.h" #include "../lib/mem.h" #include "../lib/u_async.h" #include "../src/stream/stream_source.h" #include "../src/stream/stream_media.h" #include "../src/stream/stream_media_buffer.h" #define CHECK(x) do { if (!(x)) { debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_ERROR); DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream media test line=%d: %s", __LINE__, #x); exit(1); } } while (0) static unsigned batches; static uint8_t expected_video[9000]; static void ready(struct stream_source *source, uint64_t seq, void *arg) { (void)source; (void)arg; CHECK(seq == batches++); } /* Канонический формат и сохранение выходного view при ошибке всех классов полей. */ static void test_media(void) { struct stream_media_block b = {.format = {1280, 720, 90, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS, 1}, .video_count = 1, .audio_count = 1}; b.video[0] = (struct stream_media_video){1, 1000000, sizeof(expected_video), 0, 6000, STREAM_MEDIA_KEY, expected_video}; b.audio[0] = (struct stream_media_audio){1000000, 3, 960, (const uint8_t *)"ops"}; uint8_t wire[7080], saved[7080]; size_t length = 0; CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_OK); memcpy(saved, wire, length); struct stream_media_block parsed; CHECK(stream_media_parse(wire, length, &parsed) == STREAM_OK && parsed.first_frame_offset == 0); CHECK(parsed.audio_count == 1 && parsed.audio[0].length == 3 && !memcmp(parsed.audio[0].data, "ops", 3)); const unsigned mutations[] = {0, 4, 5, 6, 7, 8, 12, 14, 16, 20, 24, 48, 52, 56, 60, 61, 72, 76, 78}; for (unsigned i = 0; i < sizeof(mutations) / sizeof(mutations[0]); i++) { memcpy(wire, saved, length); wire[mutations[i]] = 255; struct stream_media_block before; memset(&before, 0xa5, sizeof(before)); parsed = before; CHECK(stream_media_parse(wire, length, &parsed) < 0 && !memcmp(&parsed, &before, sizeof(before))); } for (size_t i = 0; i < length; i++) CHECK(stream_media_parse(saved, i, &parsed) < 0); b.video[0].offset = 6000; b.video[0].length = 3000; b.video[0].data = expected_video + 6000; b.audio_count = 0; CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_OK); CHECK(stream_media_parse(wire, length, &parsed) == STREAM_OK && parsed.first_frame_offset == UINT32_MAX); b.video_count = 2; b.video[1] = (struct stream_media_video){2, 1040000, sizeof(expected_video), 0, 3000, STREAM_MEDIA_KEY, expected_video}; CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_OK); CHECK(stream_media_parse(wire, length, &parsed) == STREAM_OK && parsed.first_frame_offset == 3000); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS media: nonzero frame start, malformed fields and atomic parser output"); } /* Две короткие записи остаются двумя пачками; каждый медиазаголовок переживает любую пару потерь 6+2. */ static void test_batches(void) { struct stream_descriptor d = {.group_id = 1, .source_node_id = 2, .source_epoch = 3, .n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}; uint8_t key[32] = {1}; struct stream_encoder *encoder = NULL; CHECK(stream_encoder_create(&encoder, &d, key) == STREAM_OK); struct stream_media_block b = {.format = {1280, 720, 90, STREAM_MEDIA_H264, 0, 1}, .video_count = 1}; b.video[0] = (struct stream_media_video){1, 1000000, 100, 0, 100, STREAM_MEDIA_KEY, expected_video}; uint8_t input[7080], output[7080]; size_t length = 0; CHECK(stream_media_encode(&b, input, sizeof(input), &length) == STREAM_OK); const uint8_t *blocks[RS_MAX_BLOCKS]; size_t part_size = 0; CHECK(stream_encoder_encode(encoder, 0, 1, 0, input, length, blocks, &part_size) == STREAM_OK && part_size < 1300); for (unsigned a = 0; a < 8; a++) for (unsigned c = a + 1; c < 8; c++) { struct stream_buffer *buffer = NULL; size_t decoded = 0; CHECK(stream_buffer_create(&buffer, &d, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK); for (unsigned i = 0; i < 8; i++) if (i != a && i != c) CHECK(stream_buffer_put(buffer, blocks[i], part_size) == STREAM_OK); CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &decoded) == STREAM_OK); CHECK(decoded == length && !memcmp(input, output, length)); struct stream_media_block parsed; CHECK(stream_media_parse(output, decoded, &parsed) == STREAM_OK); stream_buffer_destroy(buffer); } struct UASYNC *ua = uasync_create(); struct stream_source *source = NULL; CHECK(ua && stream_source_create(&source, ua, &d, key, ready, NULL) == STREAM_OK); CHECK(stream_source_write_block(source, input, length) == STREAM_OK && batches == 1); CHECK(stream_source_write_block(source, input, length) == STREAM_OK && batches == 2); size_t accepted = 10; CHECK(stream_source_write(source, input, length, &accepted) == STREAM_ERR_ARGUMENT && accepted == 0); CHECK(stream_source_finish(source) == STREAM_OK); CHECK(stream_source_write_block(source, input, length) == STREAM_ERR_STATE); stream_source_destroy(source); CHECK(stream_source_create(&source, ua, &d, key, ready, NULL) == STREAM_ERR_ARGUMENT); source = NULL; CHECK(stream_source_create(&source, ua, &d, key, NULL, NULL) == STREAM_OK); CHECK(stream_source_write(source, input, length, &accepted) == STREAM_OK && accepted == length); CHECK(stream_source_write_block(source, input, length) == STREAM_ERR_ARGUMENT); stream_source_destroy(source); uasync_destroy(ua, 0); stream_encoder_destroy(encoder); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS batches: atomic boundaries, all FEC losses and mode isolation"); } /* Стерео обозначается в wire явно; 960 — число кадров на канал. */ static void test_stereo(void) { struct stream_media_block b = {.format = {1280, 720, 0, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 1}, .audio_count = 1}; b.audio[0] = (struct stream_media_audio){1000000, 3, 960, (const uint8_t *)"ops"}; uint8_t wire[7080]; size_t length = 0; CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_OK); struct stream_media_block parsed; CHECK(stream_media_parse(wire, length, &parsed) == STREAM_OK && parsed.format.audio_codec == STREAM_MEDIA_OPUS_STEREO); CHECK(parsed.audio[0].samples == 960 && parsed.audio[0].length == 3 && !memcmp(parsed.audio[0].data, "ops", 3)); b.audio[0].samples = 1920; CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_ERR_FORMAT); b.audio[0].samples = 960; b.format.audio_codec = 3; CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_ERR_FORMAT); } static void test_codecs(void) { uint8_t wire[256], key[] = {0, 0, 0, 2, 11, 12, 13}; size_t length = 0; struct stream_media_block b = {.format = {64, 64, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1}; b.video[0] = (struct stream_media_video){1, 0, sizeof(key), 0, sizeof(key), STREAM_MEDIA_KEY, key}; struct stream_media_block parsed; for (unsigned codec = STREAM_MEDIA_H264; codec <= STREAM_MEDIA_H266; codec++) { b.format.video_codec = codec; CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_OK && !memcmp(wire, "uAV2", 4)); CHECK(stream_media_parse(wire, length, &parsed) == STREAM_OK && parsed.format.video_codec == codec); } wire[3] = '1'; CHECK(stream_media_parse(wire, length, &parsed) == STREAM_ERR_FORMAT); const uint8_t *config, *frame; size_t config_size, frame_size; CHECK(stream_media_key_parse(key, sizeof(key), &config, &config_size, &frame, &frame_size) == STREAM_OK); CHECK(config_size == 2 && frame_size == 1 && config[0] == 11 && frame[0] == 13); for (size_t n = 0; n < sizeof(key); n++) CHECK(stream_media_key_parse(key, n, &config, &config_size, &frame, &frame_size) < 0); key[0] = 1; CHECK(stream_media_key_parse(key, sizeof(key), &config, &config_size, &frame, &frame_size) < 0); memset(key, 0, 4); CHECK(stream_media_key_parse(key, sizeof(key), &config, &config_size, &frame, &frame_size) == STREAM_OK); } struct media_fixture { struct stream_encoder *encoder; struct stream_buffer *buffer; struct stream_media_store *store; struct stream_media_cursor *cursor; uint64_t seq; }; static void media_fixture_open(struct media_fixture *f) { struct stream_descriptor d = {.group_id = 1, .source_node_id = 2, .source_epoch = 3, .n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}; uint8_t key[32] = {1}; memset(f, 0, sizeof(*f)); CHECK(stream_encoder_create(&f->encoder, &d, key) == STREAM_OK); CHECK(stream_buffer_create(&f->buffer, &d, stream_encoder_public_key(f->encoder), 128 * 1300) == STREAM_OK); CHECK(stream_media_store_create(&f->store, f->buffer, &d) == STREAM_OK); CHECK(stream_media_cursor_create(&f->cursor, f->store, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO) == STREAM_OK); } static void media_fixture_close(struct media_fixture *f) { stream_media_cursor_destroy(f->cursor); stream_media_store_destroy(f->store); stream_buffer_destroy(f->buffer); stream_encoder_destroy(f->encoder); } /* Проверяем настоящий путь: сериализация, RS-кольцо, проверка подписи, индекс и независимый cursor. */ static int media_fixture_feed(struct media_fixture *f, const struct stream_media_block *b) { uint8_t wire[7080], decoded[7080]; size_t length = 0, part_size = 0, decoded_size = 0; const uint8_t *parts[RS_MAX_BLOCKS]; CHECK(stream_media_encode(b, wire, sizeof(wire), &length) == STREAM_OK); CHECK(stream_encoder_encode(f->encoder, f->seq, 1, 0, wire, length, parts, &part_size) == STREAM_OK); for (unsigned i = 0; i < 6; i++) CHECK(stream_buffer_put(f->buffer, parts[i], part_size) == STREAM_OK); CHECK(stream_buffer_decode(f->buffer, f->seq, decoded, sizeof(decoded), &decoded_size) == STREAM_OK); CHECK(length == decoded_size && !memcmp(wire, decoded, length)); return stream_media_store_feed(f->store, f->seq++, decoded, decoded_size); } /* Формат внутри revision и аудио внутри эпохи неизменны; регрессия live revision запрещена. */ static void test_format_conflicts(void) { for (unsigned conflict = 0; conflict < 4; conflict++) { struct media_fixture f; media_fixture_open(&f); struct stream_media_block b = {.format = {1280, 720, 0, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 2}, .video_count = 1}; b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); if (conflict == 0) b.format.revision = 1; else if (conflict == 1) b.format.width++; else { b.format.revision++; b.format.audio_codec = conflict == 2 ? STREAM_MEDIA_OPUS : 0; } b.video[0].frame_id++; b.video[0].pts_us += 40000; CHECK(media_fixture_feed(&f, &b) == STREAM_ERR_CONFLICT); media_fixture_close(&f); } DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS media store: revision regression, conflicting video and changed/removed audio rejected"); } /* Новая revision начинается с целого KEY; аудио и отстающий reader сохраняют собственные позиции. */ static void test_revision_transition(void) { struct media_fixture f; media_fixture_open(&f); struct stream_media_block b = {.format = {1280, 720, 90, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 1}, .video_count = 1, .audio_count = 1}; b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; b.audio[0] = (struct stream_media_audio){1000000, 3, 960, (const uint8_t *)"ops"}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); struct stream_media_packet packet; struct stream_media_status status; struct stream_media_cursor *slow = NULL; CHECK(stream_media_cursor_create(&slow, f.store, STREAM_MEDIA_TRACK_VIDEO) == STREAM_OK); CHECK(stream_media_cursor_status(slow, &status) == STREAM_OK && status.initialized); CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK && packet.format.revision == 1); b.audio_count = 0; b.video[0] = (struct stream_media_video){2, 1040000, 12, 0, 6, 0, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); b.format = (struct stream_media_format){1920, 1080, 270, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 2}; b.video[0] = (struct stream_media_video){3, 1080000, 12, 0, 12, 0, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_status(f.cursor, &status) == STREAM_OK && status.video_us == 1000000); CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); CHECK(stream_media_cursor_read(f.cursor, 1000000, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_AUDIO); CHECK(packet.audio.pts_us == 1000000 && packet.format.audio_codec == STREAM_MEDIA_OPUS_STEREO); b.video_count = 0; b.audio_count = 1; b.audio[0].pts_us = 1020000; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, 1020000, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_AUDIO); CHECK(packet.audio.pts_us == 1020000); b.video_count = 1; b.audio_count = 0; b.video[0] = (struct stream_media_video){4, 1120000, 12, 0, 6, STREAM_MEDIA_KEY, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); b.video[0].offset = 6; b.video[0].data = expected_video + 6; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); CHECK(packet.format.revision == 2 && packet.video.frame_id == 4 && packet.video.flags == STREAM_MEDIA_KEY); CHECK(packet.video.length == 12 && !memcmp(packet.video.data, expected_video, 12)); CHECK(stream_media_cursor_read(slow, -1, 1, -1, &packet) == STREAM_OK && packet.format.revision == 1); CHECK(stream_media_cursor_read(slow, 1000000, 1, -1, &packet) == STREAM_OK && packet.format.revision == 2); CHECK(packet.video.frame_id == 4); /* Полный KEY новой конфигурации выдаётся сразу, без таймера и искусственной задержки. */ b.format.revision = 3; b.video[0] = (struct stream_media_video){5, 1160000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, 1120000, 1, -1, &packet) == STREAM_OK && packet.format.revision == 3); CHECK(packet.video.frame_id == 5 && packet.video.flags == STREAM_MEDIA_KEY); stream_media_cursor_destroy(slow); media_fixture_close(&f); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS cursor revisions: missing/fragmented KEY, immediate complete KEY, independent audio and history"); } /* Пропущенный блок, нарушенные фрагменты и вытеснение запрещают P-кадры до нового целого KEY. */ static void test_cursor_key_recovery(void) { for (unsigned failure = 0; failure < 3; failure++) { struct media_fixture f; media_fixture_open(&f); struct stream_media_block b = {.format = {1280, 720, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1}; b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); struct stream_media_packet packet; CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK); b.video[0] = (struct stream_media_video){2, 1040000, 12, 0, 6, 0, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); if (failure == 0) { stream_media_store_gap(f.store, f.seq, f.seq); f.seq++; } else if (failure == 1) { b.video[0].offset = 5; b.video[0].data = expected_video + 5; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); } else { for (unsigned i = 0; i < 48; i++) { b.video[0] = (struct stream_media_video){3 + i, 1080000 + i * 40000, 12, 0, 12, 0, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); } } int result = stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet); CHECK(result == (failure == 2 ? STREAM_EXPIRED : STREAM_GAP)); stream_media_cursor_reset(f.cursor); b.video[0] = (struct stream_media_video){100, 4000000, 12, 0, 12, 0, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_WAIT); b.video[0] = (struct stream_media_video){101, 4040000, 12, 0, 6, STREAM_MEDIA_KEY, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_WAIT); b.video[0].offset = 6; b.video[0].data = expected_video + 6; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); CHECK(packet.video.frame_id == 101 && packet.video.flags == STREAM_MEDIA_KEY && packet.video.length == 12); CHECK(!memcmp(packet.video.data, expected_video, 12)); media_fixture_close(&f); } DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS cursor recovery: GAP, fragment discontinuity and expiry resume only from a complete KEY"); } int main(int argc, char **argv) { debug_config_init(); debug_set_level(DEBUG_LEVEL_INFO); /* Усечения проверяются массово; ожидаемые отказы не загромождают диагностический лог. */ debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DISABLED); size_t before = u_get_allocated_count(); for (unsigned i = 0; i < sizeof(expected_video); i++) expected_video[i] = (uint8_t)(i * 31 + 7); if (argc == 2 && !strcmp(argv[1], "--revision-transition")) { debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); test_revision_transition(); test_cursor_key_recovery(); } else { CHECK(argc == 1); test_media(); test_batches(); test_stereo(); test_codecs(); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); test_revision_transition(); test_cursor_key_recovery(); test_format_conflicts(); } CHECK(u_get_allocated_count() == before); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS stream_media: protocol, FEC, frame recovery, boundaries, no leaks"); return 0; }