fix tests

Author: stephen-derosa

src/tests/integration/test_room_event_deduplication.cpp

@@ -18,9 +18,12 @@
 #include <chrono>
 #include <condition_variable>
 #include <map>
+#include <memory>
 #include <mutex>
+#include <sstream>
 #include <string>
 #include <thread>
+#include <vector>
 
 #include "../common/audio_utils.h"
 #include "../common/test_common.h"
@@ -42,6 +45,7 @@ struct RoomEventCounts {
   std::map<std::string, int> participant_disconnected;
   std::map<std::string, int> track_published;
   std::map<std::string, int> track_subscribed;
+  std::map<std::string, int> track_unsubscribed;
   std::map<std::string, int> track_unpublished;
   int disconnected = 0;
 };
@@ -51,6 +55,7 @@ struct RoomEventCountsSnapshot {
   std::map<std::string, int> participant_disconnected;
   std::map<std::string, int> track_published;
   std::map<std::string, int> track_subscribed;
+  std::map<std::string, int> track_unsubscribed;
   std::map<std::string, int> track_unpublished;
   int disconnected = 0;
 };
@@ -62,6 +67,7 @@ RoomEventCountsSnapshot snapshotCounts(RoomEventCounts& counts) {
   snapshot.participant_disconnected = counts.participant_disconnected;
   snapshot.track_published = counts.track_published;
   snapshot.track_subscribed = counts.track_subscribed;
+  snapshot.track_unsubscribed = counts.track_unsubscribed;
   snapshot.track_unpublished = counts.track_unpublished;
   snapshot.disconnected = counts.disconnected;
   return snapshot;
@@ -101,6 +107,13 @@ class RoomEventCounterDelegate : public RoomDelegate {
     notify([&]() { incrementMap(counts_.track_subscribed, event.publication->name()); });
   }
 
+  void onTrackUnsubscribed(Room&, const TrackUnsubscribedEvent& event) override {
+    if (event.publication == nullptr) {
+      return;
+    }
+    notify([&]() { incrementMap(counts_.track_unsubscribed, event.publication->name()); });
+  }
+
   void onTrackUnpublished(Room&, const TrackUnpublishedEvent& event) override {
     if (event.publication == nullptr) {
       return;
@@ -178,13 +191,97 @@ void expectCountsUnchangedAfterGrace(RoomEventCounts& counts, const RoomEventCou
                        (std::string(phase) + " track_published duplicate").c_str());
   expectMapCountsExact(after.track_subscribed, before.track_subscribed,
                        (std::string(phase) + " track_subscribed duplicate").c_str());
+  expectMapCountsExact(after.track_unsubscribed, before.track_unsubscribed,
+                       (std::string(phase) + " track_unsubscribed duplicate").c_str());
   expectMapCountsExact(after.track_unpublished, before.track_unpublished,
                        (std::string(phase) + " track_unpublished duplicate").c_str());
   EXPECT_EQ(after.disconnected, before.disconnected) << phase << " onDisconnected duplicate";
 }
 
 std::string makeUniqueTrackName(const std::string& prefix) { return prefix + "-" + std::to_string(getTimestampUs()); }
 
+std::string describeCounts(const std::map<std::string, int>& counts) {
+  std::ostringstream out;
+  bool first = true;
+  out << "{";
+  for (const auto& [key, count] : counts) {
+    if (!first) {
+      out << ", ";
+    }
+    first = false;
+    out << key << ": " << count;
+  }
+  out << "}";
+  return out.str();
+}
+
+class MediaLoopGuard {
+public:
+  MediaLoopGuard() = default;
+  MediaLoopGuard(const MediaLoopGuard&) = delete;
+  MediaLoopGuard& operator=(const MediaLoopGuard&) = delete;
+
+  ~MediaLoopGuard() { stop(); }
+
+  void addAudioSource(const std::shared_ptr<AudioSource>& source) {
+    threads_.emplace_back([this, source]() {
+      runToneLoop(source, running_, 440.0, false, kDefaultAudioSampleRate, kDefaultAudioChannels);
+    });
+  }
+
+  void addVideoSource(const std::shared_ptr<VideoSource>& source) {
+    threads_.emplace_back([this, source]() { runVideoLoop(source, running_, fillWebcamWrapper); });
+  }
+
+  void stop() {
+    running_.store(false, std::memory_order_relaxed);
+    for (auto& thread : threads_) {
+      if (thread.joinable()) {
+        thread.join();
+      }
+    }
+  }
+
+private:
+  std::atomic<bool> running_{true};
+  std::vector<std::thread> threads_;
+};
+
+class PublishedTrackGuard {
+public:
+  explicit PublishedTrackGuard(LocalParticipant* participant) : participant_(participant) {}
+  PublishedTrackGuard(const PublishedTrackGuard&) = delete;
+  PublishedTrackGuard& operator=(const PublishedTrackGuard&) = delete;
+
+  ~PublishedTrackGuard() noexcept {
+    try {
+      unpublishAll();
+    } catch (...) {
+    }
+  }
+
+  void addTrackSid(const std::string& sid) {
+    if (!sid.empty()) {
+      track_sids_.push_back(sid);
+    }
+  }
+
+  void unpublishAll() {
+    if (participant_ != nullptr) {
+      for (const auto& sid : track_sids_) {
+        if (!sid.empty()) {
+          participant_->unpublishTrack(sid);
+        }
+      }
+    }
+    track_sids_.clear();
+  }
+
+private:
+  LocalParticipant* participant_ = nullptr;
+  std::vector<std::string> track_sids_;
+};
+
 } // namespace
 
 class RoomEventDeduplicationIntegrationTest : public LiveKitTestBase, public ::testing::WithParamInterface<bool> {
@@ -222,6 +319,7 @@ TEST_P(RoomEventDeduplicationIntegrationTest, RoomLifecycleDelegateCallbacksFire
   const std::map<std::string, int> peer_identity_expected{{peer_identity, 1}};
   ASSERT_TRUE(waitForMapCountAtLeast(observer_counts, peer_identity_expected, 1, kEventWaitTimeout))
       << "Timed out waiting for onParticipantConnected";
+  ASSERT_TRUE(waitForParticipant(&observer_room, peer_identity, 10s)) << "Peer not visible to observer room";
   {
     const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts);
     expectMapCountsExact(snapshot.participant_connected, peer_identity_expected, "onParticipantConnected");
@@ -241,54 +339,46 @@ TEST_P(RoomEventDeduplicationIntegrationTest, RoomLifecycleDelegateCallbacksFire
   TrackPublishOptions video_opts;
   video_opts.source = TrackSource::SOURCE_CAMERA;
 
-  lockLocalParticipant(peer_room)->publishTrack(audio_track, audio_opts);
-  lockLocalParticipant(peer_room)->publishTrack(video_track, video_opts);
+  auto peer_participant = lockLocalParticipant(peer_room);
+  PublishedTrackGuard published_tracks(peer_participant.get());
+  MediaLoopGuard media_loops;
+
+  ASSERT_NO_THROW(peer_participant->publishTrack(audio_track, audio_opts));
   ASSERT_NE(audio_track->publication(), nullptr);
-  ASSERT_NE(video_track->publication(), nullptr);
+  published_tracks.addTrackSid(audio_track->publication()->sid());
+  media_loops.addAudioSource(audio_source);
 
-  std::atomic<bool> media_running{true};
-  std::thread audio_thread([&]() {
-    runToneLoop(audio_source, media_running, 440.0, false, kDefaultAudioSampleRate, kDefaultAudioChannels);
-  });
-  std::thread video_thread([&]() { runVideoLoop(video_source, media_running, fillWebcamWrapper); });
+  ASSERT_NO_THROW(peer_participant->publishTrack(video_track, video_opts));
+  ASSERT_NE(video_track->publication(), nullptr);
+  published_tracks.addTrackSid(video_track->publication()->sid());
+  media_loops.addVideoSource(video_source);
 
-  const std::map<std::string, int> expected_published_counts{{audio_track_name, 1}, {video_track_name, 1}};
   const std::map<std::string, int> expected_subscribed_counts{{audio_track_name, 1}, {video_track_name, 1}};
 
-  ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_published_counts, &RoomEventCounts::track_published,
-                                          kEventWaitTimeout))
-      << "Timed out waiting for onTrackPublished";
   ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_subscribed_counts,
                                           &RoomEventCounts::track_subscribed, kEventWaitTimeout))
