Files
obs-streamer-tools-plugin/core/tests/test_session.cpp
shadowdaoandClaude Sonnet 5 80904a3e85
Build / macOS (macos-latest) (push) Successful in 14s
Build / Linux (ubuntu-latest) (push) Successful in 41s
Build / Windows (windows-latest) (push) Failing after 8m18s
Add the LiveKit session wrapper, verified end-to-end against a real room
stplugin::LiveKitSession wraps livekit::Room for exactly one subscribed slot:
connect with the wsUrl/lkToken the API client minted, find the chosen
participant's camera (and microphone), and hand decoded frames to
callback-shaped handlers the OBS adapter can consume directly.

Two architectural decisions worth recording, both forced by reading the SDK
rather than guessed:

1. Frames come from VideoStream/AudioStream::fromTrack with our own reader
   threads, NOT from Room::setOnVideoFrameCallback. The dispatcher API is
   keyed by (participant identity, track NAME), which we cannot know before
   the track is published -- and disassembling liblivekit.so confirms that
   both Room::setOnVideoFrameCallback and the dispatcher's own
   setOnVideoFrameCallback merely record the registration: neither starts a
   reader for a track that is already subscribed. Registering after the
   subscription event, which is the only time the track name exists, would
   therefore have silently produced no video. Taking the shared_ptr<Track>
   straight off the TrackSubscribedEvent sidesteps the name entirely, and
   lets us pick the camera by TrackSource (streamer-tools publishes cameras
   as Source.Camera and screenshares separately -- apps/web/src/avatar/
   publish.ts), which is what we actually mean.

2. Every stream operation runs on one owned worker thread, never on a room
   event thread. The SDK documents that Room::disconnect() from inside a
   delegate callback deadlocks, and Room's own event dispatch holds a mutex,
   so delegate callbacks only ever enqueue a command here.

VideoStream::Options::capacity is set (3 frames) so the SDK's queue is a
drop-oldest ring buffer: a stalled consumer can only fall three frames
behind, and what it then sees is the newest frame rather than a backlog.
That is the structural answer to the stale-media bug that motivated this
plugin.

The pure decision-making -- the state machine, track selection, frame
geometry validation -- lives in session_types.h/.cpp with no LiveKit or OBS
types, so it is unit-testable headlessly (81 checks in test_session,
including the publisher-swap and reconnect transitions, plus the real
connect() failure paths against the real SDK: unreachable host, garbage
token, incomplete config, and destruction mid-connect).

test_integration_livekit is the test that proves media actually flows. It
publishes a synthetic camera and microphone into a real LiveKit room using
the same SDK, subscribes through LiveKitSession, and asserts on the exact
fields the OBS adapter will dereference. It skips (exit 0) unless
STPLUGIN_IT_* is set, so the three build runners stay green;
scripts/livekit-dev-room.py mints the tokens for a local `livekit-server
--dev`.

Verified locally against livekit-server 1.13.6 in dev mode:

  integration_livekit: 36 video frames, 323 audio frames, 10 state changes
  integration_livekit: 32 checks passed

covering: connect; subscribe to the named participant's camera; 320x240
I420 frames with three planes, non-null plane pointers and strides >= the
frame's own width; 48kHz audio; unpublish -> hasVideo() false, state stays
Connected (a dark camera is the placeholder state, never an error) and NO
further frames arrive from the dead publisher; republish -> video resumes;
clean disconnect.

One real finding from that run, now handled: WebRTC ramps a new subscription
up from a downscaled spatial layer, so the first frames after (re)subscribing
legitimately arrive smaller than what is being published. The OBS adapter
must cope with a mid-stream resolution change; the test asserts per-frame
geometry rather than the publisher's, and separately asserts the stream does
reach full size.

Full suite: ctest -> 6/6 passed.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RL8abRmgFXkVASHkkqiJbE
2026-09-06 21:39:49 -07:00

356 lines
13 KiB
C++

