Merge pull request #4568 from vector-im/feature/fga/fix_sdk_integration_tests

Feature/fga/fix sdk integration tests

Author: bmarty

changelog.d/4546.bugfix

@@ -0,0 +1 @@
+Fix lots of integration tests by introducing TestMatrix class and MatrixWorkerFactory.

matrix-sdk-android/src/androidTest/java/org/matrix/android/sdk/common/CommonTestHelper.kt

@@ -20,17 +20,17 @@ import android.content.Context
 import android.net.Uri
 import androidx.lifecycle.Observer
 import androidx.test.internal.runner.junit4.statement.UiThreadStatement
-import androidx.test.internal.runner.junit4.statement.UiThreadStatement.runOnUiThread
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNotNull
 import org.junit.Assert.assertTrue
-import org.matrix.android.sdk.api.Matrix
 import org.matrix.android.sdk.api.MatrixCallback
 import org.matrix.android.sdk.api.MatrixConfiguration
 import org.matrix.android.sdk.api.auth.data.HomeServerConnectionConfig
@@ -45,7 +45,7 @@ import org.matrix.android.sdk.api.session.room.timeline.Timeline
 import org.matrix.android.sdk.api.session.room.timeline.TimelineEvent
 import org.matrix.android.sdk.api.session.room.timeline.TimelineSettings
 import org.matrix.android.sdk.api.session.sync.SyncState
-import java.util.ArrayList
+import timber.log.Timber
 import java.util.UUID
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
@@ -56,21 +56,22 @@ import java.util.concurrent.TimeUnit
  */
 class CommonTestHelper(context: Context) {
 
-    val matrix: Matrix
+    internal val matrix: TestMatrix
+    val coroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Main)
 
