apps\iot\src\test\java\cn\nfarms\iot\DeviceSessionTest.kt

源码路径:apps\iot\src\test\java\cn\nfarms\iot\DeviceSessionTest.kt · 92 行

package cn.nfarms.iot

import cn.nfarms.iot.device.DeviceSession
import cn.nfarms.iot.device.LinkState
import cn.nfarms.iot.ota.FirmwareFactory
import cn.nfarms.iot.ota.OtaVerdict
import cn.nfarms.iot.ota.OtaVerifier
import cn.nfarms.iot.transport.MockTransport
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Test

/**
 * **完整会话**的端到端测试:握手 → 读状态 → OTA。
 *
 * 为什么还要这一条(已经有一条直接操作传输层的测试了):
 * `DeviceSession` 里有一层「等回包」的派发逻辑(waiter / CompletableDeferred),
 * 它和传输层分片是**两个独立的失败点**。上一轮只测了传输层,
 * 于是"设备收到了帧、但会话侧没认出回包"这一类问题仍然漏得掉。
 */
class DeviceSessionTest {

    @Test
    fun handshake_then_status_then_ota_through_the_real_session() = runBlocking {
        val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
        val fw = FirmwareFactory()
        val transport = MockTransport(dropRate = 0.0)
        transport.pendingVersion = "1.5.0"
        val session = DeviceSession(
            scope = scope,
            transport = transport,
            verifier = OtaVerifier(fw.publicKeyDer()),
            deviceTarget = "NF-Demo-01",
            initialVersion = "1.4.2",
        )

        try {
            // ---- 握手 ----
            session.connect("demo-01")
            assertTrue(
                "握手后必须进入 Ready(实际:${session.ui.value.link})",
                awaitLink(session, LinkState.Ready, 5_000),
            )
            assertTrue("握手后应当读到设备版本", session.ui.value.status.version.isNotBlank())

            // ---- OTA ----
            val pkg = fw.make("1.5.0", target = "NF-Demo-01")
            assertEquals(OtaVerdict.Ok, OtaVerifier(fw.publicKeyDer()).verify(pkg, "1.4.2", "NF-Demo-01"))
            session.runOta(pkg)

            val done = withTimeoutOrNull(60_000) {
                while (true) {
                    if (session.ui.value.otaProgress >= 1f && session.ui.value.link == LinkState.Ready) return@withTimeoutOrNull true
                    delay(100)
                }
                @Suppress("UNREACHABLE_CODE")
                false
            } ?: false

            val ui = session.ui.value
            assertTrue(
                "OTA 必须跑完并回到 Ready。progress=${ui.otaProgress} link=${ui.link} " +
                    "verdict=${ui.otaVerdict} retries=${ui.retries}\n日志:\n" +
                    ui.log.takeLast(15).joinToString("\n"),
                done,
            )
            assertEquals(
                "升级后设备版本必须变化",
                "1.5.0",
                transport.status().version,
            )
        } finally {
            session.close()
        }
    }

    private suspend fun awaitLink(session: DeviceSession, want: LinkState, timeoutMs: Long): Boolean =
        withTimeoutOrNull(timeoutMs) {
            while (session.ui.value.link != want) {
                if (session.ui.value.link == LinkState.Failed) return@withTimeoutOrNull false
                delay(50)
            }
            true
        } ?: false
}