Fix flake on ReaperTest#testRemoveWatchWhenCloudRemoved (#1516)

* Fix Flake on ReaperTest#testRemoveWatchWhenCloudRemoved

Failure reproduced with

```
Index: src/main/java/org/csanchez/jenkins/plugins/kubernetes/pod/retention/Reaper.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/main/java/org/csanchez/jenkins/plugins/kubernetes/pod/retention/Reaper.java b/src/main/java/org/csanchez/jenkins/plugins/kubernetes/pod/retention/Reaper.java
--- a/src/main/java/org/csanchez/jenkins/plugins/kubernetes/pod/retention/Reaper.java	(revision a81cd9f)
+++ b/src/main/java/org/csanchez/jenkins/plugins/kubernetes/pod/retention/Reaper.java	(date 1709649329339)
@@ -198,6 +198,11 @@

             // close any cloud watchers that have been removed
             cloudNames.stream().map(this.watchers::get).filter(Objects::nonNull).forEach(cpw -> {
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
                 LOGGER.info(() -> "stopping pod watcher for deleted kubernetes cloud " + cpw.cloudName);
                 cpw.stop();
             });
```

The cause is that watcher really gets removed in a separate thread, so the test must await and avoid checking immediately.

* Use structured assertion everywhere

Also happened to find a problem with org.csanchez.jenkins.plugins.kubernetes.pod.retention.ReaperTest.testKeepWatchingOnStatusWatchEvent so fixed it.

src/test/java/org/csanchez/jenkins/plugins/kubernetes/pod/retention/ReaperTest.java

@@ -97,7 +97,7 @@ public void testMaybeActivate() throws IOException, InterruptedException {
         assertEquals("node removed from jenkins", j.jenkins.getNodes().size(), 0);
 
         // watch was created
-        assertTrue(r.isWatchingCloud(cloud.name));
+        assertShouldBeWatching(r, cloud);
 
         kubeClientRequests()
                 // expect pod not running to be removed
@@ -129,7 +129,7 @@ public void testWatchFailOnActivate() throws IOException, InterruptedException {
         kubeClientRequests()
                 .assertRequestCountAtLeast("/api/v1/namespaces/foo/pods?allowWatchBookmarks=true&watch=true", 1);
         // watch failed to register
-        assertFalse(r.isWatchingCloud(cloud.name));
+        assertShouldNotBeWatching(r, cloud);
     }
 
     @Test
@@ -151,15 +151,13 @@ public void testActivateOnNewComputer() throws IOException, InterruptedException
         KubernetesComputer kc = new KubernetesComputer(n2);
 
         // should not be watching the newly created cloud at this point
-        assertFalse("should not be watching cloud", r.isWatchingCloud(cloud.name));
+        assertShouldNotBeWatching(r, cloud);
 
         // fire compute on-line event
         r.preLaunch(kc, tl);
 
         // expect new cloud registered
-        while (!r.isWatchingCloud(cloud.name)) {
-            Thread.sleep(100);
-        }
+        assertShouldBeWatching(r, cloud);
         kubeClientRequests()
                 .assertRequestCountAtLeast("/api/v1/namespaces/foo/pods?allowWatchBookmarks=true&watch=true", 1);
     }
@@ -202,9 +200,7 @@ public void testReconnectOnNewComputer() throws InterruptedException, IOExceptio
 
         // wait until watch is removed
         System.out.println("Waiting for watch to be removed");
-        while (r.isWatchingCloud(cloud.name)) {
-            Thread.sleep(250L);
-        }
+        assertShouldNotBeWatching(r, cloud);
         System.out.println("Watch removed");
 
         // launch computer
@@ -215,9 +211,7 @@ public void testReconnectOnNewComputer() throws InterruptedException, IOExceptio
 
         // should have started new watch
         System.out.println("Waiting for a new watch to be started");
-        while (!r.isWatchingCloud(cloud.name)) {
-            Thread.sleep(100);
-        }
+        assertShouldBeWatching(r, cloud);
         System.out.println("Watch started");
     }
 
@@ -244,7 +238,7 @@ public void testAddWatchWhenCloudAdded() throws InterruptedException, IOExceptio
         j.jenkins.clouds.add(cloud);
 
         // watch is added
-        assertTrue("should be watching cloud", r.isWatchingCloud(cloud.name));
+        assertShouldBeWatching(r, cloud);
         kubeClientRequests().assertRequestCountAtLeast(watchPodsPath, 1);
     }
 
@@ -264,13 +258,15 @@ public void testRemoveWatchWhenCloudRemoved() throws InterruptedException, IOExc
         r.maybeActivate();
 
         // should not be watching the newly created cloud at this point
