feat: Save am instrument logs to file & handle parsing error.

corellium/adapter/build.gradle.kts

@@ -15,5 +15,6 @@ dependencies {
     implementation(project(":corellium:client"))
     implementation(Dependencies.KOTLIN_COROUTINES_CORE)
     testImplementation(Dependencies.JUNIT)
+    testImplementation(Dependencies.MOCKK)
 }
 

corellium/adapter/src/main/kotlin/flank/corellium/Api.kt

@@ -1,6 +1,7 @@
 package flank.corellium
 
 import flank.corellium.adapter.executeAndroidTestPlan
+import flank.corellium.adapter.getCorellium
 import flank.corellium.adapter.installAndroidApps
 import flank.corellium.adapter.invokeAndroidDevices
 import flank.corellium.adapter.requestAuthorization
@@ -16,5 +17,5 @@ fun corelliumApi(
     invokeAndroidDevices = invokeAndroidDevices(
         projectName = projectName
     ),
-    executeTest = executeAndroidTestPlan,
+    executeTest = executeAndroidTestPlan(getCorellium),
 )

corellium/adapter/src/main/kotlin/flank/corellium/adapter/ClientReference.kt

@@ -9,3 +9,7 @@ internal val corellium: Corellium
 
 // It's totally ok to keep corellium as singleton since we don't need handle more than one connection for single run.
 internal var corelliumRef: Corellium? = null
+
+internal typealias GetCorellium = () -> Corellium
+
+internal val getCorellium: GetCorellium = { corellium }

corellium/adapter/src/main/kotlin/flank/corellium/adapter/ExecuteAndroidTestPlan.kt

@@ -2,21 +2,26 @@ package flank.corellium.adapter
 
 import flank.corellium.api.AndroidTestPlan
 import flank.corellium.client.console.clear
+import flank.corellium.client.console.close
 import flank.corellium.client.console.flowLogs
 import flank.corellium.client.console.sendCommand
 import flank.corellium.client.console.waitForIdle
 import flank.corellium.client.core.connectConsole
+import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.launch
 
-val executeAndroidTestPlan = AndroidTestPlan.Execute { config ->
+fun executeAndroidTestPlan(
+    corellium: GetCorellium
+) = AndroidTestPlan.Execute { config ->
     config.instances.map { (instanceId, commands: List<String>) ->
         instanceId to channelFlow<String> {
-            corellium.connectConsole(instanceId).apply {
+            corellium().connectConsole(instanceId).apply {
+                invokeOnClose { launch { close() } }
                 clear()
                 launch { commands.forEach { string -> sendCommand(string) } }
-                launch { flowLogs().collect(channel::send) }
+                launch { flowLogs().collect(channel::trySendBlocking) }
                 waitForIdle(10_000)
             }
         }

corellium/adapter/src/test/kotlin/flank/corellium/adapter/ExecuteAndroidTestPlanKtTest.kt

@@ -0,0 +1,72 @@
+package flank.corellium.adapter
+
+import flank.corellium.api.AndroidTestPlan
+import flank.corellium.client.console.Console
+import flank.corellium.client.console.close
+import flank.corellium.client.console.flowLogs
+import flank.corellium.client.console.sendCommand
+import flank.corellium.client.core.Corellium
+import flank.corellium.client.core.connectConsole
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.mockkStatic
+import io.mockk.unmockkAll
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.channelFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.runBlocking
+import org.junit.After
+import org.junit.Assert
+import org.junit.Test
+
+class ExecuteAndroidTestPlanKtTest {
+
+    /**
+     * Connection to [Console] should be closed once subscription detach.
+     */
+    @Test
+    fun closeConsole() {
+        // given
+        mockkStatic(
+            "flank.corellium.client.console.ApiKt",
+            "flank.corellium.client.core.ApiKt",
+        )
+
+        val instanceId = "1"
+        val command = "command"
+        val emitted = mutableListOf<Char>()
+        val range = ('a'..'z')
+        val linesFlow: Flow<String> = channelFlow { range.forEach { send("$it"); emitted += it; delay(10) } }
+        val config = AndroidTestPlan.Config(mapOf(instanceId to listOf(command)))
+
+        val console: Console = mockk(relaxed = true) {
+            val context = Job()
+            every { flowLogs() } returns linesFlow
+            every { coroutineContext } returns context
+        }
+
+        val corellium: Corellium = mockk(relaxed = true) {
+            coEvery { connectConsole(instanceId) } returns console
+        }
+
+        val flow: Flow<String> = executeAndroidTestPlan { corellium }
+            .invoke(config).first().second
+
+        // when
+        runBlocking { flow.first { it == "c" } } // Detach the subscription after "c" element
+
+        // then
+        coVerify(exactly = 1) { console.sendCommand(command) } // Verify sendCommand called.
+        coVerify(exactly = 1) { console.close() } // Verify console close.
+        Assert.assertNotEquals(range.last, emitted.last()) // Check the emission not reach last element.
+    }
+
+    @After
+    fun tearDown() {
+        unmockkAll()
+    }
+}

corellium/api/src/main/kotlin/flank/corellium/api/CorelliumApi.kt

@@ -4,8 +4,8 @@ package flank.corellium.api
  * Corellium API functions.
  */
 class CorelliumApi(
-    val authorize: Authorization.Request,
-    val invokeAndroidDevices: AndroidInstance.Invoke,
-    val installAndroidApps: AndroidApps.Install,
-    val executeTest: AndroidTestPlan.Execute,
+    val authorize: Authorization.Request = Authorization.Request { throw NotImplementedError() },
+    val invokeAndroidDevices: AndroidInstance.Invoke = AndroidInstance.Invoke { throw NotImplementedError() },
+    val installAndroidApps: AndroidApps.Install = AndroidApps.Install { throw NotImplementedError() },
+    val executeTest: AndroidTestPlan.Execute = AndroidTestPlan.Execute { throw NotImplementedError() },
 )

corellium/cli/src/main/kotlin/flank/corellium/cli/RunTestCorelliumAndroidCommand.kt

@@ -232,6 +232,13 @@ internal val format = buildFormatter<String> {
             else -> null
         }
     }
+    ExecuteTests.Error::class {
+        """
+            Error while parsing results from instance $id.
+            For details check $logFile lines $lines.
+
+        """.trimIndent() + cause.stackTraceToString()
+    }
     RunTestCorelliumAndroid.Created { "Created $path" }
     RunTestCorelliumAndroid.AlreadyExist { "Already exist $path" }
 

corellium/cli/src/test/kotlin/flank/corellium/cli/RunTestCorelliumAndroidCommandTest.kt

@@ -239,6 +239,7 @@ password: $password
                     details = Instrument.Status.Details(emptyMap(), "Class", "Test", null)
                 )
             ),
+            Unit event ExecuteTests.Error("1", Exception(), "path/to/log/1", 5..10),
             Unit event RunTestCorelliumAndroid.Created(File("path/to/apk.apk")),
             Unit event RunTestCorelliumAndroid.AlreadyExist(File("path/to/apk.apk")),
         )

