apps\iot\src\test\java\cn\nfarms\iot\MockTransportTest.kt¶
源码路径:
apps\iot\src\test\java\cn\nfarms\iot\MockTransportTest.kt· 188 行
package cn.nfarms.iot
import cn.nfarms.iot.ota.OtaVerdict
import cn.nfarms.iot.ota.FirmwareFactory
import cn.nfarms.iot.ota.OtaVerifier
import cn.nfarms.iot.protocol.Cmd
import cn.nfarms.iot.protocol.Codec
import cn.nfarms.iot.protocol.Decoded
import cn.nfarms.iot.protocol.Frame
import cn.nfarms.iot.transport.MockTransport
import java.util.Base64
import java.util.Collections
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotEquals
import org.junit.Assert.assertTrue
import org.junit.Test
/**
* 模拟设备 + 分片传输的端到端测试。
*
* # 这条测试防的是一个真实出现过的 bug
*
* 模拟设备原本把**每个 MTU 分片**(20 字节)单独 `Frame.decode()`。
* 短命令(Hello / ReadStatus)刚好塞进一片,所以一切正常;而 OTA 的
* `OtaChunk` 有 68 字节,被切成 4 片后每一片都解不出一帧 ——
* 设备永远不回 ack,上层只有超时重试直到耗尽。
*
* 现象是「OTA 卡在 95%,而其它功能都好」,从现象几乎不可能定位到根因。
*/
class MockTransportTest {
/**
* 一个把「收到的字节」攒起来、按帧切开的测试侧读取器。
*
* 为什么不直接用 `transport.incoming().collect`:那是热流,
* 在 `runBlocking` 里用 `collect` 会永远挂着,没有"取到就停"的语义。
* 这里起一个后台协程把数据搬进列表,测试侧轮询取用 —— 和真机上的
* 「通知回调 + UI 轮询」是同一种模型。
*/
private class Reader(transport: MockTransport, scope: CoroutineScope) {
private val chunks = Collections.synchronizedList(mutableListOf<ByteArray>())
@Volatile private var buffer: ByteArray = ByteArray(0)
private val frames = Collections.synchronizedList(mutableListOf<Decoded>())
init {
scope.launch {
transport.incoming().collect { c -> chunks += c }
}
scope.launch {
while (true) {
delay(10)
synchronized(chunks) {
while (chunks.isNotEmpty()) {
buffer += chunks.removeAt(0)
}
}
val res = Frame.decode(buffer)
buffer = res.rest
frames += res.frames.filter { it.ok }
}
}
}
/** 等一个指定命令的帧,超时返回 null。 */
suspend fun await(cmd: Cmd, timeoutMs: Long = 3_000): Decoded? =
withTimeoutOrNull(timeoutMs) {
while (true) {
synchronized(frames) {
val i = frames.indexOfFirst { it.cmd == cmd }
if (i >= 0) return@withTimeoutOrNull frames.removeAt(i)
}
delay(10)
}
@Suppress("UNREACHABLE_CODE")
null
}
}
private fun <T> withTransport(dropRate: Double = 0.0, block: suspend (MockTransport, Reader) -> T): T =
runBlocking {
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
val t = MockTransport(dropRate = dropRate)
t.connect("demo-01")
val reader = Reader(t, scope)
try {
block(t, reader)
} finally {
t.disconnect()
}
}
@Test
fun transport_accumulates_fragments_before_decoding() {
// 68 字节的 OtaChunk 必须跨多个 MTU 分片
val bytes = Frame.encode(Cmd.OtaChunk, ByteArray(64) { 7 })
assertTrue("这个 payload 必须跨多个 MTU 分片", bytes.size > 20)
var off = 0
var parts = 0
while (off < bytes.size) { off += 20; parts++ }
assertTrue(parts >= 4)
// 逐片累积解码:只有最后一片之后才能解出一帧
var acc = ByteArray(0)
var count = 0
off = 0
while (off < bytes.size) {
val n = minOf(20, bytes.size - off)
acc += bytes.copyOfRange(off, off + n)
off += n
val r = Frame.decode(acc)
count += r.frames.count { it.ok }
acc = r.rest
}
assertEquals("分片累积后必须恰好解出 1 帧", 1, count)
assertEquals(0, acc.size)
}
@Test
fun short_command_roundtrip() = withTransport { t, reader ->
t.write(Frame.encode(Cmd.Hello, byteArrayOf(1)))
assertTrue("Hello 必须拿到 HelloAck", reader.await(Cmd.HelloAck) != null)
}
/**
* 跨分片的大载荷命令。
*
* **修复前这条会超时** —— 这就是"OTA 卡在 95%"的那一处。
*/
@Test
fun large_payload_command_roundtrip() = withTransport { t, reader ->
val chunk = Codec.chunkPayload(0, ByteArray(64) { 7 })
t.write(Frame.encode(Cmd.OtaChunk, chunk))
assertTrue(
"跨分片的 OtaChunk 必须拿到 ack",
reader.await(Cmd.OtaAck) != null,
)
}
@Test
fun read_status_returns_parsable_payload() = withTransport { t, reader ->
t.write(Frame.encode(Cmd.ReadStatus))
val f = reader.await(Cmd.Status)
assertTrue("Status 必须到达", f != null)
val st = Codec.parseStatus(f!!.payload)
assertTrue("Status 必须能解析", st != null)
assertNotEquals("", st!!.version)
}
/**
* 整条 OTA:从验包到设备版本变化。
*
* 会真的跑完 48 KB 的分片传输(约 1~2 秒),但它覆盖的正是"卡在 95%"那个场景。
*/
@Test
fun full_ota_updates_device_version() = withTransport { t, reader ->
val fw = FirmwareFactory()
val verifier = OtaVerifier(fw.publicKeyDer())
t.pendingVersion = "1.5.0"
val pkg = fw.make("1.5.0", target = "NF-Demo-01")
assertEquals(OtaVerdict.Ok, verifier.verify(pkg, "1.4.2", "NF-Demo-01"))
val bytes = Base64.getDecoder().decode(pkg.payloadB64)
t.write(Frame.encode(Cmd.OtaBegin, byteArrayOf(1)))
assertTrue(reader.await(Cmd.OtaAck) != null)
var off = 0
var idx = 0
while (off < bytes.size) {
val n = minOf(64, bytes.size - off)
t.write(Frame.encode(Cmd.OtaChunk, Codec.chunkPayload(idx, bytes.copyOfRange(off, off + n))))
assertTrue("第 $idx 片必须被 ack", reader.await(Cmd.OtaAck) != null)
off += n
idx++
}
t.write(Frame.encode(Cmd.OtaEnd, byteArrayOf(1)))
assertTrue(reader.await(Cmd.OtaAck) != null)
assertEquals("升级后设备版本必须变化", "1.5.0", t.status().version)
}
}