core/restream/restream_test.go

package restream

import (
	"fmt"
	"testing"
	"time"

	"github.com/datarhei/core/v16/ffmpeg"
	"github.com/datarhei/core/v16/internal/testhelper"
	"github.com/datarhei/core/v16/net"
	"github.com/datarhei/core/v16/restream/app"

	"github.com/stretchr/testify/require"
)

func getDummyRestreamer(portrange net.Portranger, validatorIn, validatorOut ffmpeg.Validator) (Restreamer, error) {
	binary, err := testhelper.BuildBinary("ffmpeg", "../internal/testhelper")
	if err != nil {
		return nil, fmt.Errorf("failed to build helper program: %w", err)
	}

	ffmpeg, err := ffmpeg.New(ffmpeg.Config{
		Binary:          binary,
		Portrange:       portrange,
		ValidatorInput:  validatorIn,
		ValidatorOutput: validatorOut,
	})
	if err != nil {
		return nil, err
	}

	rs, err := New(Config{
		FFmpeg: ffmpeg,
	})
	if err != nil {
		return nil, err
	}

	return rs, nil
}

func getDummyProcess() *app.Config {
	return &app.Config{
		ID: "process",
		Input: []app.ConfigIO{
			{
				ID:      "in",
				Address: "testsrc=size=1280x720:rate=25",
				Options: []string{
					"-f",
					"lavfi",
					"-re",
				},
			},
		},
		Output: []app.ConfigIO{
			{
				ID:      "out",
				Address: "-",
				Options: []string{
					"-codec",
					"copy",
					"-f",
					"null",
				},
			},
		},
		Options: []string{
			"-loglevel",
			"info",
		},
		Reconnect:      true,
		ReconnectDelay: 10,
		Autostart:      false,
		StaleTimeout:   0,
	}
}

func TestAddProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()
	require.NotNil(t, process)

	_, err = rs.GetProcess(process.ID)
	require.NotEqual(t, nil, err, "Unset process found (%s)", process.ID)

	err = rs.AddProcess(process)
	require.Equal(t, nil, err, "Failed to add process (%s)", err)

	_, err = rs.GetProcess(process.ID)
	require.Equal(t, nil, err, "Set process not found (%s)", process.ID)

	state, _ := rs.GetProcessState(process.ID)
	require.Equal(t, "stop", state.Order, "Process should be stopped")
}

func TestAutostartProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()
	process.Autostart = true

	rs.AddProcess(process)

	state, _ := rs.GetProcessState(process.ID)
	require.Equal(t, "start", state.Order, "Process should be started")

	rs.StopProcess(process.ID)
}

func TestAddInvalidProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	// Invalid process ID
	process := getDummyProcess()
	process.ID = ""

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process without ID")

	// Invalid input ID
	process = getDummyProcess()
	process.Input[0].ID = ""

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process input without ID")

	// Invalid input address
	process = getDummyProcess()
	process.Input[0].Address = ""

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process input without address")

	// Duplicate input ID
	process = getDummyProcess()
	process.Input = append(process.Input, process.Input[0])

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process input with duplicate ID")

	// No inputs
	process = getDummyProcess()
	process.Input = nil

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process without inputs")

	// Invalid output ID
	process = getDummyProcess()
	process.Output[0].ID = ""

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process output without ID")

	// Invalid output address
	process = getDummyProcess()
	process.Output[0].Address = ""

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process output without address")

	// Duplicate output ID
	process = getDummyProcess()
	process.Output = append(process.Output, process.Output[0])

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process output with duplicate ID")

	// No outputs
	process = getDummyProcess()
	process.Output = nil

	err = rs.AddProcess(process)
	require.NotEqual(t, nil, err, "Succeeded to add process without outputs")
}

func TestRemoveProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	err = rs.AddProcess(process)
	require.Equal(t, nil, err, "Failed to add process (%s)", err)

	err = rs.DeleteProcess(process.ID)
	require.Equal(t, nil, err, "Set process not found (%s)", process.ID)

	_, err = rs.GetProcess(process.ID)
	require.NotEqual(t, nil, err, "Unset process found (%s)", process.ID)
}

func TestGetProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	rs.AddProcess(process)

	_, err = rs.GetProcess(process.ID)
	require.Equal(t, nil, err, "Process not found (%s)", process.ID)

	list := rs.GetProcessIDs()
	require.Len(t, list, 1, "expected 1 process")
	require.Equal(t, process.ID, list[0], "expected same process ID")
}

func TestStartProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	rs.AddProcess(process)

	err = rs.StartProcess("foobar")
	require.NotEqual(t, nil, err, "shouldn't be able to start non-existing process")

	err = rs.StartProcess(process.ID)
	require.Equal(t, nil, err, "should be able to start existing process")

	state, _ := rs.GetProcessState(process.ID)
	require.Equal(t, "start", state.Order, "Process should be started")

	err = rs.StartProcess(process.ID)
	require.Equal(t, nil, err, "should be able to start already running process")

	state, _ = rs.GetProcessState(process.ID)
	require.Equal(t, "start", state.Order, "Process should be started")

	rs.StopProcess(process.ID)
}

func TestStopProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	rs.AddProcess(process)
	rs.StartProcess(process.ID)

	err = rs.StopProcess("foobar")
	require.NotEqual(t, nil, err, "shouldn't be able to stop non-existing process")

	err = rs.StopProcess(process.ID)
	require.Equal(t, nil, err, "should be able to stop existing running process")

	state, _ := rs.GetProcessState(process.ID)
	require.Equal(t, "stop", state.Order, "Process should be stopped")

	err = rs.StopProcess(process.ID)
	require.Equal(t, nil, err, "should be able to stop already stopped process")

	state, _ = rs.GetProcessState(process.ID)
	require.Equal(t, "stop", state.Order, "Process should be stopped")
}

func TestRestartProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	rs.AddProcess(process)

	err = rs.RestartProcess("foobar")
	require.NotEqual(t, nil, err, "shouldn't be able to restart non-existing process")

	err = rs.RestartProcess(process.ID)
	require.Equal(t, nil, err, "should be able to restart existing stopped process")

	state, _ := rs.GetProcessState(process.ID)
	require.Equal(t, "stop", state.Order, "Process should be stopped")

	rs.StartProcess(process.ID)

	state, _ = rs.GetProcessState(process.ID)
	require.Equal(t, "start", state.Order, "Process should be started")

	rs.StopProcess(process.ID)
}

func TestReloadProcess(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	rs.AddProcess(process)

	err = rs.ReloadProcess("foobar")
	require.NotEqual(t, nil, err, "shouldn't be able to reload non-existing process")

	err = rs.ReloadProcess(process.ID)
	require.Equal(t, nil, err, "should be able to reload existing stopped process")

	state, _ := rs.GetProcessState(process.ID)
	require.Equal(t, "stop", state.Order, "Process should be stopped")

	rs.StartProcess(process.ID)

	state, _ = rs.GetProcessState(process.ID)
	require.Equal(t, "start", state.Order, "Process should be started")

	err = rs.ReloadProcess(process.ID)
	require.Equal(t, nil, err, "should be able to reload existing process")

	state, _ = rs.GetProcessState(process.ID)
	require.Equal(t, "start", state.Order, "Process should be started")

	rs.StopProcess(process.ID)
}

func TestProcessData(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	rs.AddProcess(process)

	data, _ := rs.GetProcessMetadata(process.ID, "foobar")
	require.Equal(t, nil, data, "nothing should be stored under the key")

	rs.SetProcessMetadata(process.ID, "foobar", process)

	data, _ = rs.GetProcessMetadata(process.ID, "foobar")
	require.NotEqual(t, nil, data, "there should be something stored under the key")

	p := data.(*app.Config)

	require.Equal(t, process.ID, p.ID, "failed to retrieve stored data")
}

func TestLog(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	rs.AddProcess(process)

	_, err = rs.GetProcessLog("foobar")
	require.NotEqual(t, nil, err, "shouldn't be able to get log from non-existing process")

	log, err := rs.GetProcessLog(process.ID)
	require.Equal(t, nil, err, "should be able to get log from existing process")
	require.Equal(t, 0, len(log.Prelude))
	require.Equal(t, 0, len(log.Log))

	rs.StartProcess(process.ID)

	time.Sleep(3 * time.Second)

	log, _ = rs.GetProcessLog(process.ID)

	require.NotEqual(t, 0, len(log.Prelude))
	require.NotEqual(t, 0, len(log.Log))

	rs.StopProcess(process.ID)

	log, _ = rs.GetProcessLog(process.ID)

	require.NotEqual(t, 0, len(log.Prelude))
	require.NotEqual(t, 0, len(log.Log))
}

func TestPlayoutNoRange(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	process.Input[0].Address = "playout:" + process.Input[0].Address

	rs.AddProcess(process)

	_, err = rs.GetPlayout("foobar", process.Input[0].ID)
	require.NotEqual(t, nil, err, "playout of non-existing process should error")

	_, err = rs.GetPlayout(process.ID, "foobar")
	require.NotEqual(t, nil, err, "playout of non-existing input should error")

	addr, _ := rs.GetPlayout(process.ID, process.Input[0].ID)
	require.Equal(t, 0, len(addr), "the playout address should be empty if no port range is given")
}

func TestPlayoutRange(t *testing.T) {
	portrange, err := net.NewPortrange(3000, 3001)
	require.NoError(t, err)

	rs, err := getDummyRestreamer(portrange, nil, nil)
	require.NoError(t, err)

	process := getDummyProcess()

	process.Input[0].Address = "playout:" + process.Input[0].Address

	rs.AddProcess(process)

	_, err = rs.GetPlayout("foobar", process.Input[0].ID)
	require.NotEqual(t, nil, err, "playout of non-existing process should error")

	_, err = rs.GetPlayout(process.ID, "foobar")
	require.NotEqual(t, nil, err, "playout of non-existing input should error")

	addr, _ := rs.GetPlayout(process.ID, process.Input[0].ID)
	require.NotEqual(t, 0, len(addr), "the playout address should not be empty if a port range is given")
	require.Equal(t, "127.0.0.1:3000", addr, "the playout address should be 127.0.0.1:3000")
}