-        assertTrue("should be watching cloud", r.isWatchingCloud(cloud.name));
+        assertShouldBeWatching(r, cloud);
 
         // invalidate client
         j.jenkins.clouds.remove(cloud);
 
         // watch is removed
-        assertFalse("should not be watching cloud", r.isWatchingCloud(cloud.name));
+        // org.csanchez.jenkins.plugins.kubernetes.pod.retention.Reaper.CloudPodWatcher.onClose() is called in a
+        // separate thread
+        assertShouldNotBeWatching(r, cloud);
     }
 
     @Test(timeout = 10_000)
@@ -313,7 +309,7 @@ public void testReplaceWatchWhenCloudUpdated() throws InterruptedException, IOEx
         r.maybeActivate();
 
         // should not be watching the newly created cloud at this point
-        assertTrue("should be watching cloud", r.isWatchingCloud(cloud.name));
+        assertShouldBeWatching(r, cloud);
 
         // invalidate client
         cloud.setNamespace("bar");
@@ -322,7 +318,7 @@ public void testReplaceWatchWhenCloudUpdated() throws InterruptedException, IOEx
         KubernetesSlave node = addNode(cloud, "node-123", "node");
 
         // watch is still active
-        assertTrue("should be watching cloud", r.isWatchingCloud(cloud.name));
+        assertShouldBeWatching(r, cloud);
 
         listener.waitForEvents().expectEvent(Watcher.Action.MODIFIED, node);
         kubeClientRequests().assertRequestCountAtLeast(watchBarPodsPath, 1);
@@ -358,10 +354,7 @@ public void testStopWatchingOnCloseException() throws InterruptedException {
         listener.expectNoEvents();
 
         // watch is removed
-        while (r.isWatchingCloud(cloud.name)) {
-            Thread.sleep(250L);
-        }
-        assertFalse(r.isWatchingCloud(cloud.name));
+        assertShouldNotBeWatching(r, cloud);
     }
 
     @Test(timeout = 10_000)
@@ -402,7 +395,7 @@ public void testKeepWatchingOnKubernetesApiServerError() throws InterruptedExcep
         listener.expectNoEvents();
 
         // watch is still active
-        assertTrue(r.isWatchingCloud(cloud.name));
+        assertShouldBeWatching(r, cloud);
     }
 
     @Test(timeout = 10_000)
@@ -416,7 +409,11 @@ public void testKeepWatchingOnStatusWatchEvent() throws InterruptedException {
         status.setCode(200);
         server.expect()
                 .withPath(watchPodsPath)
-                .andReturnChunked(200, new WatchEvent(status, "ERROR"))
+                .andUpgradeToWebSocket()
+                .open()
+                .immediately()
+                .andEmit(new WatchEvent(status, "ERROR"))
+                .done()
                 .once();
 
         // activate reaper
@@ -429,7 +426,7 @@ public void testKeepWatchingOnStatusWatchEvent() throws InterruptedException {
         listener.expectNoEvents();
 
         // watch is still active
-        assertTrue(r.isWatchingCloud(cloud.name));
+        assertShouldBeWatching(r, cloud);
     }
 
     @Test
@@ -453,17 +450,13 @@ public void testCloseWatchersOnShutdown() throws InterruptedException {
         r.maybeActivate();
 
         // watching cloud
-        assertTrue(r.isWatchingCloud(cloud.name));
-        assertTrue(r.isWatchingCloud(cloud2.name));
-        assertTrue(r.isWatchingCloud(cloud3.name));
+        assertShouldBeWatching(r, cloud, cloud2, cloud3);
 
         // trigger shutdown listener
         new Reaper.ReaperShutdownListener().onBeforeShutdown();
 
         // watchers removed
-        await().until(() -> r.isWatchingCloud(cloud.name), is(false));
-        await().until(() -> r.isWatchingCloud(cloud2.name), is(false));
-        await().until(() -> r.isWatchingCloud(cloud3.name), is(false));
+        assertShouldNotBeWatching(r, cloud, cloud2, cloud3);
     }
 
     @Test(timeout = 10_000)
@@ -895,4 +888,16 @@ private static WatchEvent errorEvent() {
                 .endStatusObject()
                 .build();
     }
+
+    private static void assertShouldBeWatching(Reaper r, KubernetesCloud... clouds) {
+        for (KubernetesCloud cloud : clouds) {
+            await("should be watching cloud " + cloud.name).until(() -> r.isWatchingCloud(cloud.name));
+        }
+    }
+
+    private static void assertShouldNotBeWatching(Reaper r, KubernetesCloud... clouds) {
+        for (KubernetesCloud cloud : clouds) {
+            await("should not be watching cloud " + cloud.name).until(() -> !r.isWatchingCloud(cloud.name));
+        }
+    }
 }