Additional subscriber tests

android-test-common/src/main/java/com/ably/tracking/test/android/common/Faults.kt

@@ -32,7 +32,10 @@ abstract class FaultSimulation {
      * The advantage of this approach is that the test code remains active and continually compiled as
      * a first class citizen of the codebase, while we work on other things to get it passing.
      */
-    open val skipTest: Boolean = false
+    open val skipPublisherTest: Boolean = false
+
+    // Causes the same behaviour as skipPublisherTest for the subscriber NetworkConnectivityTests
+    open val skipSubscriberTest: Boolean = false
 
     /**
      * A RealtimeProxy instance that will be manipulated by this fault
@@ -108,7 +111,10 @@ class TcpConnectionRefused(apiKey: String) : TransportLayerFault(apiKey) {
      * - https://github.com/ably/ably-asset-tracking-android/issues/859
      * - https://github.com/ably/ably-asset-tracking-android/issues/871
      */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // May be able to be removed once the issues surrounding skipTest are resolved
+    override val skipSubscriberTest = true
 
     override fun enable() {
         tcpProxy.stop()
@@ -143,7 +149,10 @@ class TcpConnectionUnresponsive(apiKey: String) : TransportLayerFault(apiKey) {
      * - https://github.com/ably/ably-asset-tracking-android/issues/859
      * - https://github.com/ably/ably-asset-tracking-android/issues/871
      */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // May be able to be removed once the issues surrounding skipTest are resolved
+    override val skipSubscriberTest = true
 
     override fun enable() {
         tcpProxy.isForwarding = false
@@ -173,7 +182,10 @@ class DisconnectAndSuspend(apiKey: String) : TransportLayerFault(apiKey) {
     /*
         Currently failing due to Issues #871 and #907
     */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // May be able to be removed once the issues surrounding skipTest are resolved
+    override val skipSubscriberTest = true
 
     companion object {
         const val SUSPEND_DELAY_MILLIS: Long = 2 * 60 * 1000
@@ -305,7 +317,10 @@ class AttachUnresponsive(apiKey: String) : DropAction(
         Currently failing due to Issue #871 -- throwing ConnectionError
         when trying to add new trackables while offline.
      */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // Test appears to crash JVM
+    override val skipSubscriberTest = true
 
     override val name = "AttachUnresponsive"
 }
@@ -320,6 +335,8 @@ class DetachUnresponsive(apiKey: String) : DropAction(
     action = Message.Action.DETACH
 ) {
     override val name = "DetachUnresponsive"
+
+    override val skipSubscriberTest = true
 }
 
 /**
@@ -397,7 +414,10 @@ class EnterUnresponsive(apiKey: String) : UnresponsiveAfterAction(
         waiting for a presence response if there's there's a reconnection
         before successful completion of enter()
     */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // May be able to be removed once the issues surrounding skipTest are resolved
+    override val skipSubscriberTest = true
 
     override val name = "EnterUnresponsive"
 
@@ -429,7 +449,10 @@ class DisconnectWithFailedResume(apiKey: String) : ApplicationLayerFault(apiKey)
     /*
         Currently failing due to ably-java#474 presence bug
      */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // May be able to be removed once the issues surrounding skipTest are resolved
+    override val skipSubscriberTest = true
 
     /**
      * State of the fault, used to control whether we're intercepting
@@ -588,7 +611,10 @@ class EnterFailedWithNonfatalNack(apiKey: String) : PresenceNackFault(
         Currently failing due to Issue #907 - non-fatal nack triggers
         an exception to be thrown to caller during publisher.track()
      */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // Can probably be removed once skipTest issues are resolved
+    override val skipSubscriberTest = true
 
     override val name = "EnterFailedWithNonfatalNack"
 
@@ -609,6 +635,8 @@ class UpdateFailedWithNonfatalNack(apiKey: String) : PresenceNackFault(
 ) {
     override val name = "UpdateFailedWithNonfatalNack"
 
+    override val skipSubscriberTest = true
+
     override fun stateReceiverForStage(stage: FaultSimulationStage) =
         // Note: 5xx presence errors should always be non-fatal and recovered seamlessly
         TrackableStateReceiver.onlineWithoutFail("$name: $stage")
@@ -630,7 +658,10 @@ class ReenterOnResumeFailed(apiKey: String) : ApplicationLayerFault(apiKey) {
        This happens during stage 2 of the test, so steps 3 and 4 have not
        yet been seen to work.
      */
-    override val skipTest = true
+    override val skipPublisherTest = true
+
+    // May be able to be removed once the issues surrounding skipTest are resolved
+    override val skipSubscriberTest = true
 
     override val name = "ReenterOnResumeFailed"
 

publishing-sdk/src/androidTest/java/com/ably/tracking/publisher/NetworkConnectivityTests.kt

@@ -105,7 +105,7 @@ class NetworkConnectivityTests(private val testFault: FaultSimulation) {
 
     @Before
     fun setUp() {
-        Assume.assumeFalse(testFault.skipTest)
+        Assume.assumeFalse(testFault.skipPublisherTest)
 
         // We cannot use ktor on API Level 21 (Lollipop) because of:
         // https://youtrack.jetbrains.com/issue/KTOR-4751/HttpCache-plugin-uses-ConcurrentMap.computeIfAbsent-method-that-is-available-only-since-Android-API-24

subscribing-sdk/src/androidTest/java/com/ably/tracking/subscriber/NetworkConnectivityTests.kt

@@ -81,7 +81,7 @@ class NetworkConnectivityTests(private val testFault: FaultSimulation) {
 
     @Before
     fun setUp() {
-        Assume.assumeFalse(testFault.skipTest)
+        Assume.assumeFalse(testFault.skipSubscriberTest)
 
         // We cannot use ktor on API Level 21 (Lollipop) because of:
         // https://youtrack.jetbrains.com/issue/KTOR-4751/HttpCache-plugin-uses-ConcurrentMap.computeIfAbsent-method-that-is-available-only-since-Android-API-24
@@ -303,6 +303,128 @@ class NetworkConnectivityTests(private val testFault: FaultSimulation) {
         }
     }
 
+    /**
+     * Test that Subscriber can handle the given fault occurring whilst tracking,
+     *
+     * Check that after resolution, changes to the channels publisher resolutions
+     * are received by the subscriber.
+     */
+    @OptIn(Experimental::class)
+    @Test
+    fun faultWhilstTrackingPublisherResolutionUpdatesReceivedAfterResolution() {
+        withResources { resources ->
+            val subscriber = resources.getSubscriber()
+
+            // Begin listening for publisher state changes - the initial state if offline, so we have to ignore the first one
+            val initialResolutionExpectation = UnitExpectation("Initial resolution received")
+            var initialResolution: Resolution? = null
+            var receivedResolution: Resolution? = null
+            val updatedResolutionReceived = UnitExpectation("Updated resolution received")
+
+            subscriber.resolutions
+                .onEach { resolution ->
+                    if (initialResolution == null) {
+                        initialResolutionExpectation.fulfill()
+                        initialResolution = resolution
+                        return@onEach
+                    }
+
+                    receivedResolution = resolution
+                    updatedResolutionReceived.fulfill()
+                }
+                .launchIn(resources.scope)
+
+            // Join the ably channel to trigger a resolution update
+            val publishingConnection = resources.createAndStartPublishingAblyConnection()
+
+            // Check the initial resolution received
+            initialResolutionExpectation.await(10)
+            initialResolutionExpectation.assertFulfilled()
+            Assert.assertEquals(initialResolution, Resolution(Accuracy.BALANCED, 1L, 0.0))
+
+            // Start the fault and then send a new resolution
+            resources.fault.enable()
+            val newResolution = Resolution(Accuracy.MAXIMUM, 5L, 1.0)
+            runBlocking {
+                publishingConnection.updatePresenceData(
+                    resources.trackableId,
+                    PresenceData(ClientTypes.PUBLISHER, newResolution, false)
+                )
+            }
+
+            /**
+             * Resolve the fault and check that we then receive the new resolution.
+             * This has to be a long wait because some of the faults take many minutes to resolve.
+             */
+            resources.fault.resolve()
+            updatedResolutionReceived.await(600)
+            updatedResolutionReceived.assertFulfilled()
+
+            Assert.assertEquals(newResolution, receivedResolution)
+        }
+    }
+
+    /**
+     * Test that Subscriber sends resolution preference updates to the publisher
+     * after a fault is resolved.
+     */
+    @OptIn(Experimental::class)
+    @Test
+    fun faultWhilstUpdatingResolutionPreferenceUpdatesReceivedByPublisherAfterFaultResolution() {
+        withResources { resources ->
+            // Join the ably channel and listen for presence updates
+            val publishingConnection = resources.createAndStartPublishingAblyConnection()
+
+            val initialResolutionPreferenceExpectation = UnitExpectation("Initial resolution preference received")
+            var initialResolutionPreference: Resolution? = null
+            val receivedResolutionPreferenceExpectation = UnitExpectation("Updated resolution preference received")
+            var receivedResolutionPreference: Resolution? = null
+            runBlocking {
+                publishingConnection.subscribeForPresenceMessages(
+                    resources.trackableId,
+                    emitCurrentMessages = false,
+                    listener = { message ->
+                        if (initialResolutionPreference == null) {
+                            message.data.resolution?.let {
+                                initialResolutionPreference = message.data.resolution
+                                initialResolutionPreferenceExpectation.fulfill()
+                            }
+
+                            return@subscribeForPresenceMessages
+                        }
+
+                        message.data.resolution?.let {
+                            receivedResolutionPreference = message.data.resolution
+                            receivedResolutionPreferenceExpectation.fulfill()
+                        }
+                    }
+                )
+            }
+
+            // Start the subscriber and wait for the initial resolution preference to come through
+            val subscriber = resources.getSubscriber()
+            initialResolutionPreferenceExpectation.await(10)
+            initialResolutionPreferenceExpectation.assertFulfilled()
+            Assert.assertEquals(Resolution(Accuracy.BALANCED, 1L, 0.0), initialResolutionPreference)
+
+            // Start the fault and trigger the subscriber sending a new resolution
+            resources.fault.enable()
+            val newResolutionPreference = Resolution(Accuracy.MAXIMUM, 5L, 2.0)
+            runBlocking {
+                subscriber.resolutionPreference(newResolutionPreference)
+            }
+
+            /**
+             * Resolve the fault and check that we then receive the new resolution preference.
+             * This has to be a long wait because some of the faults take many minutes to resolve.
+             */
+            resources.fault.resolve()
+            receivedResolutionPreferenceExpectation.await(600)
+            receivedResolutionPreferenceExpectation.assertFulfilled()
+            Assert.assertEquals(newResolutionPreference, receivedResolutionPreference)
+        }
+    }
+
     /**
      * Checks that we have TestResources initialized and executes the test body
      */