corellium/domain/src/main/kotlin/flank/corellium/domain/RunTestCorelliumAndroid.kt

@@ -143,7 +143,15 @@ object RunTestCorelliumAndroid {
     object OutputDir
     object DumpShards
     object ExecuteTests {
+        const val ADB_LOG = "adb_log"
+
         data class Status(val id: String, val status: Instrument) : Event.Data
+        data class Error(
+            val id: String,
+            val cause: Throwable,
+            val logFile: String,
+            val lines: IntRange
+        ) : Event.Data
     }
 
     object CompleteTests

corellium/domain/src/main/kotlin/flank/corellium/domain/run/test/android/step/ExecuteTests.kt

@@ -5,18 +5,28 @@ import flank.corellium.domain.RunTestCorelliumAndroid
 import flank.corellium.domain.RunTestCorelliumAndroid.ExecuteTests
 import flank.corellium.domain.step
 import flank.instrument.command.formatAmInstrumentCommand
+import flank.instrument.log.Instrument
 import flank.instrument.log.parseAdbInstrumentLog
+import flank.shard.Shard
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.dropWhile
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.flow.take
+import java.io.File
 
 /**
- * The step is executing tests on previously invoked devices.
+ * The step is executing tests on previously invoked devices, and returning the test results.
+ *
+ * The side effect is console logs from `am instrument` saved inside [RunTestCorelliumAndroid.ExecuteTests.ADB_LOG] output subdirectory.
+ *
+ * The optional parsing errors are sent through [RunTestCorelliumAndroid.Context.out].
  *
  * require:
  * * [RunTestCorelliumAndroid.Context.authorize]
@@ -29,15 +39,28 @@ import kotlinx.coroutines.flow.toList
  * * [RunTestCorelliumAndroid.State.shards]
  */
 internal fun RunTestCorelliumAndroid.Context.executeTests() = step(ExecuteTests) { out ->
+    val outputDir = File(args.outputDir, ExecuteTests.ADB_LOG).apply { mkdir() }
     val testPlan: AndroidTestPlan.Config = prepareTestPlan()
     val list = coroutineScope {
         api.executeTest(testPlan).map { (id, flow) ->
             async {
-                flow.flowOn(Dispatchers.IO)
-                    .dropWhile { line -> !line.startsWith("INSTRUMENTATION_STATUS") }
+                var read = 0
+                var parsed = 0
+                val file = outputDir.resolve(id)
+                val results = mutableListOf<Instrument>()
+                flow.onEach { file.appendText(it + "\n") }
+                    .buffer(100)
+                    .flowOn(Dispatchers.IO)
+                    .onEach { ++read }
                     .parseAdbInstrumentLog()
-                    .onEach { status -> ExecuteTests.Status(id, status).out() }
-                    .toList()
+                    .onEach { parsed = read }
+                    .onEach { result -> results += result }
+                    .onEach { result -> ExecuteTests.Status(id, result).out() }
+                    .catch { cause -> ExecuteTests.Error(id, cause, file.path, ++parsed..read).out() }
+                    .filterIsInstance<Instrument.Result>()
+                    .take(expectedResultsCountFor(id))
+                    .collect()
+                results
             }
         }.awaitAll()
     }
@@ -46,21 +69,22 @@ internal fun RunTestCorelliumAndroid.Context.executeTests() = step(ExecuteTests)
 
 /**
  * Prepare [AndroidTestPlan.Config] for test execution.
- * It just mapping and formatting the data collected in state.
+ * It is just mapping and formatting the data collected in state.
  */
 private fun RunTestCorelliumAndroid.State.prepareTestPlan(): AndroidTestPlan.Config =
     AndroidTestPlan.Config(
         shards.mapIndexed { index, shards ->
-            ids[index] to shards.flatMap { shard ->
+            ids[index] to shards.flatMap { shard: Shard.App ->
                 shard.tests.map { test ->
                     formatAmInstrumentCommand(
                         packageName = packageNames.getValue(test.name),
                         testRunner = testRunners.getValue(test.name),
-                        testCases = test.cases.map { case ->
-                            "class " + case.name
-                        }
+                        testCases = test.cases.map { case -> "class " + case.name }
                     )
                 }
             }
         }.toMap()
     )
+
+private fun RunTestCorelliumAndroid.State.expectedResultsCountFor(id: String): Int =
+    shards[ids.indexOf(id)].size