[❄️flakes] Another round of flakes fixing (#5969)

* Another round of flakes fixing

* keep using watch()

* use proper dispatcher

* do not break delays

Author: martinbonnin

tests/integration-tests/src/commonTest/kotlin/test/FetchPolicyTest.kt

@@ -31,13 +31,13 @@ import com.apollographql.apollo3.integration.normalizer.CharacterNameByIdQuery
 import com.apollographql.apollo3.integration.normalizer.HeroNameQuery
 import com.apollographql.apollo3.interceptor.ApolloInterceptor
 import com.apollographql.apollo3.interceptor.ApolloInterceptorChain
+import com.apollographql.apollo3.testing.assertNoElement
+import com.apollographql.apollo3.testing.awaitElement
+import com.apollographql.apollo3.testing.internal.runTest
 import com.apollographql.mockserver.MockServer
 import com.apollographql.mockserver.awaitRequest
 import com.apollographql.mockserver.enqueueError
 import com.apollographql.mockserver.enqueueString
-import com.apollographql.apollo3.testing.assertNoElement
-import com.apollographql.apollo3.testing.awaitElement
-import com.apollographql.apollo3.testing.internal.runTest
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
@@ -66,7 +66,7 @@ class FetchPolicyTest {
     apolloClient = ApolloClient.Builder().serverUrl(mockServer.url()).store(store = store).build()
   }
 
-  private suspend fun tearDown() {
+  private fun tearDown() {
     mockServer.close()
     apolloClient.close()
   }

tests/integration-tests/src/commonTest/kotlin/test/WatcherErrorHandlingTest.kt

@@ -1,6 +1,7 @@
 package test
 
 import IdCacheKeyGenerator
+import app.cash.turbine.test
 import com.apollographql.apollo3.ApolloClient
 import com.apollographql.apollo3.api.ApolloResponse
 import com.apollographql.apollo3.cache.normalized.ApolloStore
@@ -14,12 +15,11 @@ import com.apollographql.apollo3.exception.ApolloHttpException
 import com.apollographql.apollo3.exception.CacheMissException
 import com.apollographql.apollo3.integration.normalizer.EpisodeHeroNameQuery
 import com.apollographql.apollo3.integration.normalizer.type.Episode
+import com.apollographql.apollo3.testing.awaitElement
+import com.apollographql.apollo3.testing.internal.runTest
 import com.apollographql.mockserver.MockServer
 import com.apollographql.mockserver.enqueueError
 import com.apollographql.mockserver.enqueueString
-import com.apollographql.apollo3.testing.awaitElement
-import com.apollographql.apollo3.testing.internal.runTest
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.first
@@ -41,66 +41,61 @@ class WatcherErrorHandlingTest {
     apolloClient = ApolloClient.Builder().serverUrl(mockServer.url()).store(store).build()
   }
 
-  private suspend fun tearDown() {
+  private fun tearDown() {
     mockServer.close()
   }
 
+  /**
+   * watch() should behave just like toFlow() in the absence of cache writes
+   */
   @Test
   fun fetchEmitsAllErrors() = runTest(before = { setUp() }, after = { tearDown() }) {
-    val channel = Channel<EpisodeHeroNameQuery.Data?>()
-    val jobs = mutableListOf<Job>()
-
-    jobs += launch {
-      mockServer.enqueueError(statusCode = 500)
-      apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
-          .fetchPolicy(FetchPolicy.CacheFirst)
-          .watch()
-          .collect {
-            channel.send(it.data)
-          }
-    }
-
-    jobs += launch {
-      apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
-          .fetchPolicy(FetchPolicy.CacheOnly)
-          .watch()
-          .collect {
-            channel.send(it.data)
-          }
-    }
-
-    jobs += launch {
-      mockServer.enqueueError(statusCode = 500)
-      apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
-          .fetchPolicy(FetchPolicy.NetworkFirst)
-          .watch()
-          .collect {
-            channel.send(it.data)
-          }
-    }
+    mockServer.enqueueError(statusCode = 500)
+    apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
+        .fetchPolicy(FetchPolicy.CacheFirst)
+        .watch()
+        .test {
+          assertIs<CacheMissException>(awaitItem().exception)
+          assertIs<ApolloHttpException>(awaitItem().exception)
+          cancelAndIgnoreRemainingEvents()
+        }
+
+    apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
+        .fetchPolicy(FetchPolicy.CacheOnly)
+        .watch()
+        .test {
+          assertIs<CacheMissException>(awaitItem().exception)
+          cancelAndIgnoreRemainingEvents()
+        }
 
-    jobs += launch {
-      mockServer.enqueueError(statusCode = 500)
-      apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
-          .fetchPolicy(FetchPolicy.NetworkOnly)
-          .watch()
-          .collect {
-            channel.send(it.data)
-          }
-    }
+    mockServer.enqueueError(statusCode = 500)
+    apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
+        .fetchPolicy(FetchPolicy.NetworkFirst)
+        .watch()
+        .test {
+          assertIs<ApolloHttpException>(awaitItem().exception)
+          assertIs<CacheMissException>(awaitItem().exception)
+          cancelAndIgnoreRemainingEvents()
+        }
 
-    jobs += launch {
-      mockServer.enqueueError(statusCode = 500)
-      apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
-          .fetchPolicy(FetchPolicy.CacheAndNetwork)
-          .watch()
-          .collect {
-            channel.send(it.data)
-          }
-    }
+    mockServer.enqueueError(statusCode = 500)
+    apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
+        .fetchPolicy(FetchPolicy.NetworkOnly)
+        .watch()
+        .test {
+          assertIs<ApolloHttpException>(awaitItem().exception)
+          cancelAndIgnoreRemainingEvents()
+        }
 
-    channel.assertCount(8)
-    jobs.forEach { it.cancel() }
+    mockServer.enqueueError(statusCode = 500)
+    apolloClient.query(EpisodeHeroNameQuery(Episode.EMPIRE))
+        .fetchPolicy(FetchPolicy.CacheAndNetwork)
+        .watch()
+        .test {
+          assertIs<CacheMissException>(awaitItem().exception)
+          assertIs<ApolloHttpException>(awaitItem().exception)
+          cancelAndIgnoreRemainingEvents()
+        }
   }
 
   @Test

tests/integration-tests/src/commonTest/kotlin/test/WatcherTest.kt

@@ -21,27 +21,30 @@ import com.apollographql.apollo3.integration.normalizer.EpisodeHeroNameWithIdQue
 import com.apollographql.apollo3.integration.normalizer.HeroAndFriendsNamesWithIDsQuery
 import com.apollographql.apollo3.integration.normalizer.StarshipByIdQuery
 import com.apollographql.apollo3.integration.normalizer.type.Episode
-import com.apollographql.mockserver.MockResponse
-import com.apollographql.mockserver.MockServer
-import com.apollographql.mockserver.enqueueString
 import com.apollographql.apollo3.testing.QueueTestNetworkTransport
-import com.apollographql.apollo3.testing.assertNoElement
-import com.apollographql.apollo3.testing.awaitElement
-import com.apollographql.apollo3.testing.enqueue
 import com.apollographql.apollo3.testing.enqueueTestNetworkError
 import com.apollographql.apollo3.testing.enqueueTestResponse
 import com.apollographql.apollo3.testing.internal.runTest
+import com.apollographql.mockserver.MockResponse
+import com.apollographql.mockserver.MockServer
+import com.apollographql.mockserver.enqueueString
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeout
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertIs
 import kotlin.test.assertNull
+import kotlin.time.Duration.Companion.minutes
 
 class WatcherTest {
   private lateinit var apolloClient: ApolloClient
@@ -72,14 +75,24 @@ class WatcherTest {
           HeroAndFriendsNamesWithIDsQuery.Friend("1003", "Leia Organa"),
       )))
 
+  @OptIn(ExperimentalCoroutinesApi::class)
+  private fun myRunTest(block: suspend CoroutineScope.() -> Unit) {
+    kotlinx.coroutines.test.runTest(timeout = 10.minutes) {
+      withContext(Dispatchers.Default.limitedParallelism(1)) {
+        block()
+      }
+    }
+  }
   /**
    * Executing the same query out of band should update the watcher
    *
    * Also, this test checks that the watcher gets control fast enough to subscribe to
    * cache changes
    */
   @Test
-  fun sameQueryTriggersWatcher() = runTest(before = { setUp() }) {
+  fun sameQueryTriggersWatcher() = myRunTest {
+    setUp()
+
     val query = EpisodeHeroNameQuery(Episode.EMPIRE)
     val channel = Channel<EpisodeHeroNameQuery.Data?>()
 
@@ -191,7 +204,7 @@ class WatcherTest {
     apolloClient.enqueueTestResponse(query, episodeHeroNameData)
     apolloClient.query(query).fetchPolicy(FetchPolicy.NetworkOnly).execute()
 
-    channel.assertNoElement()
+    channel.assertEmpty()
 
     job.cancel()
   }
@@ -256,7 +269,7 @@ class WatcherTest {
         .fetchPolicy(FetchPolicy.NetworkOnly)
         .execute()
 
-    channel.assertNoElement()
+    channel.assertEmpty()
 
     job.cancel()
   }
@@ -322,7 +335,7 @@ class WatcherTest {
     }
     job.cancelAndJoin()
 
-    channel.assertNoElement()
+    channel.assertEmpty()
   }
 
   /**
@@ -373,7 +386,7 @@ class WatcherTest {
 
     // Should see 1 cache miss values
     assertIs<CacheMissException>(channel.awaitElement().exception)
-    channel.assertNoElement()
+    channel.assertEmpty()
 
     job.cancel()
   }
@@ -578,7 +591,7 @@ class WatcherTest {
 
     // 2. Exception from the network (null data)
     assertNull(channel.awaitElement())
-    channel.assertNoElement()
+    channel.assertEmpty()
 
     // Another newer call updates the cache with "ArTwo"
     apolloClient.enqueueTestResponse(query, episodeHeroNameChangedTwoData)
@@ -624,9 +637,24 @@ class WatcherTest {
 
 }
 
-suspend fun <D> Channel<D>.assertCount(count: Int) {
+internal suspend fun <D> Channel<D>.assertCount(count: Int) {
   repeat(count) {
     awaitElement()
   }
-  assertNoElement()
+  assertEmpty()
+}
+
+internal suspend fun <T> Channel<T>.awaitElement(timeoutMillis: Long = 30000) = withTimeout(timeoutMillis) {
+  receive()
+}
+
+internal suspend fun <T> Channel<T>.assertEmpty(timeoutMillis: Long = 300): Unit {
+  try {
+    withTimeout(timeoutMillis) {
+      receive()
+    }
+    error("An item was unexpectedly received")
+  } catch (_: TimeoutCancellationException) {
+    // nothing
+  }
 }