clipper/backend/media/get_segment_test.go

package media_test

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"testing"

	"git.netflux.io/rob/clipper/config"
	"git.netflux.io/rob/clipper/generated/mocks"
	"git.netflux.io/rob/clipper/generated/store"
	"git.netflux.io/rob/clipper/media"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v4"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap"
)

func fixtureReader(t *testing.T, fixturePath string, limit int64) io.ReadCloser {
	fptr, err := os.Open(fixturePath)
	require.NoError(t, err)

	// limitReader to make the mock work realistically, not intended for assertions:
	return struct {
		io.Reader
		io.Closer
	}{
		Reader: io.LimitReader(fptr, limit),
		Closer: fptr,
	}
}

func helperCommand(t *testing.T, wantCommand, stdoutFile, stderrString string, forceExitCode int) media.CommandFunc {
	return func(ctx context.Context, name string, args ...string) *exec.Cmd {
		cs := []string{"-test.run=TestHelperProcess", "--", name}
		cs = append(cs, args...)
		cmd := exec.CommandContext(ctx, os.Args[0], cs...)
		cmd.Env = []string{
			"GO_WANT_HELPER_PROCESS=1",
			"GO_WANT_COMMAND=" + wantCommand,
			"GO_STDOUT_FILE=" + stdoutFile,
			"GO_STDERR_STRING=" + stderrString,
			"GO_FORCE_EXIT_CODE=" + strconv.Itoa(forceExitCode),
		}
		return cmd
	}
}

func TestHelperProcess(t *testing.T) {
	if os.Getenv("GO_WANT_HELPER_PROCESS") != "1" {
		return
	}

	defer func() {
		// Stop the helper process writing to stdout after the test has finished.
		// This prevents it from writing the "PASS" string which is unwanted in
		// this context.
		if !t.Failed() {
			os.Stdout, _ = os.Open(os.DevNull)
		}
	}()

	if exitCode := os.Getenv("GO_FORCE_EXIT_CODE"); exitCode != "0" {
		c, _ := strconv.Atoi(exitCode)
		os.Stderr.WriteString(os.Getenv("GO_STDERR_STRING"))
		os.Exit(c)
	}

	if wantCommand := os.Getenv("GO_WANT_COMMAND"); wantCommand != "" {
		gotCmd := strings.Split(strings.Join(os.Args, " "), " -- ")[1]
		if wantCommand != gotCmd {
			fmt.Printf("GO_WANT_COMMAND assertion failed:\nwant = %v\ngot =  %v", wantCommand, gotCmd)
			return
		}
	}

	// Copy stdin to /dev/null. This is required to avoid broken pipe errors in
	// the tests:
	_, err := io.Copy(io.Discard, os.Stdin)
	require.NoError(t, err)

	// If an output file is provided, then copy that to stdout:
	if fname := os.Getenv("GO_STDOUT_FILE"); fname != "" {
		fptr, err := os.Open(fname)
		require.NoError(t, err)

		_, err = io.Copy(os.Stdout, fptr)
		require.NoError(t, err)
	}
}