/*
streamer-tools OBS Camera Plugin - session wrapper tests
Copyright (C) 2026 CyberCoveLLC <jknapp85@gmail.com>
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License along
with this program. If not, see <https://www.gnu.org/licenses/>
*/
// What is and is not covered here, stated plainly because it matters:
//
// COVERED headlessly -- the session's own decision-making: the state
// machine's transitions (including the publisher-swap and reconnect paths
// that motivated this plugin), track selection, frame geometry validation,
// and the real connect() failure paths against the real SDK (bad URL,
// unreachable host, garbage token).
//
// NOT COVERED here -- anything that needs a LiveKit server to answer:
// a successful connect, actual subscription, and actual decoded frames
// reaching the handlers. Those can only be verified against a real room,
// and the design doc's Testing section puts that in the integration-test /
// manual-sign-off bucket.
#include <atomic>
#include <chrono>
#include <string>
#include <thread>
#include "stplugin/session.h"
#include "stplugin/session_types.h"
#include "test_util.h"
using namespace stplugin;
namespace {
// ---------------------------------------------------------------------------
// Pure logic
// ---------------------------------------------------------------------------
void testFrameGeometry()
{
ST_ASSERT_EQ(planeCount(PixelFormat::I420), 3);
ST_ASSERT_EQ(planeCount(PixelFormat::NV12), 2);
ST_ASSERT_EQ(planeCount(PixelFormat::BGRA), 1);
// 1280x720 I420: 921600 luma + 2 * 230400 chroma.
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 1280, 720), std::size_t(1382400));
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::NV12, 1280, 720), std::size_t(1382400));
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::BGRA, 1280, 720), std::size_t(3686400));
// Odd dimensions round the chroma planes up, the way libyuv does.
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 3, 3), std::size_t(9 + 2 * 4));
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 1, 1), std::size_t(1 + 2));
// Degenerate geometry is 0, which the reader treats as "drop the frame".
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 0, 720), std::size_t(0));
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, 1280, 0), std::size_t(0));
ST_ASSERT_EQ(expectedFrameBytes(PixelFormat::I420, -1, -1), std::size_t(0));
ST_ASSERT_EQ(std::string(describePixelFormat(PixelFormat::I420)), std::string("I420"));
}
void testTrackSelection()
{
const std::string want = "cam1";
// The camera we asked for.
ST_ASSERT(isWantedVideoTrack(want, "cam1", MediaKind::Video, MediaSource::Camera));
// A video track with no declared source is taken on kind alone.
ST_ASSERT(isWantedVideoTrack(want, "cam1", MediaKind::Video, MediaSource::Unknown));
// Someone else's camera.
ST_ASSERT(!isWantedVideoTrack(want, "cam2", MediaKind::Video, MediaSource::Camera));
// The right participant's screenshare is explicitly NOT the camera --
// streamer-tools publishes those as separate sources.
ST_ASSERT(!isWantedVideoTrack(want, "cam1", MediaKind::Video, MediaSource::Screenshare));
// Their microphone is not a video track.
ST_ASSERT(!isWantedVideoTrack(want, "cam1", MediaKind::Audio, MediaSource::Microphone));
// No selection means nothing matches -- never "the first thing we see".
ST_ASSERT(!isWantedVideoTrack("", "cam1", MediaKind::Video, MediaSource::Camera));
ST_ASSERT(!isWantedVideoTrack("", "", MediaKind::Video, MediaSource::Camera));
ST_ASSERT(isWantedAudioTrack(want, "cam1", MediaKind::Audio, MediaSource::Microphone));
ST_ASSERT(isWantedAudioTrack(want, "cam1", MediaKind::Audio, MediaSource::Unknown));
ST_ASSERT(!isWantedAudioTrack(want, "cam1", MediaKind::Audio, MediaSource::ScreenshareAudio));
ST_ASSERT(!isWantedAudioTrack(want, "cam1", MediaKind::Video, MediaSource::Camera));
ST_ASSERT(!isWantedAudioTrack(want, "other", MediaKind::Audio, MediaSource::Microphone));
}
void testStateMachineHappyPath()
{
SessionStateMachine m;
ST_ASSERT(m.state() == SessionState::Idle);
ST_ASSERT(!m.hasVideo());
ST_ASSERT(!m.waitingForCamera());
m.onConnectRequested();
ST_ASSERT(m.state() == SessionState::Connecting);
// Connecting is not "waiting for camera": the placeholder belongs to a
// live connection with a dark slot, not to a connection in progress.
ST_ASSERT(!m.waitingForCamera());
m.onConnectSucceeded();
ST_ASSERT(m.state() == SessionState::Connected);
ST_ASSERT(m.waitingForCamera());
m.onVideoAttached();
ST_ASSERT(m.hasVideo());
ST_ASSERT(!m.waitingForCamera());
m.onAudioAttached();
ST_ASSERT(m.hasAudio());
m.onLocalDisconnect();
ST_ASSERT(m.state() == SessionState::Disconnected);
ST_ASSERT(!m.hasVideo());
ST_ASSERT(!m.hasAudio());
}
void testPublisherSwapIsNotAnError()
{
// The motivating bug: a slot's publisher restarts mid-show. That must
// read as "waiting for camera", never as a failure, and the connection
// state must not move at all.
SessionStateMachine m;
m.onConnectRequested();
m.onConnectSucceeded();
m.onVideoAttached();
m.onVideoDetached();
ST_ASSERT(m.state() == SessionState::Connected);
ST_ASSERT(!m.hasVideo());
ST_ASSERT(m.waitingForCamera());
ST_ASSERT(m.detail().empty());
m.onVideoAttached();
ST_ASSERT(m.state() == SessionState::Connected);
ST_ASSERT(m.hasVideo());
ST_ASSERT(!m.waitingForCamera());
}
void testReconnect()
{
SessionStateMachine m;
m.onConnectRequested();
m.onConnectSucceeded();
m.onVideoAttached();
m.onReconnecting();
ST_ASSERT(m.state() == SessionState::Reconnecting);
// Tracks are re-subscribed on the far side, so video is not live yet.
ST_ASSERT(!m.hasVideo());
ST_ASSERT(m.waitingForCamera());
ST_ASSERT_EQ(m.detail(), std::string("reconnecting"));
m.onReconnected();
ST_ASSERT(m.state() == SessionState::Connected);
ST_ASSERT(m.detail().empty());
// A stray reconnect notification after a hard failure must not resurrect
// the session.
SessionStateMachine dead;
dead.onConnectRequested();
dead.onConnectFailed("token rejected");
dead.onReconnecting();
ST_ASSERT(dead.state() == SessionState::Failed);
dead.onReconnected();
ST_ASSERT(dead.state() == SessionState::Failed);
}
void testFailureAndRecovery()
{
SessionStateMachine m;
m.onConnectRequested();
m.onConnectFailed("token rejected");
ST_ASSERT(m.state() == SessionState::Failed);
ST_ASSERT_EQ(m.detail(), std::string("token rejected"));
ST_ASSERT(!m.waitingForCamera());
// A fresh attempt clears the stale reason, so a healthy connection can
// never be shown next to the previous failure's message.
m.onConnectRequested();
ST_ASSERT(m.detail().empty());
m.onConnectSucceeded();
ST_ASSERT(m.state() == SessionState::Connected);
ST_ASSERT(m.detail().empty());
// A fatal room end (duplicate identity, token rejected) is Failed; an
// ordinary drop is Disconnected.
SessionStateMachine fatal;
fatal.onConnectRequested();
fatal.onConnectSucceeded();
fatal.onRoomEnded("another client joined with the same identity", true);
ST_ASSERT(fatal.state() == SessionState::Failed);
SessionStateMachine dropped;
dropped.onConnectRequested();
dropped.onConnectSucceeded();
dropped.onRoomEnded("the signalling connection closed", false);
ST_ASSERT(dropped.state() == SessionState::Disconnected);
// Room-ended events after we are already down are ignored, so a late
// event cannot overwrite the reason the operator needs to see.
SessionStateMachine idle;
idle.onRoomEnded("stray", true);
ST_ASSERT(idle.state() == SessionState::Idle);
}
// ---------------------------------------------------------------------------
// Real SDK, failure paths only (no LiveKit server available headlessly)
// ---------------------------------------------------------------------------
void testConnectRejectsIncompleteConfig()
{
LiveKitSession session;
std::atomic<int> state_calls{0};
session.setStateHandler([&](SessionState, const std::string &) { state_calls.fetch_add(1); });
SessionConfig config;
config.ws_url = "";
config.token = "t";
config.participant_identity = "cam1";
ST_ASSERT(!session.connect(config));
ST_ASSERT(session.state() == SessionState::Failed);
ST_ASSERT(!session.stateDetail().empty());
config.ws_url = "ws://127.0.0.1:1";
config.token = "";
ST_ASSERT(!session.connect(config));
ST_ASSERT(session.state() == SessionState::Failed);
config.token = "t";
config.participant_identity = "";
ST_ASSERT(!session.connect(config));
ST_ASSERT(session.state() == SessionState::Failed);
// The state handler fired for each attempt (Connecting + Failed).
ST_ASSERT(state_calls.load() >= 6);
// Frame counters stay at zero and nothing crashes on teardown.
ST_ASSERT_EQ(session.videoFrameCount(), std::uint64_t(0));
ST_ASSERT_EQ(session.audioFrameCount(), std::uint64_t(0));
session.disconnect();
session.disconnect(); // idempotent
ST_ASSERT(session.state() == SessionState::Failed || session.state() == SessionState::Disconnected);
}
void testConnectToUnreachableServerFailsCleanly()
{
// Port 1 on loopback: nothing is listening, and the connection is
// refused immediately rather than hanging. This exercises the real
// livekit::Room::connect() failure path, with a real (garbage) token.
LiveKitSession session;
std::atomic<int> video_frames{0};
session.setVideoHandler([&](const VideoFrameData &) { video_frames.fetch_add(1); });
SessionConfig config;
config.ws_url = "ws://127.0.0.1:1";
config.token = "not.a.real.token";
config.participant_identity = "cam1";
config.connect_timeout_ms = 3000;
const auto start = std::chrono::steady_clock::now();
const bool ok = session.connect(config);
const auto elapsed = std::chrono::steady_clock::now() - start;
ST_ASSERT(!ok);
ST_ASSERT(session.state() == SessionState::Failed);
ST_ASSERT(!session.stateDetail().empty());
ST_ASSERT_EQ(video_frames.load(), 0);
// Must not sit on the caller's thread indefinitely -- this runs on an OBS
// thread in the real adapter.
ST_ASSERT(std::chrono::duration_cast<std::chrono::seconds>(elapsed).count() < 60);
session.disconnect();
}
void testConnectToNonLiveKitServerFailsCleanly()
{
// A URL that resolves and connects but is not a LiveKit signalling
// endpoint. The realistic operator mistake: pasting the app URL.
LiveKitSession session;
SessionConfig config;
config.ws_url = "ws://127.0.0.1:1/rtc";
config.token = "eyJhbGciOiJIUzI1NiJ9.bm90YXRva2Vu.x";
config.participant_identity = "cam1";
config.connect_timeout_ms = 3000;
ST_ASSERT(!session.connect(config));
ST_ASSERT(session.state() == SessionState::Failed);
session.disconnect();
}
void testDestroyWithoutDisconnect()
{
// The OBS adapter destroys sources without necessarily having called
// disconnect() first (an OBS shutdown mid-connect, say). The destructor
// must join every thread it started rather than terminating.
{
LiveKitSession session;
SessionConfig config;
config.ws_url = "ws://127.0.0.1:1";
config.token = "t";
config.participant_identity = "cam1";
config.connect_timeout_ms = 2000;
(void)session.connect(config);
}
ST_ASSERT(true); // reaching here at all is the assertion
}
void testGlobalInitIsReferenceCounted()
{
// Several OBS sources may each hold the SDK open; the last one out turns
// the lights off, and an unbalanced extra shutdown must not underflow.
LiveKitSession::globalInitialize();
LiveKitSession::globalInitialize();
LiveKitSession::globalShutdown();
LiveKitSession::globalShutdown();
LiveKitSession::globalShutdown(); // extra, must be harmless
LiveKitSession::globalInitialize();
LiveKitSession::globalShutdown();
ST_ASSERT(true);
}
} // namespace
int main()
{
testFrameGeometry();
testTrackSelection();
testStateMachineHappyPath();
testPublisherSwapIsNotAnError();
testReconnect();
testFailureAndRecovery();
LiveKitSession::globalInitialize();
testConnectRejectsIncompleteConfig();
testConnectToUnreachableServerFailsCleanly();
testConnectToNonLiveKitServerFailsCleanly();
testDestroyWithoutDisconnect();
LiveKitSession::globalShutdown();
testGlobalInitIsReferenceCounted();
return st_test_report("session");
}