/* Настоящее ядро/actor, подписанный медиаблок и H.264 decoder: каталог до подписки, эпоха, показ и teardown. */ #include "../src/streamplayer_window.h" #include "../src/sound_manager.h" #include "../transport/stream_bridge.h" #include #include #include #include #include #include #include extern "C" { #include "utun_instance.h" #include "chat/chat_core.h" #include "chat/chat_event.h" #include "stream/stream.h" #include "stream/stream_member.h" #include "debug_config.h" #include "mem.h" #include "sqlite3.h" #include "opus_codec.h" #ifdef HAVE_FFMPEG #include #endif } static QList snapshots; #ifdef TEST_NULL_AUDIO static std::atomic originalAudioCallback{nullptr}; static std::atomic nonzeroSamples{0}; extern "C" ma_result __real_ma_context_init(const ma_backend*, ma_uint32, const ma_context_config*, ma_context*); extern "C" ma_result __real_ma_device_init(ma_context*, const ma_device_config*, ma_device*); extern "C" ma_result __wrap_ma_context_init(const ma_backend*, ma_uint32, const ma_context_config*, ma_context* context) { ma_backend backend = ma_backend_null; return __real_ma_context_init(&backend, 1, nullptr, context); } static void observeAudio(ma_device* device, void* out, const void* in, ma_uint32 frames) { originalAudioCallback.load()(device, out, in, frames); const auto* pcm = static_cast(out); for (unsigned i = 0; i < frames; i++) if (pcm[i]) nonzeroSamples.fetch_add(1); } extern "C" ma_result __wrap_ma_device_init(ma_context* context, const ma_device_config* config, ma_device* device) { ma_device_config copy = *config; if (copy.deviceType == ma_device_type_playback && copy.playback.format == ma_format_s16 && copy.playback.channels == 1) { originalAudioCallback = copy.dataCallback; copy.dataCallback = observeAudio; } return __real_ma_device_init(context, ©, device); } #endif static void sourceEvent(UTUN_INSTANCE*, int type, const uint8_t* data, int size) { if (type == CHAT_EVT_STREAM_SOURCES) snapshots.push_back(QByteArray(reinterpret_cast(data), size)); } class StreamViewTest : public QObject { Q_OBJECT UASYNC* ua = nullptr; UTUN_INSTANCE* inst = nullptr; STREAM_HANDLE* publisher = nullptr; uint64_t group = 0, epoch = 0, frameId = 0; QTimer poll; size_t allocations = 0; bool receive(uint64_t id, StreamRecord& record) { QElapsedTimer elapsed; elapsed.start(); while (elapsed.elapsed() < 1000) { if (gui_stream_pull(id, record)) return true; QTest::qWait(1); } DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop test missing record id=%llu", (unsigned long long)id); return false; } void publish(const QByteArray& video, const QByteArray& audio = {}) { stream_media_block media{}; media.format = {160, 90, 90, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS}; media.first_frame_offset = 0; media.video_count = 1; media.video[0] = {++frameId, frameId * 100000, static_cast(video.size()), 0, static_cast(video.size()), STREAM_MEDIA_KEY, reinterpret_cast(video.constData())}; if (!audio.isEmpty()) { media.audio_count = 1; media.audio[0] = {frameId * 100000, static_cast(audio.size()), 960, reinterpret_cast(audio.constData())}; } uint8_t wire[7080]; size_t length = 0; QCOMPARE(stream_media_encode(&media, wire, sizeof(wire), &length), STREAM_OK); QCOMPARE(stream_publish_block(publisher, wire, length), STREAM_OK); } private slots: void initTestCase() { debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR); allocations = u_get_allocated_count(); ua = uasync_create(); QVERIFY(ua); inst = utun_instance_create_from_str(ua, "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n"); QVERIFY(inst); QCOMPARE(utun_core_start(inst), 0); QCOMPARE(chat_service_start(inst), 0); chat_core_create_channel_auto(inst, "Stream test"); sqlite3_stmt* statement = nullptr; QCOMPARE(sqlite3_prepare_v2(chat_core_get_db(inst), "SELECT channel_id FROM channels LIMIT 1", -1, &statement, nullptr), SQLITE_OK); QCOMPARE(sqlite3_step(statement), SQLITE_ROW); group = QString::fromUtf8(reinterpret_cast(sqlite3_column_text(statement, 0))).toULongLong(); sqlite3_finalize(statement); QVERIFY(group); chat_event_set_handler(inst, sourceEvent); QVERIFY(gui_stream_attach(inst)); uasync_mark_running(ua); connect(&poll, &QTimer::timeout, this, [this] { uasync_poll(ua, 0); }); poll.start(1); } void catalogueWithoutViewer() { QVERIFY(gui_stream_watch(group)); QTRY_COMPARE(snapshots.size(), 1); QCOMPARE(snapshots.last().mid(9), QByteArray("[]")); QVERIFY(!stream_service_needs_activity(inst)); QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK); stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); epoch = info.descriptor.source_epoch; QTRY_COMPARE(snapshots.size(), 2); auto sources = QJsonDocument::fromJson(snapshots.last().mid(9)).array(); QCOMPARE(sources.size(), 1); QCOMPARE(stream_read_be(reinterpret_cast(snapshots.last().constData()), 8), group); QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); } void staleSelectionAndCancellation() { QVERIFY(gui_stream_open(100, group, inst->node_id, epoch ^ 1)); StreamRecord event; QVERIFY(receive(100, event)); QCOMPARE(event.kind, StreamRecordKind::Closed); QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED); QVERIFY(gui_stream_open(101, group, inst->node_id, epoch)); gui_stream_stop(101); QTest::qWait(20); stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); } void mediaAndPressure() { QVERIFY(gui_stream_open(102, group, inst->node_id, epoch)); StreamRecord event; QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Opened); const QByteArray frame(1500, 'k'); const QByteArray audio("opus"); publish(frame, audio); QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video); QCOMPARE(event.data, frame); QCOMPARE(event.format.rotation, uint16_t(90)); QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Audio); QCOMPARE(event.data, audio); for (int i = 0; i < 270; i++) { publish(frame); uasync_poll(ua, 0); } bool gap = false; unsigned records = 0; while (gui_stream_pull(102, event)) { gap |= event.kind == StreamRecordKind::Gap; records++; } QVERIFY(gap); QVERIFY(records <= 256); publish(frame); QVERIFY(receive(102, event)); QCOMPARE(event.kind, StreamRecordKind::Video); gui_stream_stop(102); QTest::qWait(20); stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); } void realH264RenderingAndClose() { #ifdef HAVE_FFMPEG const AVCodec* encoder = avcodec_find_encoder(AV_CODEC_ID_H264); if (!encoder) QSKIP("H264 encoder is unavailable for the decoder fixture"); AVCodecContext* context = avcodec_alloc_context3(encoder); QVERIFY(context); context->width = 160; context->height = 90; context->pix_fmt = AV_PIX_FMT_YUV420P; context->time_base = {1, 10}; context->framerate = {10, 1}; context->max_b_frames = 0; AVDictionary* options = nullptr; av_dict_set(&options, "preset", "ultrafast", 0); av_dict_set(&options, "tune", "zerolatency", 0); QCOMPARE(avcodec_open2(context, encoder, &options), 0); av_dict_free(&options); AVFrame* frame = av_frame_alloc(); QVERIFY(frame); frame->width = 160; frame->height = 90; frame->format = context->pix_fmt; QCOMPARE(av_frame_get_buffer(frame, 32), 0); for (int plane = 0; plane < 3; plane++) for (int y = 0; y < (plane ? 45 : 90); y++) memset(frame->data[plane] + y * frame->linesize[plane], plane ? 128 : 80, plane ? 80 : 160); frame->pts = 0; QCOMPARE(avcodec_send_frame(context, frame), 0); AVPacket* packet = av_packet_alloc(); QVERIFY(packet); QCOMPARE(avcodec_receive_packet(context, packet), 0); QByteArray video(reinterpret_cast(packet->data), packet->size); av_packet_free(&packet); av_frame_free(&frame); avcodec_free_context(&context); QByteArray audio; #ifdef TEST_NULL_AUDIO QVERIFY(SoundManager::instance()->init()); auto* opus = opus_codec_encoder_create(48000, 1); QVERIFY(opus); int16_t pcm[960]; for (unsigned i = 0; i < 960; i++) pcm[i] = static_cast(6000 * std::sin(i * 0.05)); uint8_t encoded[1275]; int length = opus_codec_encode(opus, pcm, 960, encoded, sizeof(encoded)); QVERIFY(length > 0); audio = QByteArray(reinterpret_cast(encoded), length); opus_codec_encoder_destroy(opus); nonzeroSamples = 0; #endif StreamPlayerWindow window(group, inst->node_id, epoch, "Source"); QSignalSpy shown(&window, &StreamPlayerWindow::framePresented); window.show(); QTimer sender; connect(&sender, &QTimer::timeout, this, [this, video, audio] { publish(video, audio); }); sender.start(100); QTRY_VERIFY_WITH_TIMEOUT(shown.count() > 0, 5000); #ifdef TEST_NULL_AUDIO QTRY_VERIFY_WITH_TIMEOUT(nonzeroSamples.load() > 0, 5000); auto* volume = window.findChild(); QVERIFY(volume); volume->setValue(0); QTest::qWait(50); uint64_t played = nonzeroSamples.load(); QTest::qWait(100); QCOMPARE(nonzeroSamples.load(), played); #endif sender.stop(); window.close(); QTest::qWait(20); stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); #else QSKIP("FFmpeg is disabled"); #endif } void finishAndDetachPendingOwner() { QCOMPARE(stream_publish_finish(publisher), STREAM_OK); QTRY_COMPARE(snapshots.last().mid(9), QByteArray("[]")); stream_close(publisher); publisher = nullptr; QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK); stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QVERIFY(info.descriptor.source_epoch != epoch); QVERIFY(gui_stream_open(103, group, inst->node_id, epoch)); StreamRecord event; QVERIFY(receive(103, event)); QCOMPARE(event.reason, STREAM_CLOSE_SOURCE_CHANGED); QVERIFY(gui_stream_open(104, group, inst->node_id, info.descriptor.source_epoch)); gui_stream_detach(inst); QVERIFY(!gui_stream_ready()); QTest::qWait(20); QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); } void channelClosureNotifiesViewer() { QVERIFY(gui_stream_attach(inst)); QVERIFY(gui_stream_watch(group)); stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QVERIFY(gui_stream_open(105, group, inst->node_id, info.descriptor.source_epoch)); StreamRecord event; QVERIFY(receive(105, event)); QCOMPARE(event.kind, StreamRecordKind::Opened); QByteArray channel = QByteArray::number(group); chat_core_delete_channel(inst, channel.constData()); QVERIFY(receive(105, event)); QCOMPARE(event.kind, StreamRecordKind::Closed); QCOMPARE(event.reason, STREAM_CLOSE_GROUP); QTRY_COMPARE(snapshots.last().mid(9), QByteArray("[]")); QVERIFY(!stream_handle_active(publisher)); QCOMPARE(stream_handle_reason(publisher), STREAM_CLOSE_GROUP); } void cleanupTestCase() { poll.stop(); gui_stream_detach(inst); stream_close(publisher); uasync_mark_stopped(ua); SoundManager::instance()->shutdown(); utun_instance_destroy(inst); uasync_destroy(ua, 0); QCOMPARE(u_get_allocated_count(), allocations); } }; QTEST_MAIN(StreamViewTest) #include "test_stream_view.moc"