func TestAddressReference(t *testing.T) {
	rs, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	process1 := getDummyProcess()
	process2 := getDummyProcess()

	process2.ID = "process2"

	rs.AddProcess(process1)

	process2.Input[0].Address = "#process:foobar=out"

	err = rs.AddProcess(process2)
	require.NotEqual(t, nil, err, "shouldn't resolve invalid reference")

	process2.Input[0].Address = "#process2:output=out"

	err = rs.AddProcess(process2)
	require.NotEqual(t, nil, err, "shouldn't resolve invalid reference")

	process2.Input[0].Address = "#process:output=foobar"

	err = rs.AddProcess(process2)
	require.NotEqual(t, nil, err, "shouldn't resolve invalid reference")

	process2.Input[0].Address = "#process:output=out"

	err = rs.AddProcess(process2)
	require.Equal(t, nil, err, "should resolve reference")
}

func TestConfigValidation(t *testing.T) {
	rsi, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	rs := rsi.(*restream)

	config := getDummyProcess()

	_, err = rs.validateConfig(config)
	require.NoError(t, err)

	config.Input = []app.ConfigIO{}
	_, err = rs.validateConfig(config)
	require.Error(t, err)

	config = getDummyProcess()
	config.Input[0].ID = ""
	_, err = rs.validateConfig(config)
	require.Error(t, err)

	config = getDummyProcess()
	config.Input[0].Address = ""
	_, err = rs.validateConfig(config)
	require.Error(t, err)

	config = getDummyProcess()
	config.Output = []app.ConfigIO{}
	_, err = rs.validateConfig(config)
	require.Error(t, err)

	config = getDummyProcess()
	config.Output[0].ID = ""
	_, err = rs.validateConfig(config)
	require.Error(t, err)

	config = getDummyProcess()
	config.Output[0].Address = ""
	_, err = rs.validateConfig(config)
	require.Error(t, err)
}

func TestConfigValidationFFmpeg(t *testing.T) {
	valIn, err := ffmpeg.NewValidator([]string{"^https?://"}, nil)
	require.NoError(t, err)

	valOut, err := ffmpeg.NewValidator([]string{"^https?://", "^rtmp://"}, nil)
	require.NoError(t, err)

	rsi, err := getDummyRestreamer(nil, valIn, valOut)
	require.NoError(t, err)

	rs := rsi.(*restream)

	config := getDummyProcess()

	_, err = rs.validateConfig(config)
	require.Error(t, err)

	config.Input[0].Address = "http://stream.example.com/master.m3u8"
	config.Output[0].Address = "http://stream.example.com/master2.m3u8"

	_, err = rs.validateConfig(config)
	require.NoError(t, err)

	config.Output[0].Address = "[f=flv]http://stream.example.com/master2.m3u8"
	_, err = rs.validateConfig(config)
	require.NoError(t, err)

	config.Output[0].Address = "[f=hls]http://stream.example.com/master2.m3u8|[f=flv]rtmp://stream.example.com/stream"
	_, err = rs.validateConfig(config)
	require.NoError(t, err)
}

func TestOutputAddressValidation(t *testing.T) {
	rsi, err := getDummyRestreamer(nil, nil, nil)
	require.NoError(t, err)

	rs := rsi.(*restream)

	type res struct {
		path string
		err  bool
	}

	paths := map[string]res{
		"/dev/null":                            {"file:/dev/null", false},
		"/dev/../etc/passwd":                   {"/etc/passwd", true},
		"/dev/fb0":                             {"file:/dev/fb0", false},
		"/etc/passwd":                          {"/etc/passwd", true},
		"/core/data/../../etc/passwd":          {"/etc/passwd", true},
		"/core/data/./etc/passwd":              {"file:/core/data/etc/passwd", false},
		"file:/core/data/foobar":               {"file:/core/data/foobar", false},
		"http://example.com":                   {"http://example.com", false},
		"-":                                    {"pipe:", false},
		"/core/data/foobar|http://example.com": {"file:/core/data/foobar|http://example.com", false},
		"/core/data/foobar|/etc/passwd":        {"/core/data/foobar|/etc/passwd", true},
		"[f=mpegts]udp://10.0.1.255:1234/":     {"[f=mpegts]udp://10.0.1.255:1234/", false},
		"[f=null]-|[f=null]-":                  {"[f=null]pipe:|[f=null]pipe:", false},
		"[onfail=ignore]/core/data/archive-20121107.mkv|[f=mpegts]udp://10.0.1.255:1234/": {"[onfail=ignore]file:/core/data/archive-20121107.mkv|[f=mpegts]udp://10.0.1.255:1234/", false},
	}

	for path, r := range paths {
		path, _, err := rs.validateOutputAddress(path, "/core/data")

		if r.err {
			require.Error(t, err)
		} else {
			require.NoError(t, err)
		}

		require.Equal(t, r.path, path)
	}
}