Function TestResumableStepAtCursor

rivertest/resumable_test.go:51–78

(t *testing.T)

}

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

	type testCursor struct {
		LastProcessedID int `json:"last_processed_id"`
	}

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

		opts := &river.InsertOpts{}
		ResumableStepAtCursor(opts, "process_ids", testCursor{LastProcessedID: 42})

		var metadata map[string]json.RawMessage
		require.NoError(t, json.Unmarshal(opts.Metadata, &metadata))

		var step string
		require.NoError(t, json.Unmarshal(metadata[rivercommon.MetadataKeyResumableStep], &step))
		require.Equal(t, "process_ids", step)

		var cursors map[string]json.RawMessage
		require.NoError(t, json.Unmarshal(metadata[rivercommon.MetadataKeyResumableCursor], &cursors))

		var cursor testCursor
		require.NoError(t, json.Unmarshal(cursors["process_ids"], &cursor))
		require.Equal(t, 42, cursor.LastProcessedID)
	})
}

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