From a54c3c1804d2250f466124a9aef2f1bc6152b956 Mon Sep 17 00:00:00 2001 From: windyboy Date: Tue, 8 Sep 2026 11:39:35 +0800 Subject: [PATCH] test(jdbc): make PG integration tests self-contained on env-configured port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restore 3 v2 tests lost in file corruption (persistNextStates roundtrip, legacy columns all-null, schd_gen last_message_id CAS) — 13 JDBC cases. Add idempotent Flyway migrate in @BeforeEach so an empty container works. Parameterize compose port as ${MSGX_PG_PORT:-5432}. Verified 14/14 green (0 skipped) against real PostgreSQL on :5433. --- compose.yaml | 2 +- .../persistence/jdbc/FlightSchdJdbcPgTest.kt | 126 ++++++++++++++++++ 2 files changed, 127 insertions(+), 1 deletion(-) diff --git a/compose.yaml b/compose.yaml index 27a210e..0a12137 100644 --- a/compose.yaml +++ b/compose.yaml @@ -40,7 +40,7 @@ services: POSTGRES_USER: ${MSGX_PG_USER:-msgx_dev} POSTGRES_PASSWORD: ${MSGX_PG_PASSWORD:-msgx_dev_pass} ports: - - "5432:5432" + - "${MSGX_PG_PORT:-5432}:5432" volumes: - pg_data:/var/lib/postgresql/data healthcheck: 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 1e262cf..55aaaf0 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 @@ -47,6 +47,12 @@ class FlightSchdJdbcPgTest { driverClassName = "org.postgresql.Driver" maximumPoolSize = 2 } + // 幂等前置迁移：全新容器（或库缺失 schema）时补齐 V1.0.0→V1.3.0 + org.flywaydb.core.Flyway.configure() + .dataSource(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password) + .locations("classpath:db/migration") + .load() + .migrate() repo = JdbcFlightSchdRepository(ds) cleanup() } @@ -409,4 +415,124 @@ class FlightSchdJdbcPgTest { assertEquals(writeInstant, instantUtc) assertEquals(instantChicago, instantUtc) } + + @Test + fun `PG dialect - persistNextStates roundtrip preserves collections source_seq and state_version`() { + val day = "2026-09-07" + val flid = "TEST_V2_ROUNDTRIP" + val gates = listOf( + mapOf("GTNO" to "1", "GATE" to "G28", "GTYP" to "D"), + mapOf("GTNO" to "3", "GATE" to "G33", "GTYP" to "I"), + mapOf("GTNO" to "5", "GATE" to "G23", "GTYP" to "D"), + ) + val state = com.gzzn.omms.msgexchange.domain.flight.FlightNextState( + flid = flid, + scalars = mapOf("FLNO" to "CA888"), + collections = mapOf("GTDT" to gates), + stateVersion = 1L, + lastMessageId = "msg-dnld-001", + ) + + repo.persistNextStates(day, listOf(state), snapshotReplace = true) + assertLegacyCollectionColumnsAllNull(flid) + + val roundtrip = repo.findNextStateByFlid(flid)!! + assertEquals(1L, roundtrip.stateVersion) + assertEquals("msg-dnld-001", roundtrip.lastMessageId) + // 明细表读回保序保源序号；未填属性以空串回填，按语义字段断言 + val roundtripGates = roundtrip.collections["GTDT"]!! + assertEquals(3, roundtripGates.size) + gates.forEachIndexed { i, expected -> + val actual = roundtripGates[i] + assertEquals(expected["GTNO"], actual["GTNO"]) + assertEquals(expected["GATE"], actual["GATE"]) + assertEquals(expected["GTYP"], actual["GTYP"]) + } + + val replaced = state.copy( + collections = mapOf("GTDT" to listOf(mapOf("GTNO" to "2", "GATE" to "Z9"))), + stateVersion = 2L, + lastMessageId = "msg-flop-002", + ) + repo.persistNextStates(null, listOf(replaced), snapshotReplace = false) + assertEquals("Z9", repo.findNextStateByFlid(flid)!!.collections["GTDT"]!![0]["GATE"]) + assertEquals(2L, repo.findNextStateByFlid(flid)!!.stateVersion) + val detailRows = ds.query( + "SELECT ordinal, source_seq, gate FROM flight_gate WHERE flid = ? ORDER BY ordinal", + { ps -> ps.setString(1, flid) }, + ) { rs -> Triple(rs.getInt("ordinal"), rs.getString("source_seq"), rs.getString("gate")) } + assertEquals(1, detailRows.size) + assertEquals(Triple(1, "2", "Z9"), detailRows[0]) + } + + @Test + fun `PG dialect - persistNextStates leaves all legacy collection storage columns null`() { + val day = "2026-09-07" + val flid = "TEST_V2_ALL_COLS" + val state = com.gzzn.omms.msgexchange.domain.flight.FlightNextState( + flid = flid, + scalars = mapOf("FLNO" to "MU123"), + collections = mapOf( + "GTDT" to listOf(mapOf("GTNO" to "1", "GATE" to "G1", "GTYP" to "D")), + "CKDT" to listOf(mapOf("CKNO" to "1", "CHKC" to "C1", "CTYP" to "Eco")), + "CLDT" to listOf(mapOf("CLNO" to "1", "BELT" to "B1", "BTYP" to "D")), + "DELY" to listOf(mapOf("DLNO" to "1", "CODE" to "01", "STRT" to "1200", "DURA" to "30")), + "ABTM" to listOf(mapOf("ASNO" to "1", "ABDG" to "B12", "ABOP" to "A", "AOTM" to "0730")), + "CHOT" to listOf(mapOf("CSNO" to "1", "CHID" to "ON", "CHTM" to "0800")), + "ROUT" to listOf(mapOf("RTNO" to "1", "APCD" to "CTU")), + ), + stateVersion = 1L, + lastMessageId = "msg-dnld-002", + ) + + repo.persistNextStates(day, listOf(state), snapshotReplace = true) + assertLegacyCollectionColumnsAllNull(flid) + assertEquals("MU123", repo.findByFlid(flid)!!["FLNO"]) + assertEquals(1, ds.query("SELECT COUNT(*) FROM flight_gate WHERE flid = ?", { ps -> ps.setString(1, flid) }) { rs -> rs.getInt(1) }.first()) + assertEquals(1, ds.query("SELECT COUNT(*) FROM flight_delay WHERE flid = ?", { ps -> ps.setString(1, flid) }) { rs -> rs.getInt(1) }.first()) + } + + @Test + fun `PG dialect - schd_gen stores last_message_id and rejects stale CAS without advancing`() { + val day = "2026-09-07" + + // 初始提交：last_message_id = "m1"，版本 1 + val ok1 = repo.putGenIfVersion( + day, 0L, + FlightSchdRepository.GenMeta(day, 1L, setOf("TEST_CAS_M"), lastMessageId = "m1"), + ) + assertTrue(ok1) + assertEquals("m1", repo.getGen(day)!!.lastMessageId) + + // 重放判定：以过期 expected=0 重试必须失败（同一消息的身份凭 getGen().lastMessageId 判定，不是靠版本相等） + val replayAttempt = repo.putGenIfVersion( + day, 0L, + FlightSchdRepository.GenMeta(day, 1L, setOf("TEST_CAS_M"), lastMessageId = "m1"), + ) + assertFalse(replayAttempt) + assertEquals(1L, repo.getGen(day)!!.version) + + // 正常推进 expected=1 → 版本 2，last_message_id 换为 "m2" + val ok2 = repo.putGenIfVersion( + day, 1L, + FlightSchdRepository.GenMeta(day, 2L, setOf("TEST_CAS_M"), lastMessageId = "m2"), + ) + assertTrue(ok2) + val gen = repo.getGen(day)!! + assertEquals(2L, gen.version) + assertEquals("m2", gen.lastMessageId) + } + + private fun assertLegacyCollectionColumnsAllNull(flid: String) { + val columnList = com.gzzn.omms.msgexchange.support.LEGACY_COLLECTION_STORAGE_COLUMNS.joinToString(", ") + ds.queryOne( + "SELECT $columnList FROM flight_schd WHERE flid = ?", + { ps -> ps.setString(1, flid) }, + ) { rs -> + com.gzzn.omms.msgexchange.support.LEGACY_COLLECTION_STORAGE_COLUMNS.forEach { column -> + assertNull(rs.getString(column), "legacy column $column must be null for v2 persistNextStates (flid=$flid)") + } + true + } + } }