Test: figure out which test is not closing

cognitedata · Aug 21, 2024 · 288653f · 288653f
1 parent 292de4d
commit 288653f
Show file tree

Hide file tree

Showing 11 changed files with 67 additions and 41 deletions.
diff --git a/Extractor/SessionManager.cs b/Extractor/SessionManager.cs
@@ -49,7 +49,6 @@ public class SessionManager : IDisposable
 
         public SessionContext Context { get; }
 
-
         public SessionManager(FullConfig config, UAClient parent, ILogger log)
         {
             client = parent;

diff --git a/Extractor/State.cs b/Extractor/State.cs
@@ -52,8 +52,8 @@ public class State
         private readonly ConcurrentDictionary<NodeId, MappedNode> mappedNodes =
             new ConcurrentDictionary<NodeId, MappedNode>();
 
-        public IEnumerable<VariableExtractionState> NodeStates => nodeStates.Values;
-        public IEnumerable<EventExtractionState> EmitterStates => emitterStates.Values;
+        public ICollection<VariableExtractionState> NodeStates => nodeStates.Values;
+        public ICollection<EventExtractionState> EmitterStates => emitterStates.Values;
 
         /// <summary>
         /// Return a NodeExtractionState by externalId

diff --git a/Extractor/UAExtractor.cs b/Extractor/UAExtractor.cs
@@ -569,6 +569,22 @@ public void OnCreatedSubscription(SubscriptionName subscription)
                     break;
             }
         }
+
+        public void RemoveKnownSubscription(SubscriptionName name)
+        {
+            lock (activeSubscriptions)
+            {
+                activeSubscriptions.Remove(name);
+            }
+        }
+
+        public void ClearKnownSubscriptions()
+        {
+            lock (activeSubscriptions)
+            {
+                activeSubscriptions.Clear();
+            }
+        }
         #endregion
 
         #region Mapping

