PgConnection extends AbstractJdbcConnection { @Override public TransactionHelper getStrictTransactionHelper(String statusTableSchema, String statusTableName, Duration cleanupDuration) { return new PgPersistentTransactionHelper(statusTableSchema, statusTableName, cleanupDuration); } protected  PgConnection(Connection connection); @VisibleForTesting static PgConnection open(PgConnectionConfig config); @Override void executeReadOnlyQuery(String sql, Consumer<JdbcResultSet> resultHandler); @Override String escapeIdent(String ident); @Override TransactionHelper getStrictTransactionHelper(String statusTableSchema, String statusTableName, Duration cleanupDuration);  }
@Test public void txHelperPrepare() throws SQLException { TransactionHelper txHelper = pgConnection.getStrictTransactionHelper(null, "__digdag_status", Duration.ofDays(1)); UUID queryId = UUID.randomUUID(); txHelper.prepare(queryId); verify(pgConnection).execute(eq( "CREATE TABLE IF NOT EXISTS \"__digdag_status\"" + " (query_id text NOT NULL UNIQUE, created_at timestamptz NOT NULL, completed_at timestamptz)")); }
@Test public void txHelperLockedTransactionWithNotCompletedStatus() throws SQLException, LockConflictException { UUID queryId = UUID.randomUUID(); TransactionHelper txHelper = pgConnection.getStrictTransactionHelper(null, "__digdag_status", Duration.ofDays(1)); ResultSet selectRs = setupMockSelectResultSet(queryId); when(selectRs.next()).thenReturn(true); when(selectRs.wasNull()).thenReturn(true); AtomicBoolean called = new AtomicBoolean(false); assertThat(txHelper.lockedTransaction(queryId, () -> called.set(true)), is(true)); verify(pgConnection).execute(eq("BEGIN")); verify(pgConnection).execute(eq("UPDATE \"__digdag_status\" SET completed_at = CURRENT_TIMESTAMP WHERE query_id = '" + queryId + "'")); verify(pgConnection).execute(eq("COMMIT")); }
@Test public void txHelperLockedTransactionWithCompletedStatus() throws SQLException, LockConflictException { UUID queryId = UUID.randomUUID(); TransactionHelper txHelper = pgConnection.getStrictTransactionHelper(null, "__digdag_status", Duration.ofDays(1)); ResultSet selectRs = setupMockSelectResultSet(queryId); when(selectRs.next()).thenReturn(true); when(selectRs.wasNull()).thenReturn(false); AtomicBoolean called = new AtomicBoolean(false); assertThat(txHelper.lockedTransaction(queryId, () -> called.set(true)), is(false)); verify(pgConnection).execute(eq("BEGIN")); verify(pgConnection).execute(eq("ROLLBACK")); }