Add IncomingMessageObserver integration test infrastructure.

2026-06-29 10:35:52 +01:00 · 2026-05-20 10:44:40 -03:00
parent 71ffc36e7f
commit 49f0c2502b
15 changed files with 617 additions and 143 deletions
@@ -0,0 +1,36 @@
+package org.thoughtcrime.securesms
+
+import org.signal.core.util.logging.AndroidLogger
+import org.signal.core.util.logging.Log
+import org.signal.libsignal.protocol.logging.SignalProtocolLoggerProvider
+import org.thoughtcrime.securesms.dependencies.AppDependencies
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencyProvider
+import org.thoughtcrime.securesms.logging.CustomSignalProtocolLogger
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverDependencyProvider
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverTestRunner
+
+/**
+ * Application used when running `IncomingMessageObserver` instrumentation tests. Installs
+ * [IncomingMessageObserverDependencyProvider] so the websocket and job manager are replaced
+ * with test-friendly implementations. Selected by [IncomingMessageObserverTestRunner] when
+ * gradle is invoked with `-PimoTests`.
+ */
+class IncomingMessageObserverInstrumentationApplicationContext : ApplicationContext() {
+
+  override fun initializeAppDependencies() {
+    val default = ApplicationDependencyProvider(this)
+    AppDependencies.init(this, IncomingMessageObserverDependencyProvider(this, default))
+    AppDependencies.deadlockDetector.start()
+  }
+
+  override fun initializeLogging() {
+    Log.initialize({ true }, AndroidLogger)
+    SignalProtocolLoggerProvider.setProvider(CustomSignalProtocolLogger())
+  }
+
+  override fun beginJobLoop() = Unit
+
+  fun beginJobLoopForTests() {
+    super.beginJobLoop()
+  }
+}
@@ -0,0 +1,27 @@
+package org.thoughtcrime.securesms.messages.incomingmessageobserver
+
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverAssertions.assertMessageReceived
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverAssertions.assertNoMessageReceived
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverRule
+
+@RunWith(AndroidJUnit4::class)
+class DecryptionErrorTest {
+
+  @get:Rule
+  val rule = IncomingMessageObserverRule(peerCount = 2)
+
+  @Test
+  fun malformedEnvelope_dropsMessage_butPipelineRecovers() {
+    val peer = rule.peers[0]
+
+    rule.deliver { malformedEnvelope() from peer }
+    assertNoMessageReceived(from = peer, body = "subsequent")
+
+    rule.deliver { text("subsequent") from peer }
+    assertMessageReceived(from = peer, body = "subsequent")
+  }
+}
@@ -0,0 +1,39 @@
+package org.thoughtcrime.securesms.messages.incomingmessageobserver
+
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverAssertions.assertGroupMessageReceived
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverRule
+
+@RunWith(AndroidJUnit4::class)
+class IncomingGroupMessageTest {
+
+  @get:Rule
+  val rule = IncomingMessageObserverRule(peerCount = 5)
+
+  @Test
+  fun deliveredGroupText_isPersistedInGroupThread() {
+    val group = rule.testGroup
+
+    rule.deliver { groupText("hello group", group = group) from rule.peers[0] }
+
+    assertGroupMessageReceived(from = rule.peers[0], group = group, body = "hello group")
+  }
+
+  @Test
+  fun multipleGroupMembers_messagesPersistedFromEach() {
+    val group = rule.testGroup
+
+    rule.deliver {
+      groupText("from peer 0", group = group) from rule.peers[0]
+      groupText("from peer 1", group = group) from rule.peers[1]
+      groupText("from peer 2", group = group) from rule.peers[2]
+    }
+
+    assertGroupMessageReceived(from = rule.peers[0], group = group, body = "from peer 0")
+    assertGroupMessageReceived(from = rule.peers[1], group = group, body = "from peer 1")
+    assertGroupMessageReceived(from = rule.peers[2], group = group, body = "from peer 2")
+  }
+}
@@ -0,0 +1,22 @@
+package org.thoughtcrime.securesms.messages.incomingmessageobserver
+
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverAssertions.assertMessageReceived
+import org.thoughtcrime.securesms.testing.incomingmessageobserver.IncomingMessageObserverRule
+
+@RunWith(AndroidJUnit4::class)
+class IncomingTextMessageTest {
+
+  @get:Rule
+  val rule = IncomingMessageObserverRule(peerCount = 2)
+
+  @Test
+  fun deliveredOneToOneText_isPersisted() {
+    rule.deliver { text("hello world") from rule.peers[0] }
+
+    assertMessageReceived(from = rule.peers[0], body = "hello world")
+  }
+}
@@ -0,0 +1,71 @@
+package org.thoughtcrime.securesms.testing.incomingmessageobserver
+
+import assertk.assertThat
+import assertk.assertions.isEqualTo
+import assertk.assertions.isNotNull
+import assertk.assertions.isTrue
+import org.signal.benchmark.setup.OtherClient
+import org.thoughtcrime.securesms.database.MessageTable
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.MessageRecord
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.whispersystems.signalservice.api.push.SignalServiceAddress
+
+/**
+ * Reads database state produced by [IncomingMessageObserverRule]-driven tests. Import members
+ * individually (e.g. `import …IncomingMessageObserverAssertions.assertMessageReceived`) so test
+ * bodies stay terse.
+ */
+object IncomingMessageObserverAssertions {
+
+  fun OtherClient.recipientId(): RecipientId = Recipient.externalPush(SignalServiceAddress(serviceId, e164)).id
+
+  fun findIncomingMessage(from: OtherClient, body: String): MessageRecord? {
+    val threadId = SignalDatabase.threads.getThreadIdFor(from.recipientId()) ?: return null
+    return SignalDatabase.messages.getConversation(threadId).use { cursor ->
+      MessageTable.MmsReader(cursor).use { reader -> reader.firstOrNull { it.body == body } }
+    }
+  }
+
+  fun findIncomingGroupMessage(from: OtherClient, group: GroupHandle, body: String): MessageRecord? {
+    val threadId = SignalDatabase.threads.getThreadIdFor(group.recipientId) ?: return null
+    return SignalDatabase.messages.getConversation(threadId).use { cursor ->
+      MessageTable.MmsReader(cursor).use { reader ->
+        reader.firstOrNull { it.body == body && it.fromRecipient.id == from.recipientId() }
+      }
+    }
+  }
+
+  fun assertMessageReceived(from: OtherClient, body: String) {
+    val record = findIncomingMessage(from, body)
+    assertThat(record, "incoming message with body \"$body\" from ${from.serviceId} not found").isNotNull()
+    assertThat(record!!.fromRecipient.id, "incoming message sender mismatch for body \"$body\"").isEqualTo(from.recipientId())
+  }
+
+  fun assertGroupMessageReceived(from: OtherClient, group: GroupHandle, body: String) {
+    val record = findIncomingGroupMessage(from, group, body)
+    assertThat(record, "group message \"$body\" from ${from.serviceId} in ${group.groupId} not found").isNotNull()
+  }
+
+  fun assertNoMessageReceived(from: OtherClient, body: String) {
+    val record = findIncomingMessage(from, body)
+    assertThat(record == null, "expected no message with body \"$body\" from ${from.serviceId}, but found one").isTrue()
+  }
+
+  fun assertNoMessagesInThread(recipientId: RecipientId) {
+    val threadId = SignalDatabase.threads.getThreadIdFor(recipientId) ?: return
+    val count = SignalDatabase.messages.getConversation(threadId).use { cursor -> cursor.count }
+    assertThat(count, "expected thread for $recipientId to be empty, but message count was").isEqualTo(0)
+  }
+
+  fun assertDeliveryReceipt(outgoingMessageId: Long) {
+    val record = SignalDatabase.messages.getMessageRecord(outgoingMessageId)
+    assertThat(record.hasDeliveryReceipt(), "expected delivery receipt on outgoing message $outgoingMessageId, but none recorded").isTrue()
+  }
+
+  fun assertReadReceipt(outgoingMessageId: Long) {
+    val record = SignalDatabase.messages.getMessageRecord(outgoingMessageId)
+    assertThat(record.hasReadReceipt(), "expected read receipt on outgoing message $outgoingMessageId, but none recorded").isTrue()
+  }
+}
@@ -0,0 +1,63 @@
+package org.thoughtcrime.securesms.testing.incomingmessageobserver
+
+import android.app.Application
+import org.signal.benchmark.setup.NoOpJob
+import org.signal.libsignal.net.Network
+import org.thoughtcrime.securesms.dependencies.AppDependencies
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencyProvider
+import org.thoughtcrime.securesms.dependencies.InstrumentationApplicationDependencyProvider
+import org.thoughtcrime.securesms.jobmanager.JobManager
+import org.thoughtcrime.securesms.jobs.JobManagerFactories
+import org.whispersystems.signalservice.api.util.UptimeSleepTimer
+import org.whispersystems.signalservice.api.websocket.SignalWebSocket
+import org.whispersystems.signalservice.internal.configuration.SignalServiceConfiguration
+import org.whispersystems.signalservice.internal.websocket.BenchmarkWebSocketConnection
+import java.util.function.Supplier
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * Dependency provider used by [org.thoughtcrime.securesms.IncomingMessageObserverInstrumentationApplicationContext].
+ * Composes [InstrumentationApplicationDependencyProvider] (so existing mocks for the account /
+ * archive / donations / billing APIs are reused) and overrides:
+ *
+ * - the auth and unauth websocket factories with [BenchmarkWebSocketConnection], so tests can
+ *   inject encrypted envelopes through the real ingest pipeline;
+ * - the job manager, swapping the startup network jobs handled by [NoOpJob.replaceFactories]
+ *   to no-ops so they can't fire against unstubbed mocks during a test.
+ */
+class IncomingMessageObserverDependencyProvider(
+  private val application: Application,
+  default: ApplicationDependencyProvider
+) : AppDependencies.Provider by InstrumentationApplicationDependencyProvider(application, default) {
+
+  override fun provideAuthWebSocket(
+    signalServiceConfigurationSupplier: Supplier<SignalServiceConfiguration>,
+    libSignalNetworkSupplier: Supplier<Network>
+  ): SignalWebSocket.AuthenticatedWebSocket {
+    return SignalWebSocket.AuthenticatedWebSocket(
+      connectionFactory = { BenchmarkWebSocketConnection.createAuthInstance() },
+      canConnect = { true },
+      sleepTimer = UptimeSleepTimer(),
+      disconnectTimeoutMs = 15.seconds.inWholeMilliseconds
+    )
+  }
+
+  override fun provideUnauthWebSocket(
+    signalServiceConfigurationSupplier: Supplier<SignalServiceConfiguration>,
+    libSignalNetworkSupplier: Supplier<Network>
+  ): SignalWebSocket.UnauthenticatedWebSocket {
+    return SignalWebSocket.UnauthenticatedWebSocket(
+      connectionFactory = { BenchmarkWebSocketConnection.createUnauthInstance() },
+      canConnect = { true },
+      sleepTimer = UptimeSleepTimer(),
+      disconnectTimeoutMs = 15.seconds.inWholeMilliseconds
+    )
+  }
+
+  override fun provideJobManager(configurationBuilder: JobManager.Configuration.Builder): JobManager {
+    val config = configurationBuilder
+      .setJobFactories(NoOpJob.replaceFactories(JobManagerFactories.getJobFactories(application)))
+      .build()
+    return JobManager(application, config)
+  }
+}
@@ -0,0 +1,201 @@
+package org.thoughtcrime.securesms.testing.incomingmessageobserver
+
+import okio.ByteString.Companion.toByteString
+import org.junit.Assume
+import org.junit.rules.ExternalResource
+import org.signal.benchmark.setup.Generator
+import org.signal.benchmark.setup.Harness
+import org.signal.benchmark.setup.OtherClient
+import org.signal.benchmark.setup.TestUsers
+import org.signal.core.util.logging.Log
+import org.signal.network.websocket.WebSocketRequestMessage
+import org.thoughtcrime.securesms.IncomingMessageObserverInstrumentationApplicationContext
+import org.thoughtcrime.securesms.dependencies.AppDependencies
+import org.thoughtcrime.securesms.groups.GroupId
+import org.thoughtcrime.securesms.jobmanager.Job
+import org.thoughtcrime.securesms.jobmanager.JobTracker
+import org.thoughtcrime.securesms.jobs.MarkerJob
+import org.thoughtcrime.securesms.jobs.PushProcessMessageJob
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.whispersystems.signalservice.internal.push.Envelope
+import org.whispersystems.signalservice.internal.websocket.BenchmarkWebSocketConnection
+import java.util.concurrent.CopyOnWriteArraySet
+import kotlin.random.Random
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * JUnit rule that drives [org.thoughtcrime.securesms.messages.IncomingMessageObserver] from
+ * instrumentation tests. Sets up self, registers [peerCount] simulated peers from
+ * [Harness.otherClients], establishes a Signal double-ratchet session with each, and exposes a
+ * small DSL for delivering encrypted envelopes through the real ingest pipeline:
+ *
+ * ```
+ * @get:Rule val rule = IncomingMessageObserverRule(peerCount = 2)
+ *
+ * @Test fun example() {
+ *   rule.deliver { text("hi") from rule.peers[0] }
+ *   rule.deliver { groupText("hi all", group = rule.testGroup) from rule.peers[0] }
+ * }
+ * ```
+ *
+ * Run with `-PimoTests`; tests are skipped under the default runner. Throws on drain timeout.
+ * Mutually exclusive with `SignalDatabaseRule` / `SignalActivityRule` — all three claim the
+ * local identity.
+ */
+class IncomingMessageObserverRule(
+  private val peerCount: Int = 2,
+  private val drainTimeout: Duration = 30.seconds
+) : ExternalResource() {
+
+  lateinit var self: Recipient
+    private set
+
+  lateinit var peers: List<OtherClient>
+    private set
+
+  /** Lazily-created group. Touching this from a test triggers setup; tests that don't use groups pay nothing. */
+  val testGroup: GroupHandle by lazy {
+    val gid = TestUsers.setupGroup(withLabels = false)
+    GroupHandle(gid, Recipient.externalGroupExact(gid).id)
+  }
+
+  override fun before() {
+    Assume.assumeTrue(
+      "IncomingMessageObserverRule requires the IMO test runner — run with -PimoTests",
+      AppDependencies.application is IncomingMessageObserverInstrumentationApplicationContext
+    )
+
+    self = TestUsers.setupSelf()
+    TestUsers.setupTestClients(peerCount)
+    peers = Harness.otherClients.take(peerCount)
+
+    val app = AppDependencies.application as IncomingMessageObserverInstrumentationApplicationContext
+    app.beginJobLoopForTests()
+
+    // IncomingMessageObserver caches `canProcessMessages` from restoreDecisionState at thread
+    // construction. If it was built before setupSelf() flipped the state it will silently drop
+    // every message; reset network so a fresh observer is constructed.
+    AppDependencies.incomingMessageObserver.notifyRestoreDecisionMade()
+    AppDependencies.startNetwork()
+    forceObserverConstruction()
+
+    val handshakeEnvelopes = peers.map { client ->
+      client.encrypt(Generator.encryptedTextMessage(System.currentTimeMillis()))
+    }
+    deliverEnvelopes(handshakeEnvelopes)
+    peers.forEach { it.completeSession() }
+  }
+
+  fun deliver(builder: DeliveryBuilder.() -> Unit) {
+    val collected = DeliveryBuilder().apply(builder).specs
+    if (collected.isEmpty()) return
+    deliverEnvelopes(collected.map { it.materialize() })
+  }
+
+  private fun forceObserverConstruction() {
+    AppDependencies.incomingMessageObserver
+  }
+
+  private fun deliverEnvelopes(envelopes: List<Envelope>) {
+    val jobManager = AppDependencies.jobManager
+    val seenQueues = CopyOnWriteArraySet<String>()
+    val queueListener = object : JobTracker.JobListener {
+      override fun onStateChanged(job: Job, jobState: JobTracker.JobState) {
+        job.parameters.queue?.let { queue ->
+          if (queue.startsWith(PushProcessMessageJob.QUEUE_PREFIX)) {
+            seenQueues += queue
+          }
+        }
+      }
+    }
+    jobManager.addListener({ job: Job -> job.parameters.queue?.startsWith(PushProcessMessageJob.QUEUE_PREFIX) == true }, queueListener)
+
+    try {
+      BenchmarkWebSocketConnection.addPendingMessages(envelopes.map { it.toWebSocketPayload() })
+      BenchmarkWebSocketConnection.addQueueEmptyMessage()
+      BenchmarkWebSocketConnection.releaseMessages()
+
+      val consumed = BenchmarkWebSocketConnection.awaitAllMessagesConsumed(drainTimeout.inWholeMilliseconds)
+      check(consumed) { "Timed out waiting for benchmark websocket to consume ${envelopes.size} envelope(s)" }
+
+      // PushProcessMessageJob enqueue happens on a background thread after the websocket marks
+      // messages consumed; this tick lets that settle before we snapshot the queues to wait on.
+      Thread.sleep(100)
+
+      val queuesToDrain = seenQueues.toSet()
+      Log.d(TAG, "Awaiting ${queuesToDrain.size} PushProcessMessageJob queue(s): $queuesToDrain")
+      for (queue in queuesToDrain) {
+        val state = jobManager.runSynchronously(MarkerJob(queue), drainTimeout.inWholeMilliseconds)
+        check(state.isPresent) { "Timed out waiting for queue $queue to drain" }
+      }
+    } finally {
+      jobManager.removeListener(queueListener)
+    }
+  }
+
+  companion object {
+    private val TAG = Log.tag(IncomingMessageObserverRule::class)
+
+    private fun Envelope.toWebSocketPayload(): WebSocketRequestMessage = WebSocketRequestMessage(
+      verb = "PUT",
+      path = "/api/v1/message",
+      id = Random.nextLong(),
+      headers = listOf("X-Signal-Timestamp: $serverTimestamp"),
+      body = encodeByteString()
+    )
+  }
+}
+
+/** Identifies the test group created by [IncomingMessageObserverRule]. Hold a reference to pass into the [DeliveryBuilder.groupText] DSL. */
+data class GroupHandle(val groupId: GroupId.V2, val recipientId: RecipientId)
+
+/**
+ * Receiver of the DSL passed to [IncomingMessageObserverRule.deliver]. Construct content with
+ * [text] / [groupText] / [deliveryReceipts] / [readReceipts] / [malformedEnvelope] and chain
+ * with the [from] infix to attach a sending peer. Each `from` adds the resulting envelope to
+ * the batch that will be delivered when the lambda returns.
+ */
+class DeliveryBuilder internal constructor() {
+  internal val specs = mutableListOf<EnvelopeSpec>()
+
+  fun text(body: String, timestamp: Long = System.currentTimeMillis()): EnvelopeContentSpec = EnvelopeContentSpec.Text(body, timestamp, group = null)
+
+  fun groupText(body: String, group: GroupHandle, timestamp: Long = System.currentTimeMillis()): EnvelopeContentSpec = EnvelopeContentSpec.Text(body, timestamp, group)
+
+  fun deliveryReceipts(targets: List<Long>, sentAt: Long = System.currentTimeMillis()): EnvelopeContentSpec = EnvelopeContentSpec.DeliveryReceipt(targets, sentAt)
+
+  fun readReceipts(targets: List<Long>, sentAt: Long = System.currentTimeMillis()): EnvelopeContentSpec = EnvelopeContentSpec.ReadReceipt(targets, sentAt)
+
+  fun malformedEnvelope(timestamp: Long = System.currentTimeMillis()): EnvelopeContentSpec = EnvelopeContentSpec.Malformed(timestamp)
+
+  infix fun EnvelopeContentSpec.from(peer: OtherClient) {
+    specs += EnvelopeSpec(this, peer)
+  }
+}
+
+/** Opaque envelope content returned by [DeliveryBuilder]. Tests never construct or inspect variants directly; the type only appears as a return / receiver of the DSL methods. */
+sealed class EnvelopeContentSpec {
+  internal data class Text(val body: String, val timestamp: Long, val group: GroupHandle?) : EnvelopeContentSpec()
+  internal data class DeliveryReceipt(val targets: List<Long>, val sentAt: Long) : EnvelopeContentSpec()
+  internal data class ReadReceipt(val targets: List<Long>, val sentAt: Long) : EnvelopeContentSpec()
+  internal data class Malformed(val timestamp: Long) : EnvelopeContentSpec()
+}
+
+internal data class EnvelopeSpec(val content: EnvelopeContentSpec, val peer: OtherClient) {
+  fun materialize(): Envelope = when (val c = content) {
+    is EnvelopeContentSpec.Text ->
+      peer.encrypt(Generator.encryptedTextMessage(c.timestamp, c.body, c.group?.let { Harness.groupMasterKey }))
+    is EnvelopeContentSpec.DeliveryReceipt ->
+      peer.encrypt(Generator.encryptedDeliveryReceipt(c.sentAt, c.targets), c.sentAt)
+    is EnvelopeContentSpec.ReadReceipt ->
+      peer.encrypt(Generator.encryptedReadReceipt(c.sentAt, c.targets), c.sentAt)
+    is EnvelopeContentSpec.Malformed -> {
+      val valid = peer.encrypt(Generator.encryptedTextMessage(c.timestamp))
+      val original = valid.content ?: error("Encrypted envelope unexpectedly had no content")
+      val corrupted = original.toByteArray().also { it[it.size / 2] = (it[it.size / 2].toInt() xor 0x01).toByte() }
+      valid.copy(content = corrupted.toByteString())
+    }
+  }
+}
@@ -0,0 +1,18 @@
+package org.thoughtcrime.securesms.testing.incomingmessageobserver
+
+import android.app.Application
+import android.content.Context
+import androidx.test.runner.AndroidJUnitRunner
+import org.thoughtcrime.securesms.IncomingMessageObserverInstrumentationApplicationContext
+
+/**
+ * Test runner that swaps in [IncomingMessageObserverInstrumentationApplicationContext] so the
+ * `IncomingMessageObserver` test harness can drive a faked websocket. Selected automatically by
+ * the build when `-PimoTests` is set.
+ */
+@Suppress("unused")
+class IncomingMessageObserverTestRunner : AndroidJUnitRunner() {
+  override fun newApplication(cl: ClassLoader?, className: String?, context: Context?): Application {
+    return super.newApplication(cl, IncomingMessageObserverInstrumentationApplicationContext::class.java.name, context)
+  }
+}