Function Test_Client_PilotUsage

client_pilot_test.go:133–337

(t *testing.T)

}

func Test_Client_PilotUsage(t *testing.T) {
	t.Parallel()

	ctx := context.Background()

	setup := func(t *testing.T, configMutate func(*Config)) (*Client[pgx.Tx], *pilotSpy) {
		t.Helper()

		var (
			dbPool = riversharedtest.DBPool(ctx, t)
			driver = riverpgxv5.New(dbPool)
			schema = riverdbtest.TestSchema(ctx, t, driver, nil)
			config = newTestConfig(t, schema)
		)

		if configMutate != nil {
			configMutate(config)
		}

		pilot := &pilotSpy{}
		pluginDriver := newDriverWithPlugin(t, dbPool)
		pluginDriver.pilot = pilot

		client, err := NewClient(pluginDriver, config)
		require.NoError(t, err)

		return client, pilot
	}

	withClientTx := func(t *testing.T, client *Client[pgx.Tx], callback func(tx pgx.Tx)) {
		t.Helper()

		exec := client.Driver().GetExecutor()
		execTx, err := exec.Begin(ctx)
		require.NoError(t, err)

		committed := false
		t.Cleanup(func() {
			if !committed {
				_ = execTx.Rollback(ctx)
			}
		})

		tx := client.Driver().UnwrapTx(execTx)
		callback(tx)

		require.NoError(t, execTx.Commit(ctx))
		committed = true
	}

	t.Run("InitUsesPilot", func(t *testing.T) {
		t.Parallel()

		client, pilot := setup(t, nil)
		require.NotNil(t, client)
		require.Equal(t, int64(1), pilot.jobCleanerQueuesExcludedCalls.Load())
		require.Equal(t, int64(1), pilot.pilotInitCalls.Load())
	})