diff --git a/ExtractorLauncher/ExtractorStarter.cs b/ExtractorLauncher/ExtractorStarter.cs
@@ -221,7 +221,7 @@ private static void VerifyAndBuildConfig(
 
             if (options != null)
             {
-                options.Restart |= config.Source.ExitOnFailure;
+                options.Restart &= !config.Source.ExitOnFailure;
             }
 
             string? configResult = VerifyConfig(log, config);
@@ -368,6 +368,7 @@ public static async Task RunExtractor(ILogger? log, ExtractorParams setup, Servi
 
             services.AddSingleton<UAClient>();
 
+            log.LogInformation("Start extractor: {R}", setup.Exit);
             var options = new ExtractorRunnerParams<FullConfig, UAExtractor>
             {
                 ConfigPath = setup.ConfigFile ?? Path.Join(configDir, "config.yml"),

diff --git a/Test/Integration/DataPointTests.cs b/Test/Integration/DataPointTests.cs
@@ -783,11 +783,9 @@ public async Task TestDisableSubscriptions()
             async Task Reset()
             {
                 extractor.State.Clear();
-                extractor.GetType().GetField("subscribed", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, 0);
-                extractor.GetType().GetField("subscribeFlag", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, false);
                 var reader = (HistoryReader)extractor.GetType().GetField("historyReader", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(extractor);
                 reader.AddIssue(HistoryReader.StateIssue.NodeHierarchyRead);
-                await tester.RemoveSubscription(SubscriptionName.DataPoints);
+                await tester.RemoveSubscription(extractor, SubscriptionName.DataPoints);
             }
 
             tester.Config.History.Enabled = true;
@@ -820,7 +818,7 @@ async Task Reset()
             Assert.False(state.ShouldSubscribe);
             state = extractor.State.GetNodeState(ids.IntVar);
             Assert.False(state.ShouldSubscribe);
-            await extractor.WaitForSubscription(SubscriptionName.DataPoints);
+            await Assert.ThrowsAsync<TimeoutException>(async () => await extractor.WaitForSubscription(SubscriptionName.Events, 20));
             Assert.DoesNotContain(session.Subscriptions, sub => sub.DisplayName.StartsWith(SubscriptionName.DataPoints.Name(), StringComparison.InvariantCulture));
             await TestUtils.WaitForCondition(() => CommonTestUtils.TestMetricValue("opcua_frontfill_data_count", 2, tester.Log), 5);
 
@@ -955,9 +953,9 @@ public async Task TestRestartHistoryOnReconnect()
 
             // First start the extractor and read the first half of history.
             var runTask = extractor.RunExtractor();
-            await extractor.WaitForSubscription(SubscriptionName.DataPoints);
 
-            await TestUtils.WaitForCondition(() => extractor.State.NodeStates.All(node =>
+            await TestUtils.WaitForCondition(() => extractor.State.NodeStates.Count > 0
+                && extractor.State.NodeStates.All(node =>
                 !node.IsFrontfilling && !node.IsBackfilling) && pusher.DataPoints[(ids.DoubleVar1, -1)].Count == 500, 10);
             var state = extractor.State.GetNodeState(ids.DoubleVar1);
 

diff --git a/Test/Integration/EventTests.cs b/Test/Integration/EventTests.cs
@@ -195,11 +195,9 @@ public async Task TestDisableSubscriptions()
             async Task Reset()
             {
                 extractor.State.Clear();
-                extractor.GetType().GetField("subscribed", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, 0);
-                extractor.GetType().GetField("subscribeFlag", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, false);
                 var reader = (HistoryReader)extractor.GetType().GetField("historyReader", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(extractor);
                 reader.AddIssue(HistoryReader.StateIssue.NodeHierarchyRead);
-                await tester.RemoveSubscription(SubscriptionName.Events);
+                await tester.RemoveSubscription(extractor, SubscriptionName.Events);
             }
 
             tester.Config.Extraction.RootNode = CommonTestUtils.ToProtoNodeId(ids.Root, tester.Client);
@@ -226,7 +224,7 @@ async Task Reset()
             Assert.False(state.ShouldSubscribe);
             state = extractor.State.GetEmitterState(ObjectIds.Server);
             Assert.False(state.ShouldSubscribe);
-            await extractor.WaitForSubscription(SubscriptionName.Events);
+            await extractor.WaitForSubscription(SubscriptionName.DataPoints);
             Assert.DoesNotContain(session.Subscriptions, sub => sub.DisplayName.StartsWith(SubscriptionName.Events.Name(), StringComparison.InvariantCulture));
             await TestUtils.WaitForCondition(() => extractor.State.EmitterStates.All(s => !s.IsFrontfilling), 10);
 

diff --git a/Test/Integration/LauncherTests.cs b/Test/Integration/LauncherTests.cs
@@ -24,6 +24,8 @@ public class LauncherTestFixture : LoggingTestFixture, IAsyncLifetime
         public ServerController Server { get; }
         public string EndpointUrl => $"opc.tcp://localhost:{Port}";
 
+        public CancellationTokenSource Source { get; } = new CancellationTokenSource();
+
         public LauncherTestFixture()
         {
             Port = CommonTestUtils.NextPort;
@@ -44,13 +46,15 @@ public async Task InitializeAsync()
             await Server.Start();
         }
 
-        public Task DisposeAsync()
+        public async Task DisposeAsync()
         {
             Server?.Dispose();
-            return Task.CompletedTask;
+            await Source.CancelAsync();
+            Source.Dispose();
         }
     }
 
+    [Collection("Program tests")]
     public sealed class LauncherTests : IClassFixture<LauncherTestFixture>, IDisposable
     {
         private readonly LauncherTestFixture tester;
@@ -62,21 +66,29 @@ public LauncherTests(ITestOutputHelper output, LauncherTestFixture tester)
             tester.Init(output);
             Program.CommandDryRun = false;
             Program.OnLaunch = (s, o) => CommonBuild(s);
+            Program.RootToken = tester.Source.Token;
             ExtractorStarter.OnCreateExtractor = (d, e) =>
             {
-                if (extractor != null)
+                lock (tester)
                 {
-                    extractor.Dispose();
+                    if (extractor != null)
+                    {
+                        extractor.Dispose();
+                    }
+                    extractor = e;
                 }
-                extractor = e;
             };
         }
 
         public void Dispose()
         {
             pusher?.Dispose();
             extractor?.Close().Wait();
-            extractor?.Dispose();
+            lock (tester)
+            {
+                extractor?.Dispose();
+            }
+
         }
 
         private void CommonBuild(ServiceCollection services)
@@ -172,7 +184,7 @@ public async Task TestRunExtractorNoConfig()
             }
         }
 
-        [Fact(Timeout = 30000)]
+        [Fact]
         public async Task TestRunExtractorToolConfig()
         {
             var args = new[]
@@ -199,7 +211,6 @@ public async Task TestRunExtractorToolConfig()
             finally
             {
                 await extractor?.Close();
-                await Task.WhenAny(task, Task.Delay(5000));
             }
         }
 

diff --git a/Test/Integration/NodeExtractionTests.cs b/Test/Integration/NodeExtractionTests.cs
@@ -823,6 +823,10 @@ public async Task TestLateInitLateFail(bool failNodes, bool failReferences)
             var runTask = extractor.RunExtractor();
             await extractor.WaitForSubscription(SubscriptionName.DataPoints);
 
+            await TestUtils.WaitForCondition(() =>
+                pusher.PendingNodes.Objects != null
+                && pusher.PendingNodes.Variables != null, 10);
+
             if (failNodes)
             {
                 Assert.Empty(pusher.PushedNodes);

diff --git a/Test/Integration/RebrowseTriggerManagerTests.cs b/Test/Integration/RebrowseTriggerManagerTests.cs
@@ -63,7 +63,7 @@ public async Task TestRebrowseIsNotSubscribed(RebrowseTriggersConfig config)
 
             // Act
             var runTask = extractor.RunExtractor();
-            await extractor.WaitForSubscription(SubscriptionName.RebrowseTriggers);
+            await extractor.WaitForSubscription(SubscriptionName.DataPoints);
 
             // Assert
             Assert.False(tester.TryGetSubscription(SubscriptionName.RebrowseTriggers, out var _));

diff --git a/Test/Unit/CDFPusherTest.cs b/Test/Unit/CDFPusherTest.cs
@@ -1024,7 +1024,7 @@ public async Task TestGetNodesFromCDF()
             Assert.Empty(result.SourceObjects);
             Assert.Equal(3, result.SourceVariables.Count());
 
-            Assert.Equal(3, extractor.State.NodeStates.Count());
+            Assert.Equal(3, extractor.State.NodeStates.Count);
             extractor.State.Clear();
 
             // Events
@@ -1107,14 +1107,14 @@ public async Task TestCDFAsSourceData()
             Assert.Empty(extractor.State.NodeStates);
             tester.Config.Cognite.RawNodeBuffer.BrowseOnEmpty = true;
             await extractor.RunExtractor(true);
-            Assert.True(extractor.State.NodeStates.Any());
+            Assert.True(extractor.State.NodeStates.Count > 0);
             Assert.True(handler.AssetsRaw.Count != 0);
             Assert.True(handler.TimeseriesRaw.Count != 0);
             Assert.True(handler.Timeseries.Count != 0);
             Assert.Empty(handler.Assets);
 
             await extractor.WaitForSubscription(SubscriptionName.DataPoints);
-            await tester.RemoveSubscription(SubscriptionName.DataPoints);
+            await tester.RemoveSubscription(extractor, SubscriptionName.DataPoints);
 
             extractor.State.Clear();
 
@@ -1123,12 +1123,11 @@ public async Task TestCDFAsSourceData()
             string oldAssets = System.Text.Json.JsonSerializer.Serialize(handler.AssetsRaw);
             string oldTimeseries = System.Text.Json.JsonSerializer.Serialize(handler.TimeseriesRaw);
             handler.Timeseries.Clear();
-            extractor.GetType().GetField("subscribed", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, 0);
-            extractor.GetType().GetField("subscribeFlag", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, false);
+            extractor.ClearKnownSubscriptions();
             var reader = (HistoryReader)extractor.GetType().GetField("historyReader", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(extractor);
             reader.AddIssue(HistoryReader.StateIssue.NodeHierarchyRead);
             await extractor.RunExtractor(true);
-            Assert.True(extractor.State.NodeStates.Any());
+            Assert.True(extractor.State.NodeStates.Count > 0);
 
             string newAssets = System.Text.Json.JsonSerializer.Serialize(handler.AssetsRaw);
             string newTimeseries = System.Text.Json.JsonSerializer.Serialize(handler.TimeseriesRaw);
@@ -1200,14 +1199,14 @@ public async Task TestCDFAsSourceEvents()
             Assert.Empty(extractor.State.NodeStates);
             tester.Config.Cognite.RawNodeBuffer.BrowseOnEmpty = true;
             await extractor.RunExtractor(true);
-            Assert.True(extractor.State.NodeStates.Any());
+            Assert.True(extractor.State.NodeStates.Count > 0);
             Assert.True(handler.AssetsRaw.Count != 0);
             Assert.True(handler.TimeseriesRaw.Count != 0);
             Assert.True(handler.Timeseries.Count != 0);
             Assert.Empty(handler.Assets);
 
             await extractor.WaitForSubscription(SubscriptionName.Events);
-            await tester.RemoveSubscription(SubscriptionName.Events);
+            await tester.RemoveSubscription(extractor, SubscriptionName.Events);
 
             extractor.State.Clear();
 
@@ -1216,10 +1215,9 @@ public async Task TestCDFAsSourceEvents()
             string oldAssets = JsonSerializer.Serialize(handler.AssetsRaw);
             string oldTimeseries = JsonSerializer.Serialize(handler.TimeseriesRaw);
             handler.Timeseries.Clear();
-            extractor.GetType().GetField("subscribed", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, 0);
-            extractor.GetType().GetField("subscribeFlag", BindingFlags.NonPublic | BindingFlags.Instance).SetValue(extractor, false);
+            extractor.ClearKnownSubscriptions();
             await extractor.RunExtractor(true);
-            Assert.True(extractor.State.NodeStates.Any());
+            Assert.True(extractor.State.NodeStates.Count > 0);
 
             string newAssets = JsonSerializer.Serialize(handler.AssetsRaw);
             string newTimeseries = JsonSerializer.Serialize(handler.TimeseriesRaw);
@@ -1274,14 +1272,14 @@ public async Task TestCDFNodeSetBackground()
             // Populate data in Raw
             tester.Config.Cognite.RawNodeBuffer.BrowseOnEmpty = true;
             await extractor.RunExtractor(true);
-            Assert.True(extractor.State.NodeStates.Any());
+            Assert.True(extractor.State.NodeStates.Count > 0);
             Assert.True(handler.AssetsRaw.Count != 0);
             Assert.True(handler.TimeseriesRaw.Count != 0);
             Assert.True(handler.Timeseries.Count != 0);
             Assert.Empty(handler.Assets);
 
             await extractor.WaitForSubscription(SubscriptionName.Events);
-            await tester.RemoveSubscription(SubscriptionName.Events);
+            await tester.RemoveSubscription(extractor, SubscriptionName.Events);
 
             extractor.State.Clear();
 

diff --git a/Test/Utils/BaseExtractorTestFixture.cs b/Test/Utils/BaseExtractorTestFixture.cs
@@ -127,10 +127,10 @@ public UAExtractor BuildExtractor(bool clear = true, IExtractionStateStore state
         {
             if (clear)
             {
-                RemoveSubscription(SubscriptionName.Events).Wait();
-                RemoveSubscription(SubscriptionName.DataPoints).Wait();
-                RemoveSubscription(SubscriptionName.Audit).Wait();
-                RemoveSubscription(SubscriptionName.RebrowseTriggers).Wait();
+                RemoveSubscription(null, SubscriptionName.Events).Wait();
+                RemoveSubscription(null, SubscriptionName.DataPoints).Wait();
+                RemoveSubscription(null, SubscriptionName.Audit).Wait();
+                RemoveSubscription(null, SubscriptionName.RebrowseTriggers).Wait();
                 Client.Browser.Transformations = null;
             }
             var ext = new UAExtractor(Config, Provider, pushers, Client, stateStore);
@@ -286,13 +286,14 @@ public virtual async Task DisposeAsync()
             }
         }
 
-        public async Task RemoveSubscription(SubscriptionName name)
+        public async Task RemoveSubscription(UAExtractor extractor, SubscriptionName name)
         {
             if (TryGetSubscription(name, out var subscription) && subscription!.Created)
             {
                 try
                 {
                     await Client.SessionManager.Session!.RemoveSubscriptionAsync(subscription);
+                    extractor?.RemoveKnownSubscription(name);
                 }
                 catch
                 {