-      << "Timed out waiting for onTrackSubscribed";
+      << "Timed out waiting for onTrackSubscribed; observed track_subscribed="
+      << describeCounts(snapshotCounts(observer_counts).track_subscribed);
 
   {
     const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts);
-    expectMapCountsExact(snapshot.track_published, expected_published_counts, "onTrackPublished");
     expectMapCountsExact(snapshot.track_subscribed, expected_subscribed_counts, "onTrackSubscribed");
     expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after track subscribe");
   }
 
-  media_running.store(false, std::memory_order_relaxed);
-  if (audio_thread.joinable()) {
-    audio_thread.join();
-  }
-  if (video_thread.joinable()) {
-    video_thread.join();
-  }
-
-  lockLocalParticipant(peer_room)->unpublishTrack(audio_track->publication()->sid());
-  lockLocalParticipant(peer_room)->unpublishTrack(video_track->publication()->sid());
+  media_loops.stop();
+  published_tracks.unpublishAll();
 
-  const std::map<std::string, int> expected_unpublished_counts{{audio_track_name, 1}, {video_track_name, 1}};
-  ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_unpublished_counts,
-                                          &RoomEventCounts::track_unpublished, kEventWaitTimeout))
-      << "Timed out waiting for onTrackUnpublished";
+  const std::map<std::string, int> expected_unsubscribed_counts{{audio_track_name, 1}, {video_track_name, 1}};
+  ASSERT_TRUE(waitForMapCountAtLeastTrack(observer_counts, expected_unsubscribed_counts,
+                                          &RoomEventCounts::track_unsubscribed, kEventWaitTimeout))
+      << "Timed out waiting for onTrackUnsubscribed; observed track_unsubscribed="
+      << describeCounts(snapshotCounts(observer_counts).track_unsubscribed);
 
   {
     const RoomEventCountsSnapshot snapshot = snapshotCounts(observer_counts);
-    expectMapCountsExact(snapshot.track_unpublished, expected_unpublished_counts, "onTrackUnpublished");
-    expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after track unpublished");
+    expectMapCountsExact(snapshot.track_unsubscribed, expected_unsubscribed_counts, "onTrackUnsubscribed");
+    expectCountsUnchangedAfterGrace(observer_counts, snapshot, "after track unsubscribed");
   }
 
   peer_room.disconnect();