-    fun getTestInterceptor(session: Session): MockOkHttpInterceptor? = TestNetworkModule.interceptorForSession(session.sessionId) as? MockOkHttpInterceptor
+    fun getTestInterceptor(session: Session): MockOkHttpInterceptor? = TestModule.interceptorForSession(session.sessionId) as? MockOkHttpInterceptor
 
     init {
         UiThreadStatement.runOnUiThread {
-            Matrix.initialize(
+            TestMatrix.initialize(
                     context,
                     MatrixConfiguration(
                             applicationFlavor = "TestFlavor",
                             roomDisplayNameFallbackProvider = TestRoomDisplayNameFallbackProvider()
                     )
             )
         }
-        matrix = Matrix.getInstance(context)
+        matrix = TestMatrix.getInstance(context)
     }
 
     fun createAccount(userNamePrefix: String, testParams: SessionTestParams): Session {
@@ -95,31 +96,45 @@ class CommonTestHelper(context: Context) {
      *
      * @param session    the session to sync
      */
-    @Suppress("EXPERIMENTAL_API_USAGE")
-    fun syncSession(session: Session, timeout: Long = TestConstants.timeOutMillis) {
+    fun syncSession(session: Session, timeout: Long = TestConstants.timeOutMillis * 10) {
         val lock = CountDownLatch(1)
-
-        val job = GlobalScope.launch(Dispatchers.Main) {
-            session.open()
+        coroutineScope.launch {
+            session.startSync(true)
+            val syncLiveData = session.getSyncStateLive()
+            val syncObserver = object : Observer<SyncState> {
+                override fun onChanged(t: SyncState?) {
+                    if (session.hasAlreadySynced()) {
+                        lock.countDown()
+                        syncLiveData.removeObserver(this)
+                    }
+                }
+            }
+            syncLiveData.observeForever(syncObserver)
         }
-        runBlocking { job.join() }
-
-        session.startSync(true)
+        await(lock, timeout)
+    }
 
-        val syncLiveData = runBlocking(Dispatchers.Main) {
-            session.getSyncStateLive()
-        }
-        val syncObserver = object : Observer<SyncState> {
-            override fun onChanged(t: SyncState?) {
-                if (session.hasAlreadySynced()) {
-                    lock.countDown()
-                    syncLiveData.removeObserver(this)
+    /**
+     * This methods clear the cache and waits for initialSync
+     *
+     * @param session    the session to sync
+     */
+    fun clearCacheAndSync(session: Session, timeout: Long = TestConstants.timeOutMillis) {
+        waitWithLatch(timeout) { latch ->
+            session.clearCache()
+            val syncLiveData = session.getSyncStateLive()
+            val syncObserver = object : Observer<SyncState> {
+                override fun onChanged(t: SyncState?) {
+                    if (session.hasAlreadySynced()) {
+                        Timber.v("Clear cache and synced")
+                        syncLiveData.removeObserver(this)
+                        latch.countDown()
+                    }
                 }
             }
+            syncLiveData.observeForever(syncObserver)
+            session.startSync(true)
         }
-        GlobalScope.launch(Dispatchers.Main) { syncLiveData.observeForever(syncObserver) }
-
-        await(lock, timeout)
     }
 
     /**
@@ -130,46 +145,57 @@ class CommonTestHelper(context: Context) {
      * @param nbOfMessages the number of time the message will be sent
      */
     fun sendTextMessage(room: Room, message: String, nbOfMessages: Int, timeout: Long = TestConstants.timeOutMillis): List<TimelineEvent> {
-        val timeline = room.createTimeline(null, TimelineSettings(10))
         val sentEvents = ArrayList<TimelineEvent>(nbOfMessages)
-        val latch = CountDownLatch(1)
-        val timelineListener = object : Timeline.Listener {
-            override fun onTimelineFailure(throwable: Throwable) {
-            }
+        val timeline = room.createTimeline(null, TimelineSettings(10))
+        timeline.start()
+        waitWithLatch(timeout + 1_000L * nbOfMessages) { latch ->
+            val timelineListener = object : Timeline.Listener {
+                override fun onTimelineFailure(throwable: Throwable) {
+                }
 
-            override fun onNewTimelineEvents(eventIds: List<String>) {
-                // noop
-            }
+                override fun onNewTimelineEvents(eventIds: List<String>) {
+                    // noop
+                }
 
-            override fun onTimelineUpdated(snapshot: List<TimelineEvent>) {
-                val newMessages = snapshot
-                        .filter { it.root.sendState == SendState.SYNCED }
-                        .filter { it.root.getClearType() == EventType.MESSAGE }
-                        .filter { it.root.getClearContent().toModel<MessageContent>()?.body?.startsWith(message) == true }
-
-                if (newMessages.size == nbOfMessages) {
-                    sentEvents.addAll(newMessages)
-                    // Remove listener now, if not at the next update sendEvents could change
-                    timeline.removeListener(this)
-                    latch.countDown()
+                override fun onTimelineUpdated(snapshot: List<TimelineEvent>) {
+                    val newMessages = snapshot
+                            .filter { it.root.sendState == SendState.SYNCED }
+                            .filter { it.root.getClearType() == EventType.MESSAGE }
+                            .filter { it.root.getClearContent().toModel<MessageContent>()?.body?.startsWith(message) == true }
+
+                    Timber.v("New synced message size: ${newMessages.size}")
+                    if (newMessages.size == nbOfMessages) {
+                        sentEvents.addAll(newMessages)
+                        // Remove listener now, if not at the next update sendEvents could change
+                        timeline.removeListener(this)
+                        latch.countDown()
+                    }
                 }
             }
+            timeline.addListener(timelineListener)
+            sendTextMessagesBatched(room, message, nbOfMessages)
         }
-        timeline.start()
-        timeline.addListener(timelineListener)
-        for (i in 0 until nbOfMessages) {
-            room.sendTextMessage(message + " #" + (i + 1))
-        }
-        // Wait 3 second more per message
-        await(latch, timeout = timeout + 3_000L * nbOfMessages)
         timeline.dispose()
-
         // Check that all events has been created
         assertEquals("Message number do not match $sentEvents", nbOfMessages.toLong(), sentEvents.size.toLong())
-
         return sentEvents
     }
 
+    /**
+     * Will send nb of messages provided by count parameter but waits a bit every 10 messages to avoid gap in sync
+     */
+    private fun sendTextMessagesBatched(room: Room, message: String, count: Int) {
+        (1 until count + 1)
+                .map { "$message #$it" }
+                .chunked(10)
+                .forEach { batchedMessages ->
+                    batchedMessages.forEach { formattedMessage ->
+                        room.sendTextMessage(formattedMessage)
+                    }
+                    Thread.sleep(1_000L)
+                }
+    }
+
     // PRIVATE METHODS *****************************************************************************
 
     /**
@@ -239,10 +265,10 @@ class CommonTestHelper(context: Context) {
 
         assertTrue(registrationResult is RegistrationResult.Success)
         val session = (registrationResult as RegistrationResult.Success).session
+        session.open()
         if (sessionTestParams.withInitialSync) {
             syncSession(session, 60_000)
         }
-
         return session
     }
 
@@ -267,7 +293,7 @@ class CommonTestHelper(context: Context) {
                     .getLoginWizard()
                     .login(userName, password, "myDevice")
         }
-
+        session.open()
         if (sessionTestParams.withInitialSync) {
             syncSession(session)
         }
@@ -332,22 +358,21 @@ class CommonTestHelper(context: Context) {
         assertTrue(latch.await(timeout ?: TestConstants.timeOutMillis, TimeUnit.MILLISECONDS))
     }
 
-    @Suppress("EXPERIMENTAL_API_USAGE")
-    fun retryPeriodicallyWithLatch(latch: CountDownLatch, condition: (() -> Boolean)) {
-        GlobalScope.launch {
-            while (true) {
-                delay(1000)
-                if (condition()) {
-                    latch.countDown()
-                    return@launch
-                }
+    suspend fun retryPeriodicallyWithLatch(latch: CountDownLatch, condition: (() -> Boolean)) {
+        while (true) {
+            delay(1000)
+            if (condition()) {
+                latch.countDown()
+                return
             }
         }
     }
 
-    fun waitWithLatch(timeout: Long? = TestConstants.timeOutMillis, block: (CountDownLatch) -> Unit) {
+    fun waitWithLatch(timeout: Long? = TestConstants.timeOutMillis, dispatcher: CoroutineDispatcher = Dispatchers.Main, block: suspend (CountDownLatch) -> Unit) {
         val latch = CountDownLatch(1)
-        block(latch)
+        coroutineScope.launch(dispatcher) {
+            block(latch)
+        }
         await(latch, timeout)
     }