func TestGetSegment(t *testing.T) {
	mediaSetID := uuid.MustParse("4c440241-cca9-436f-adb0-be074588cf2b")
	const inFixturePath = "testdata/tone-44100-stereo-int16-30000ms.raw"

	t.Run("invalid range", func(t *testing.T) {
		var mockStore mocks.Store
		var fileStore mocks.FileStore
		service := media.NewMediaSetService(&mockStore, nil, &fileStore, nil, config.Config{}, zap.NewNop().Sugar())

		stream, err := service.GetAudioSegment(context.Background(), mediaSetID, 1, 0, media.AudioFormatMP3)
		require.Nil(t, stream)
		require.EqualError(t, err, "invalid range")
	})

	t.Run("error fetching media set", func(t *testing.T) {
		var mockStore mocks.Store
		mockStore.On("GetMediaSet", mock.Anything, mediaSetID).Return(store.MediaSet{}, pgx.ErrNoRows)
		var fileStore mocks.FileStore
		service := media.NewMediaSetService(&mockStore, nil, &fileStore, nil, config.Config{}, zap.NewNop().Sugar())

		stream, err := service.GetAudioSegment(context.Background(), mediaSetID, 0, 1, media.AudioFormatMP3)
		require.Nil(t, stream)
		require.EqualError(t, err, "error getting media set: no rows in result set")
	})

	t.Run("error fetching audio data", func(t *testing.T) {
		mediaSet := store.MediaSet{ID: mediaSetID, AudioChannels: 2}

		var mockStore mocks.Store
		mockStore.On("GetMediaSet", mock.Anything, mediaSetID).Return(mediaSet, nil)

		var fileStore mocks.FileStore
		fileStore.On("GetObjectWithRange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
			Return(nil, errors.New("network error"))

		service := media.NewMediaSetService(&mockStore, nil, &fileStore, nil, config.Config{}, zap.NewNop().Sugar())

		stream, err := service.GetAudioSegment(context.Background(), mediaSetID, 0, 1, media.AudioFormatMP3)
		require.Nil(t, stream)
		require.EqualError(t, err, "error getting object from store: network error")
	})

	t.Run("ffmpeg returns non-zero error code", func(t *testing.T) {
		mediaSet := store.MediaSet{ID: mediaSetID, AudioChannels: 2}

		var mockStore mocks.Store
		mockStore.On("GetMediaSet", mock.Anything, mediaSetID).Return(mediaSet, nil)

		var fileStore mocks.FileStore
		fileStore.On("GetObjectWithRange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
			Return(fixtureReader(t, inFixturePath, 1), nil)

		cmd := helperCommand(t, "", "", "something bad happened", 2)
		service := media.NewMediaSetService(&mockStore, nil, &fileStore, cmd, config.Config{}, zap.NewNop().Sugar())

		stream, err := service.GetAudioSegment(context.Background(), mediaSetID, 0, 1, media.AudioFormatMP3)
		require.NoError(t, err)

		_, err = stream.Next(context.Background())
		require.EqualError(t, err, "error waiting for ffmpeg: exit status 2, output: something bad happened")

	})

	testCases := []struct {
		name                       string
		audioFormat                media.AudioFormat
		audioChannels              int32
		inStartFrame, inEndFrame   int64
		wantStartByte, wantEndByte int64
		outFixturePath             string
		wantCommand                string
		wantOutput                 string
	}{
		{
			name:           "mono to mp3",
			audioFormat:    media.AudioFormatMP3,
			audioChannels:  1,
			inStartFrame:   500,
			inEndFrame:     2_000,
			wantStartByte:  1_000,
			wantEndByte:    4_000,
			outFixturePath: "testdata/fake.mp3",
			wantCommand:    "ffmpeg -hide_banner -loglevel error -f s16le -ac 1 -ar 48000 -i - -f mp3 -",
			wantOutput:     "this is a fake mp3",
		},
		{
			name:           "stereo to mp3",
			audioFormat:    media.AudioFormatMP3,
			audioChannels:  2,
			inStartFrame:   0,
			inEndFrame:     1_323_000,
			wantStartByte:  0,
			wantEndByte:    5_292_000,
			outFixturePath: "testdata/fake.mp3",
			wantCommand:    "ffmpeg -hide_banner -loglevel error -f s16le -ac 2 -ar 48000 -i - -f mp3 -",
			wantOutput:     "this is a fake mp3",
		},
		{
			name:           "mono to wav",
			audioFormat:    media.AudioFormatWAV,
			audioChannels:  1,
			inStartFrame:   16_384,
			inEndFrame:     32_768,
			wantStartByte:  32_768,
			wantEndByte:    65_536,
			outFixturePath: "testdata/fake.wav",
			wantCommand:    "ffmpeg -hide_banner -loglevel error -f s16le -ac 1 -ar 48000 -i - -f wav -",
			wantOutput:     "this is a fake wav",
		},
		{
			name:           "stereo to wav",
			audioFormat:    media.AudioFormatWAV,
			audioChannels:  2,
			inStartFrame:   2_048,
			inEndFrame:     4_096,
			wantStartByte:  8_192,
			wantEndByte:    16_384,
			outFixturePath: "testdata/fake.wav",
			wantCommand:    "ffmpeg -hide_banner -loglevel error -f s16le -ac 2 -ar 48000 -i - -f wav -",
			wantOutput:     "this is a fake wav",
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			ctx := context.Background()

			mediaSet := store.MediaSet{ID: mediaSetID, AudioChannels: tc.audioChannels}
			var mockStore mocks.Store
			mockStore.On("GetMediaSet", mock.Anything, mediaSetID).Return(mediaSet, nil)
			defer mockStore.AssertExpectations(t)

			var fileStore mocks.FileStore
			fileStore.
				On("GetObjectWithRange", mock.Anything, "media_sets/4c440241-cca9-436f-adb0-be074588cf2b/audio.raw", tc.wantStartByte, tc.wantEndByte).
				Return(fixtureReader(t, inFixturePath, tc.wantEndByte-tc.wantStartByte), nil)
			defer fileStore.AssertExpectations(t)

			cmd := helperCommand(t, tc.wantCommand, tc.outFixturePath, "", 0)
			service := media.NewMediaSetService(&mockStore, nil, &fileStore, cmd, config.Config{}, zap.NewNop().Sugar())

			stream, err := service.GetAudioSegment(ctx, mediaSetID, tc.inStartFrame, tc.inEndFrame, tc.audioFormat)
			require.NoError(t, err)

			var data bytes.Buffer
			var lastPercentComplete float32
			var progress media.AudioSegmentProgress
			for {
				progress, err = stream.Next(ctx)
				if err == io.EOF {
					break
				}

				require.NoError(t, err)
				assert.GreaterOrEqual(t, progress.PercentComplete, lastPercentComplete)
				lastPercentComplete = progress.PercentComplete
				data.Write(progress.Data)
			}

			assert.Equal(t, tc.wantOutput, data.String())
			assert.Equal(t, float32(100), lastPercentComplete)
		})
	}
}