diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt index 12128dd..f4bebe8 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt @@ -57,33 +57,43 @@ class PipelineSmokeTest { assertNotNull(controller) } + companion object { + /** 合法 META + FLOP，但未注册 Handler → FAILED(UNSUPPORTED) */ + val UNSUPPORTED_FLOP_XML = """ + + AODB120260908120000FLOPDELY + F1 + + """.trimIndent() + } + @Test - fun `accept then pump tick transitions message to FAILED CODEC_ERROR with backoff`() { - val receipt = controller.send("") + fun `accept then pump tick transitions message to FAILED UNSUPPORTED with backoff`() { + val receipt = controller.send(UNSUPPORTED_FLOP_XML) assertNotNull(receipt.body()) // 受理 ID val id = receipt.body()!!.toLong() - pump.tick() // 主泵领取 → stub codec 未实装 → FAILED(CODEC_ERROR) + pump.tick() // 解码成功但无 DELY Handler → FAILED(UNSUPPORTED) val stub = ctx.getBean(StubProcState::class.java) val s = stub.snapshotOf(id) assertNotNull(s) assertEquals(ProcStatus.FAILED, s!!.state) - assertEquals(ErrorClass.CODEC_ERROR, s.errorClass) + assertEquals(ErrorClass.UNSUPPORTED, s.errorClass) assertEquals(1, s.attempts) assertNotNull(s.nextAttemptAt) } @Test - fun `replay reopens failed CODEC_ERROR row to PENDING`() { - val receipt = controller.send("") + fun `replay reopens failed UNSUPPORTED row to PENDING`() { + val receipt = controller.send(UNSUPPORTED_FLOP_XML) val id = receipt.body()!!.toLong() pump.tick() val stub = ctx.getBean(StubProcState::class.java) assertEquals(ProcStatus.FAILED, stub.snapshotOf(id)!!.state) val n = ctx.getBean(com.gzzn.omms.msgexchange.infra.retry.ReplayService::class.java) - .replay(listOf(ErrorClass.CODEC_ERROR)) + .replay(listOf(ErrorClass.UNSUPPORTED)) assertEquals(1, n) val reopened = stub.snapshotOf(id)!! diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt new file mode 100644 index 0000000..6b6a99b --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt @@ -0,0 +1,94 @@ +package com.gzzn.omms.msgexchange.codec + +import com.gzzn.omms.msgexchange.domain.MsgKind +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertInstanceOf +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class JacksonXmlCodecTest { + + private val codec = JacksonXmlCodec() + + @Test + fun `decode AODBGTDT sample extracts META and three gates`() { + val raw = """ + + + + AODB + 1243 + 20021010090311 + FLOP + GTDT + + + 121112312 + CA-CA101-A-12DEC031345-D + + G28 + 15DEC031805 + 15DEC031925 + 15DEC031825 + D + + + G33 + 15DEC031805 + 15DEC031925 + I + + + G23 + 15DEC031805 + 15DEC031925 + D + + + + """.trimIndent() + + val result = codec.decode(raw) + assertInstanceOf(DecodeResult.Ok::class.java, result) + val msg = (result as DecodeResult.Ok).message + assertEquals(MsgKind.Flop("GTDT"), msg.kind) + assertEquals("AODB", msg.meta.sndr) + assertEquals(1243L, msg.meta.seqn) + + val body = msg.body as FlopPayload + assertEquals("121112312", body.flid) + assertEquals("CA-CA101-A-12DEC031345-D", body.scalars["FFID"]) + assertEquals(3, body.collections["GTDT"]!!.size) + assertEquals("G28", body.collections["GTDT"]!![0]["GATE"]) + assertEquals("1", body.collections["GTDT"]!![0]["GTNO"]) + assertEquals("G33", body.collections["GTDT"]!![1]["GATE"]) + assertEquals("3", body.collections["GTDT"]!![2]["GTNO"]) + } + + @Test + fun `decode GTNO zero marks gate clear snapshot`() { + val raw = """ + + + AODB120021010090311 + FLOPGTDT + + + F_CLEAR + + + + """.trimIndent() + + val msg = (codec.decode(raw) as DecodeResult.Ok).message + val body = msg.body as FlopPayload + assertEquals(listOf(mapOf("GTNO" to "0")), body.collections["GTDT"]) + } + + @Test + fun `malformed xml returns MALFORMED not CODEC_ERROR`() { + val err = codec.decode("not-xml") as DecodeResult.Err + assertEquals(com.gzzn.omms.msgexchange.domain.ErrorClass.MALFORMED, err.failure.errorClass) + assertTrue(err.failure.detail.contains("empty-or-non-xml")) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt new file mode 100644 index 0000000..020e7ff --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt @@ -0,0 +1,109 @@ +package com.gzzn.omms.msgexchange.domain.flight + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class FlightStateEngineTest { + + @Test + fun `scalar unchanged does not modify existing value`() { + val current = FlightNextState("F1", mapOf("FLNO" to "CA100"), emptyMap(), 1L, "m1") + val next = FlightStateEngine.apply( + current, + FlightFieldCommands("F1", scalars = emptyMap()), + "m2", + bumpVersion = true, + ) + assertEquals("CA100", next.scalars["FLNO"]) + assertEquals(2L, next.stateVersion) + } + + @Test + fun `collection replace preserves order and source sequence`() { + val items = listOf( + mapOf("GTNO" to "1", "GATE" to "A1"), + mapOf("GTNO" to "3", "GATE" to "B2"), + mapOf("GTNO" to "7", "GATE" to "C3"), + ) + val next = FlightStateEngine.apply( + null, + FlightFieldCommands("F1", collections = mapOf("GTDT" to CollectionCommand.Replace(items))), + "m1", + bumpVersion = true, + ) + assertEquals(3, next.collections["GTDT"]!!.size) + assertEquals("3", next.collections["GTDT"]!![1]["GTNO"]) + assertEquals("C3", next.collections["GTDT"]!![2]["GATE"]) + } + + @Test + fun `collection clear removes prior entries`() { + val current = FlightNextState( + "F1", + emptyMap(), + mapOf("GTDT" to listOf(mapOf("GTNO" to "1", "GATE" to "A1"))), + 1L, + "m1", + ) + val next = FlightStateEngine.apply( + current, + FlightFieldCommands("F1", collections = mapOf("GTDT" to CollectionCommand.Clear)), + "m2", + bumpVersion = true, + ) + assertFalse(next.collections.containsKey("GTDT")) + } + + @Test + fun `same gate number different attributes both retained on replace`() { + val items = listOf( + mapOf("GTNO" to "1", "GATE" to "A1", "GTYP" to "D"), + mapOf("GTNO" to "1", "GATE" to "A1", "GTYP" to "I"), + ) + val next = FlightStateEngine.apply( + null, + FlightFieldCommands("F1", collections = mapOf("GTDT" to CollectionCommand.Replace(items))), + "m1", + bumpVersion = true, + ) + assertEquals("D", next.collections["GTDT"]!![0]["GTYP"]) + assertEquals("I", next.collections["GTDT"]!![1]["GTYP"]) + } + + @Test + fun `incremental apply bumps state version from persisted value`() { + val current = FlightNextState("F1", mapOf("FLNO" to "CA100"), emptyMap(), 5L, "m1") + val next = FlightStateEngine.apply( + current, + FlightFieldCommands("F1", scalars = mapOf("STAT" to ScalarCommand.Set("DEP"))), + "m2", + bumpVersion = true, + ) + assertEquals(6L, next.stateVersion) + } + + @Test + fun `apply by source sequence updates single row`() { + val current = FlightNextState( + "F1", + emptyMap(), + mapOf("GTDT" to listOf(mapOf("GTNO" to "2", "GATE" to "OLD"))), + 1L, + "m1", + ) + val next = FlightStateEngine.apply( + current, + FlightFieldCommands( + "F1", + collections = mapOf( + "GTDT" to CollectionCommand.Apply(mapOf("GTNO" to "2", "GATE" to "NEW"), "2"), + ), + ), + "m2", + bumpVersion = true, + ) + assertEquals("NEW", next.collections["GTDT"]!![0]["GATE"]) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt index b361055..1e262cf 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt @@ -11,6 +11,7 @@ import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Assumptions.assumeTrue import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test +import com.gzzn.omms.msgexchange.support.PgTestSupport import com.zaxxer.hikari.HikariDataSource import java.sql.DriverManager import java.time.Instant @@ -37,22 +38,12 @@ class FlightSchdJdbcPgTest { @BeforeEach fun setUp() { - val url = "jdbc:postgresql://localhost:5432/msgx" - val user = "msgx_dev" - val pass = "msgx_dev_pass" - - // 仅当本地 PostgreSQL 容器可用时执行 - val canConnect = try { - DriverManager.getConnection(url, user, pass).use { true } - } catch (_: Exception) { - false - } - assumeTrue(canConnect, "Local PostgreSQL on port 5432 is not accessible, skipping PG dialect integration tests") + assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage()) ds = HikariDataSource().apply { - jdbcUrl = url - username = user - this.password = pass + jdbcUrl = PgTestSupport.jdbcUrl + username = PgTestSupport.user + password = PgTestSupport.password driverClassName = "org.postgresql.Driver" maximumPoolSize = 2 } @@ -244,7 +235,7 @@ class FlightSchdJdbcPgTest { // 持锁事务内：另一连接 NOWAIT 获取同行锁必须立即失败（互斥生效） assertEquals("held", txManager.inTransaction { - DriverManager.getConnection("jdbc:postgresql://localhost:5432/msgx", "msgx_dev", "msgx_dev_pass").use { other -> + DriverManager.getConnection(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password).use { other -> other.autoCommit = false try { other.prepareStatement( diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt index b1fc815..784e360 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt @@ -1,5 +1,6 @@ package com.gzzn.omms.msgexchange.infra.persistence.jdbc +import com.gzzn.omms.msgexchange.support.PgTestSupport import org.flywaydb.core.Flyway import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertTrue @@ -18,16 +19,10 @@ class FlywayMigrationTest { @Test fun `Flyway automatically migrates V1_0_0 and V1_1_0 onto real PostgreSQL`() { - val url = "jdbc:postgresql://localhost:5432/msgx" - val user = "msgx_dev" - val pass = "msgx_dev_pass" - - val canConnect = try { - DriverManager.getConnection(url, user, pass).use { true } - } catch (_: Exception) { - false - } - assumeTrue(canConnect, "PostgreSQL on port 5432 not accessible, skipping Flyway test") + assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage()) + val url = PgTestSupport.jdbcUrl + val user = PgTestSupport.user + val pass = PgTestSupport.password val flyway = Flyway.configure() .dataSource(url, user, pass) diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt index 4d858df..ad239e9 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt @@ -1,5 +1,6 @@ package com.gzzn.omms.msgexchange.ingress +import com.gzzn.omms.msgexchange.domain.ErrorClass import com.gzzn.omms.msgexchange.domain.ProcStatus import com.gzzn.omms.msgexchange.infra.stub.StubInbox import com.gzzn.omms.msgexchange.infra.stub.StubProcState @@ -40,6 +41,7 @@ class InboxPollerTest { poller.pollOnce() assertEquals(0, poller.pollOnce()) pump.tick() - assertEquals(ProcStatus.FAILED, stubProc.snapshotOf(id)!!.state) + assertEquals(ProcStatus.DEAD, stubProc.snapshotOf(id)!!.state) + assertEquals(ErrorClass.MALFORMED, stubProc.snapshotOf(id)!!.errorClass) } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/GtdtPipelineTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/GtdtPipelineTest.kt new file mode 100644 index 0000000..791ad59 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/GtdtPipelineTest.kt @@ -0,0 +1,82 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.codec.DecodeResult +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.ProcState +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler +import com.gzzn.omms.msgexchange.infra.retry.ProcFailure +import com.gzzn.omms.msgexchange.infra.stub.StubFlightSchd +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack +import com.gzzn.omms.msgexchange.processing.handlers.GtdtHandler +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +/** P2-0-1：GTDT 报文端到端 SUCCEEDED + 三门读回（真实 XmlCodec + GtdtHandler）。 */ +class GtdtPipelineTest { + + @Test + fun `GTDT XML end-to-end SUCCEEDED with three gates persisted`() { + val raw = """ + + + + AODB900120260908120000 + FLOPGTDT + + + FL_GTDT_E2E + A1D + B2I + C3D + + + """.trimIndent() + + val procState = StubProcState() + val flightSchd = StubFlightSchd() + val msgEvents = StubMsgEvents() + val reqTrack = StubReqTrack() + val txManager = object : PipelineTransactionManager { + override fun inTransaction(block: () -> T): T = block() + } + val inbox = object : CminmsgInboxRepository { + override fun insertRaw(rawXml: String) = 1L + override fun rawOf(cminmsgsId: Long) = raw + override fun pollUnprocessed(afterId: Long, limit: Int) = emptyList() + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit + } + val clock = MutableClock(MutableClock.BASE) + val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), clock)) + val codec = JacksonXmlCodec() + val handler = GtdtHandler() + val snapshot = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox) + val processor = MessageProcessor( + inbox, procState, msgEvents, + CodecHolder(codec), + HandlerHolder(HandlerRegistry(listOf(handler))), + flightSchd, snapshot, procFailure, PipelineProps(), txManager, + ) + + val headId = 42L + procState.insert(headId, ProcStatus.PENDING) + processor.processOne(ProcState(headId, ProcStatus.PENDING)) + + assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state) + val next = flightSchd.findNextStateByFlid("FL_GTDT_E2E")!! + assertEquals(3, next.collections["GTDT"]!!.size) + assertEquals("3", next.collections["GTDT"]!![1]["GTNO"]) + assertEquals("C3", next.collections["GTDT"]!![2]["GATE"]) + assertEquals(1L, next.stateVersion) + + val roundtrip = FlightStateEngine.fromFlightFields("FL_GTDT_E2E", flightSchd.findByFlid("FL_GTDT_E2E")!!) + assertEquals(3, roundtrip.collections["GTDT"]!!.size) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt index 653034d..87d1333 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/MessageProcessorTest.kt @@ -175,6 +175,22 @@ class MessageProcessorTest { return true } override fun deleteGenBefore(cutoffDay: String): Int = 0 + override fun findNextStateByFlid(flid: String): com.gzzn.omms.msgexchange.domain.flight.FlightNextState? = + flights[flid]?.let { com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine.fromFlightFields(flid, it) } + override fun persistNextStates( + day: String?, + states: List, + snapshotReplace: Boolean, + now: Instant, + ) { + states.forEach { state -> + val scalarFields = linkedMapOf("FLID" to state.flid) + state.scalars.forEach { (k, v) -> scalarFields[k] = v } + flights[state.flid] = scalarFields + state.collections.mapValues { (_, items) -> + com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(items) + } + } + } } private class FakeTxManager : PipelineTransactionManager { diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt new file mode 100644 index 0000000..8616e80 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt @@ -0,0 +1,110 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.MetaFields +import com.gzzn.omms.msgexchange.domain.MsgKind +import com.gzzn.omms.msgexchange.domain.ProcState +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository +import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository +import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository +import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler +import com.gzzn.omms.msgexchange.infra.retry.ProcFailure +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.infra.stub.StubFlightSchd +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +/** v2 §2.6 / §5：提交后回填失败不得回滚 PROC_STATE。 */ +class SnapshotFlowBackfillTest { + + private class FailingInbox : CminmsgInboxRepository { + override fun insertRaw(rawXml: String): Long = 1L + override fun rawOf(cminmsgsId: Long): String? = null + override fun pollUnprocessed(afterId: Long, limit: Int): List = emptyList() + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) { + throw RuntimeException("mysql-down") + } + } + + @Test + fun `backfill failure after commit leaves PROC_STATE SUCCEEDED`() { + val flightSchd = StubFlightSchd() + val procState = StubProcState() + val msgEvents = StubMsgEvents() + val reqTrack = StubReqTrack() + val txManager = StubPipelineTransactionManager() + val inbox = FailingInbox() + val clock = MutableClock(MutableClock.BASE) + val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), clock)) + + val headId = 901L + procState.insert(headId, ProcStatus.PENDING) + val day = "2026-09-07" + val meta = MetaFields("AODB", "SCHD", "DNLD", 9L, 20260907090000L) + val decoded = DecodedMessage(meta, MsgKind.Schd(MsgKind.SchdSubtype.DNLD), "") + SnapshotFlow.StageResult.parser = { + SnapshotFlow.StageResult.Ok(day, listOf("FL_X" to mapOf("FLID" to "FL_X"))) + } + + val flow = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox) + flow.publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded) + + assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state) + assertTrue(flightSchd.findByFlid("FL_X") != null) + } + + @Test + fun `CAS replay requires matching lastMessageId not version alone`() { + val base = StubFlightSchd() + val procState = StubProcState() + val msgEvents = StubMsgEvents() + val reqTrack = StubReqTrack() + val txManager = StubPipelineTransactionManager() + val inbox = object : CminmsgInboxRepository { + override fun insertRaw(rawXml: String) = 1L + override fun rawOf(cminmsgsId: Long) = null + override fun pollUnprocessed(afterId: Long, limit: Int): List = emptyList() + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit + } + val clock = MutableClock(MutableClock.BASE) + val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), clock)) + + val headId = 902L + val day = "2026-09-07" + procState.insert(headId, ProcStatus.PENDING) + + val conflictingGen = FlightSchdRepository.GenMeta(day, 2L, setOf("OTHER"), lastMessageId = "999") + base.putGenIfVersion(day, 0L, FlightSchdRepository.GenMeta(day, 1L, setOf("OTHER"), lastMessageId = "999")) + val mockFlightSchd = object : FlightSchdRepository by base { + override fun putGenIfVersion( + day: String, + expected: Long, + newGen: FlightSchdRepository.GenMeta, + now: java.time.Instant, + ): Boolean { + base.putGenIfVersion(day, 1L, conflictingGen) + return false + } + } + + SnapshotFlow.StageResult.parser = { + SnapshotFlow.StageResult.Ok(day, listOf("FL_Y" to mapOf("FLID" to "FL_Y"))) + } + val meta = MetaFields("AODB", "SCHD", "DNLD", 10L, 20260907100000L) + val decoded = DecodedMessage(meta, MsgKind.Schd(MsgKind.SchdSubtype.DNLD), "") + val flow = SnapshotFlow(procState, mockFlightSchd, msgEvents, reqTrack, procFailure, txManager, inbox) + flow.publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded) + + assertEquals(ProcStatus.FAILED, procState.snapshotOf(headId)!!.state) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/V2PipelineWritePathContractTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/V2PipelineWritePathContractTest.kt new file mode 100644 index 0000000..8a62a53 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/V2PipelineWritePathContractTest.kt @@ -0,0 +1,115 @@ +package com.gzzn.omms.msgexchange.processing + +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.Decision +import com.gzzn.omms.msgexchange.domain.FlightChange +import com.gzzn.omms.msgexchange.domain.MetaFields +import com.gzzn.omms.msgexchange.domain.MsgKind +import com.gzzn.omms.msgexchange.domain.ProcState +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository +import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler +import com.gzzn.omms.msgexchange.infra.retry.ProcFailure +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTransactionManager +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack +import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec +import com.gzzn.omms.msgexchange.processing.handlers.GtdtHandler +import com.gzzn.omms.msgexchange.support.GuardedFlightSchd +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.nio.file.Paths +import kotlin.io.path.readText + +class V2PipelineWritePathContractTest { + + @Test + fun `SnapshotFlow and MessageProcessor sources must not call legacy upsert helpers`() { + val processingDir = Paths.get("src/main/kotlin/com/gzzn/omms/msgexchange/processing") + val snapshotSource = processingDir.resolve("SnapshotFlow.kt").readText() + val pumpSource = processingDir.resolve("Pump.kt").readText() + + assertFalse(snapshotSource.contains("upsertSnapshotBatch")) + assertFalse(snapshotSource.contains("upsertIncremental")) + assertTrue(snapshotSource.contains("persistNextStates")) + + assertFalse(pumpSource.contains("upsertSnapshotBatch")) + assertFalse(pumpSource.contains("upsertIncremental")) + assertTrue(pumpSource.contains("persistNextStates")) + } + + @Test + fun `SnapshotFlow DNLD path calls persistNextStates not legacy upsert helpers`() { + val flightSchd = GuardedFlightSchd() + val procState = StubProcState() + val msgEvents = StubMsgEvents() + val reqTrack = StubReqTrack() + val txManager = StubPipelineTransactionManager() + val inbox = object : CminmsgInboxRepository { + override fun insertRaw(rawXml: String) = 1L + override fun rawOf(cminmsgsId: Long) = null + override fun pollUnprocessed(afterId: Long, limit: Int): List = emptyList() + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit + } + val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), MutableClock(MutableClock.BASE))) + + val headId = 1001L + procState.insert(headId, ProcStatus.PENDING) + val day = "2026-09-07" + SnapshotFlow.StageResult.parser = { + SnapshotFlow.StageResult.Ok(day, listOf("FL_DNLD" to mapOf("FLID" to "FL_DNLD", "FLNO" to "CA100", "GTDT" to """[{"GTNO":"1","GATE":"A1"}]"""))) + } + val decoded = DecodedMessage(MetaFields("AODB", "SCHD", "DNLD", 11L, 20260907110000L), MsgKind.Schd(MsgKind.SchdSubtype.DNLD), "") + SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox) + .publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded) + + assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state) + assertEquals(1, flightSchd.persistNextStatesCalls) + assertEquals(0, flightSchd.legacySnapshotCalls) + assertEquals(0, flightSchd.legacyIncrementalCalls) + } + + @Test + fun `MessageProcessor FLOP path calls persistNextStates not legacy upsert helpers`() { + val flightSchd = GuardedFlightSchd() + val procState = StubProcState() + val msgEvents = StubMsgEvents() + val reqTrack = StubReqTrack() + val txManager = StubPipelineTransactionManager() + val raws = mutableMapOf() + val inbox = object : CminmsgInboxRepository { + override fun insertRaw(rawXml: String) = 1L + override fun rawOf(cminmsgsId: Long) = raws[cminmsgsId] + override fun pollUnprocessed(afterId: Long, limit: Int): List = emptyList() + override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit + } + val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), MutableClock(MutableClock.BASE))) + val handler = GtdtHandler() + val snapshot = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox) + val processor = MessageProcessor( + inbox, procState, msgEvents, + CodecHolder(JacksonXmlCodec()), + HandlerHolder(HandlerRegistry(listOf(handler))), + flightSchd, snapshot, procFailure, PipelineProps(), txManager, + ) + val headId = 1002L + procState.insert(headId, ProcStatus.PENDING) + raws[headId] = """ + + AODB120260907120000FLOPGTDT + FL_FLOPB9 + + """.trimIndent() + processor.processOne(ProcState(headId, ProcStatus.PENDING)) + + assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state) + assertEquals(1, flightSchd.persistNextStatesCalls) + assertEquals(0, flightSchd.legacySnapshotCalls) + assertEquals(0, flightSchd.legacyIncrementalCalls) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/handlers/GtdtHandlerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/handlers/GtdtHandlerTest.kt new file mode 100644 index 0000000..b50151e --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/handlers/GtdtHandlerTest.kt @@ -0,0 +1,41 @@ +package com.gzzn.omms.msgexchange.processing.handlers + +import com.gzzn.omms.msgexchange.codec.FlopPayload +import com.gzzn.omms.msgexchange.domain.DecodedMessage +import com.gzzn.omms.msgexchange.domain.MetaFields +import com.gzzn.omms.msgexchange.domain.MsgKind +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class GtdtHandlerTest { + + private val handler = GtdtHandler() + + @Test + fun `decide replaces GTDT collection for three gates`() { + val msg = DecodedMessage( + meta = MetaFields("AODB", "FLOP", "GTDT", 1L, 20260908120000L), + kind = MsgKind.Flop("GTDT"), + rawXml = "", + body = FlopPayload( + flid = "121112312", + scalars = mapOf("FFID" to "CA-CA101-A-12DEC031345-D"), + collections = mapOf( + "GTDT" to listOf( + mapOf("GTNO" to "1", "GATE" to "G28", "GTYP" to "D"), + mapOf("GTNO" to "2", "GATE" to "G33", "GTYP" to "I"), + mapOf("GTNO" to "3", "GATE" to "G23", "GTYP" to "D"), + ), + ), + ), + ) + + val decision = handler.decide(emptyMap(), msg) + assertEquals(1, decision.flightChanges.size) + assertEquals("121112312", decision.flightChanges[0].flid) + val gtdtJson = decision.flightChanges[0].fields["GTDT"]!! + assertEquals(true, gtdtJson.contains("\"G28\"")) + assertEquals(1, decision.schdPush.size) + assertEquals("121112312", decision.schdPush[0].flid) + } +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/support/PgTestSupport.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/support/PgTestSupport.kt new file mode 100644 index 0000000..0a79c87 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/support/PgTestSupport.kt @@ -0,0 +1,30 @@ +package com.gzzn.omms.msgexchange.support + +/** + * JDBC 集成测试 PostgreSQL 连接参数（环境变量化，解决本机 5432 端口冲突）。 + * + * | 变量 | 默认 | + * |---|---| + * | MSGX_PG_URL | jdbc:postgresql://${MSGX_PG_HOST:127.0.0.1}:${MSGX_PG_PORT:5432}/${MSGX_PG_NAME:msgx} | + * | MSGX_PG_USER | msgx_dev | + * | MSGX_PG_PASSWORD | msgx_dev_pass | + */ +object PgTestSupport { + private val host = System.getenv("MSGX_PG_HOST") ?: "127.0.0.1" + private val port = System.getenv("MSGX_PG_PORT") ?: "5432" + private val name = System.getenv("MSGX_PG_NAME") ?: "msgx" + + val jdbcUrl: String = System.getenv("MSGX_PG_URL") + ?: "jdbc:postgresql://$host:$port/$name" + val user: String = System.getenv("MSGX_PG_USER") ?: "msgx_dev" + val password: String = System.getenv("MSGX_PG_PASSWORD") ?: "msgx_dev_pass" + + fun canConnect(): Boolean = try { + java.sql.DriverManager.getConnection(jdbcUrl, user, password).use { true } + } catch (_: Exception) { + false + } + + fun skipMessage(): String = + "PostgreSQL not accessible at $jdbcUrl (set MSGX_PG_URL / MSGX_PG_PORT to avoid port conflicts)" +} diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/support/V2WritePathGuard.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/support/V2WritePathGuard.kt new file mode 100644 index 0000000..7cb5733 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/support/V2WritePathGuard.kt @@ -0,0 +1,60 @@ +package com.gzzn.omms.msgexchange.support + +import com.gzzn.omms.msgexchange.domain.FlightChange +import com.gzzn.omms.msgexchange.domain.flight.FlightNextState +import com.gzzn.omms.msgexchange.infra.persistence.FlightFields +import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository +import com.gzzn.omms.msgexchange.infra.stub.StubFlightSchd +import java.time.Instant + +/** V1.2.0 宽表集合存储列；v2 persistNextStates 不得写入。 */ +val LEGACY_COLLECTION_STORAGE_COLUMNS: List = listOf( + "gate1", "pgot1", "pgct1", "gotm1", "gctm1", "gtyp1", + "gate2", "pgot2", "pgct2", "gotm2", "gctm2", "gtyp2", + "chkc1", "ccls1", "pcot1", "pcct1", "cotm1", "cctm1", "ctyp1", + "chkc2", "ccls2", "pcot2", "pcct2", "cotm2", "cctm2", "ctyp2", + "chkc3", "ccls3", "pcot3", "pcct3", "cotm3", "cctm3", "ctyp3", + "belt1", "bcls1", "bpcot1", "bpcct1", "fbag1", "lbag1", "btyp1", + "belt2", "bcls2", "bpcot2", "bpcct2", "fbag2", "lbag2", "btyp2", + "psst1", "stst1", "stet1", "psst2", "stst2", "stet2", + "chut1", "chcls1", "pcbt1", "pcet1", "cbtm1", "cetm1", "chtyp1", + "chut2", "chcls2", "pcbt2", "pcet2", "cbtm2", "cetm2", "chtyp2", + "dely_code", "dely_strt", "dely_dura", "dely_remc", + "abtm_a", "abtm_d", "chot_on", "chot_off", + "rout_path", "erut_path", +) + +class GuardedFlightSchd( + private val inner: FlightSchdRepository = StubFlightSchd(), +) : FlightSchdRepository { + var legacySnapshotCalls = 0 + var legacyIncrementalCalls = 0 + var persistNextStatesCalls = 0 + + override fun upsertSnapshotBatch(day: String, flights: List>, now: Instant) { + legacySnapshotCalls++ + throw AssertionError("v2 path must not call upsertSnapshotBatch") + } + + override fun upsertIncremental(changes: List, now: Instant) { + legacyIncrementalCalls++ + throw AssertionError("v2 path must not call upsertIncremental") + } + + override fun persistNextStates(day: String?, states: List, snapshotReplace: Boolean, now: Instant) { + persistNextStatesCalls++ + inner.persistNextStates(day, states, snapshotReplace, now) + } + + override fun deleteDiffByDay(day: String, delFlids: Collection): Int = inner.deleteDiffByDay(day, delFlids) + override fun findByFlid(flid: String): FlightFields? = inner.findByFlid(flid) + override fun findNextStateByFlid(flid: String): FlightNextState? = inner.findNextStateByFlid(flid) + override fun findByFlids(flids: Collection): Map = inner.findByFlids(flids) + override fun findByDay(day: String): List> = inner.findByDay(day) + override fun findAll(): Map = inner.findAll() + override fun deleteByFlids(flids: Set): Int = inner.deleteByFlids(flids) + override fun getGen(day: String): FlightSchdRepository.GenMeta? = inner.getGen(day) + override fun putGenIfVersion(day: String, expected: Long, newGen: FlightSchdRepository.GenMeta, now: Instant): Boolean = + inner.putGenIfVersion(day, expected, newGen, now) + override fun deleteGenBefore(cutoffDay: String): Int = inner.deleteGenBefore(cutoffDay) +}