chore: Expect event sink to match predicate.

pkgs/sdk/server/test/AssertHelpers.cs

@@ -1,7 +1,8 @@
-using LaunchDarkly.Logging;
+using System;
+using LaunchDarkly.Logging;
+using LaunchDarkly.TestHelpers;
 using Xunit;
 using Xunit.Sdk;
-
 using static LaunchDarkly.Sdk.Server.Subsystems.DataStoreTypes;
 using static LaunchDarkly.TestHelpers.JsonAssertions;
 
@@ -34,13 +35,48 @@ public static void LogMessageText(LogCapture logCapture, bool shouldHave, LogLev
             }
         }
 
-        private static void ThrowLogMatchException(LogCapture logCapture, bool shouldHave, LogLevel level, string text, bool isRegex) =>
+        private static void ThrowLogMatchException(LogCapture logCapture, bool shouldHave, LogLevel level, string text,
+            bool isRegex) =>
             throw new AssertActualExpectedException(shouldHave, !shouldHave,
                 string.Format("Expected log {0} the {1} \"{2}\" at level {3}\n\nActual log output follows:\n{4}",
                     shouldHave ? "to have" : "not to have",
                     isRegex ? "pattern" : "exact message",
                     text,
                     level,
                     logCapture.ToString()));
+
+        /// <summary>
+        /// Expect that the given sink will receive an event that passes the provided predicate within the specified
+        /// timeout.
+        ///
+        /// The total time for the execution of this method may be greater than the timeout, because its implementation
+        /// depends on a function which itself has a timeout.
+        /// </summary>
+        /// <param name="sink">the sink to check events from</param>
+        /// <param name="predicate">the predicate to run against events</param>
+        /// <param name="message">the message to show if the test fails</param>
+        /// <param name="timeout">the overall timeout</param>
+        /// <typeparam name="T">the type of the sink and predicate</typeparam>
+        public static void ExpectPredicate<T>(EventSink<T> sink, Predicate<T> predicate, string message,
+            TimeSpan timeout)
+        {
+            while (true)
+            {
+                var startTime = DateTimeOffset.Now.ToUnixTimeMilliseconds();
+
+                var value = sink.ExpectValue(timeout);
+
+                if (predicate(value))
+                {
+                    break;
+                }
+
+                if (!(DateTimeOffset.Now.ToUnixTimeMilliseconds() - startTime > timeout.TotalMilliseconds)) continue;
+
+                // XUnit 2.5+ adds Assert.Fail.
+                Assert.True(false, message);
+                return;
+            }
+        }
     }
 }

pkgs/sdk/server/test/Internal/DataSources/FileDataSourceTest.cs

@@ -1,5 +1,7 @@
 using System;
+using System.Linq;
 using System.Threading;
+using Castle.Core.Internal;
 using LaunchDarkly.Sdk.Server.Integrations;
 using LaunchDarkly.Sdk.Server.Interfaces;
 using LaunchDarkly.Sdk.Server.Internal.Model;
@@ -8,7 +10,6 @@
 using YamlDotNet.Serialization;
 using Xunit;
 using Xunit.Abstractions;
-
 using static LaunchDarkly.Sdk.Server.Subsystems.DataStoreTypes;
 using static LaunchDarkly.Sdk.Server.TestUtils;
 using static LaunchDarkly.TestHelpers.JsonAssertions;
@@ -24,7 +25,9 @@ public class FileDataSourceTest : BaseTest
         private readonly FileDataSourceBuilder factory = FileData.DataSource();
         private readonly Context user = Context.New("key");
 
-        public FileDataSourceTest(ITestOutputHelper testOutput) : base(testOutput) { }
+        public FileDataSourceTest(ITestOutputHelper testOutput) : base(testOutput)
+        {
+        }
 
         private IDataSource MakeDataSource() =>
             factory.Build(BasicContext.WithDataSourceUpdates(_updateSink));
@@ -148,9 +151,52 @@ public void ModifiedFileIsReloadedIfAutoUpdateIsOn()
 
                     file.SetContentFromPath(TestUtils.TestFilePath("segment-only.json"));
 
-                    var newData = _updateSink.Inits.ExpectValue(TimeSpan.FromSeconds(5));
-
-                    AssertJsonEqual(DataSetAsJson(ExpectedDataSetForSegmentOnlyFile(2)), DataSetAsJson(newData));
+                    AssertHelpers.ExpectPredicate(_updateSink.Inits, actual =>
+                        {
+                            var segments = actual.Data.First(item => item.Key == DataModel.Segments);
+                            var features = actual.Data.First(item => item.Key == DataModel.Features);
+                            if (!features.Value.Items.IsNullOrEmpty())
+                            {
+                                return false;
+                            }
+
+                            var segmentItems = segments.Value.Items.ToList();
+
+                            if (segmentItems.Count != 1)
+                            {
+                                return false;
+                            }
+
+                            var segmentDescriptor = segmentItems[0];
+                            if (segmentDescriptor.Key != "seg1")
+                            {
+                                return false;
+                            }
+
+                            if (segmentDescriptor.Value.Version == 1)
+                            {
+                                return false;
+                            }
+
+                            if (!(segmentDescriptor.Value.Item is Segment segment))
+                            {
+                                return false;
+                            }
+
+                            if (segment.Deleted)
+                            {
+                                return false;
+                            }
+
+                            if (segment.Included.Count != 1)
+                            {
+                                return false;
+                            }
+
+                            return segment.Included[0] == "user1";
+                        },
+                        "Did not receive expected update from the file data source.",
+                        TimeSpan.FromSeconds(30));
                 }
             }
         }