go-mxl-player/internal/playback/playlist_pipeline_test.go

package playback

import (
	"context"
	"errors"
	"testing"
	"time"
)

func TestPlaylistFailurePipeline(t *testing.T) {
	for _, test := range []struct {
		name        string
		policy      PlaylistFailurePolicy
		wantAdvance bool
	}{
		{name: "wait retains failed entry", policy: PlaylistFailureWait},
		{name: "next advances failed entry", policy: PlaylistFailureNext, wantAdvance: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			playlist := Playlist{
				OnFailure: test.policy,
				Entries: []PlaylistEntry{
					{
						Name:  "independent",
						Video: PlaylistFeed{Domain: "/video", UUID: "video"},
						Audio: PlaylistFeed{Domain: "/audio", UUID: "audio"},
					},
					{Name: "next", Video: PlaylistFeed{Domain: "/video", UUID: "next"}},
				},
			}
			h := startPlaylistPipeline(t, playlist)
			defer h.stop(t)

			h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
			selected := receivePlaylistSession(t, h.sessions).Session
			waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
				return snapshot.Revision == 1
			})
			h.setSession(selected, 4)
			failureErr := errors.New("audio retries exhausted")
			h.statuses.set(PlaybackStatusSnapshot{
				Generation: 4,
				Audio: Status{
					Unit: UnitAudio, State: StateFailed, Generation: 4,
					Feed: selected.Audio, Attempt: 3, FailedAttempts: 3, Err: failureErr,
				},
				HasAudio: true,
			})
			h.ticker.tick()

			if test.wantAdvance {
				next := receivePlaylistSession(t, h.sessions)
				if next.Session.Video.UUID != "next" || next.Session.Audio.IsConfigured() {
					t.Fatalf("advanced session = %#v", next.Session)
				}
			} else {
				select {
				case command := <-h.sessions:
					t.Fatalf("wait policy advanced with %#v", command)
				case <-time.After(20 * time.Millisecond):
				}
			}

			snapshot := waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
				return snapshot.HasFailure
			})
			if snapshot.Failure.Status.Unit != UnitAudio ||
				!errors.Is(snapshot.Failure.Status.Err, failureErr) {
				t.Fatalf("failure snapshot = %#v", snapshot.Failure)
			}
			wantIndex := 0
			if test.wantAdvance {
				wantIndex = 1
			}
			if snapshot.State.CurrentIndex != wantIndex {
				t.Fatalf("current index = %d, want %d", snapshot.State.CurrentIndex, wantIndex)
			}
		})
	}
}

func TestPlaylistManualSelectionWhileFeedIsReconnecting(t *testing.T) {
	playlist := navigationPlaylist(false)
	h := startPlaylistPipeline(t, playlist)
	defer h.stop(t)

	h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
	selected := receivePlaylistSession(t, h.sessions).Session
	waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
		return snapshot.Revision == 1
	})
	h.setSession(selected, 2)
	h.statuses.set(PlaybackStatusSnapshot{
		Generation: 2,
		Video: Status{
			Unit: UnitVideo, State: StateReconnecting, Generation: 2,
			Feed: selected.Video, Attempt: 2, FailedAttempts: 1,
		},
		HasVideo: true,
	})
	h.ticker.tick()

	h.commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 1}
	next := receivePlaylistSession(t, h.sessions)
	if next.Session.Audio.UUID != "audio-2" || next.Session.Video.IsConfigured() {
		t.Fatalf("manual selection session = %#v", next.Session)
	}
	waitForPlaylistSnapshot(t, h.controller, func(snapshot PlaylistSnapshot) bool {
		return snapshot.Revision == 2 && snapshot.State.CurrentIndex == 1
	})
}

type playlistPipelineHarness struct {
	controller *PlaylistController
	commands   chan PlaylistCommand
	sessions   chan SessionCommand
	session    *fakeSessionSnapshotSource
	statuses   *fakePlaybackStatusSnapshotSource
	ticker     *fakePlaylistEventTicker
	cancel     context.CancelFunc
	results    chan error
}

func startPlaylistPipeline(t *testing.T, playlist Playlist) *playlistPipelineHarness {
	t.Helper()
	sessions := make(chan SessionCommand, 8)
	controller, err := NewPlaylistController(playlist, validPlaylistRetryPolicy(), sessions)
	if err != nil {
		t.Fatal(err)
	}
	events := make(chan PlaylistEvent, 8)
	session := &fakeSessionSnapshotSource{}
	statuses := &fakePlaybackStatusSnapshotSource{}
	coordinator, err := NewPlaylistEventCoordinator(
		controller, session, statuses, events, time.Millisecond,
	)
	if err != nil {
		t.Fatal(err)
	}
	ticker := newFakePlaylistEventTicker()
	coordinator.newTicker = func(time.Duration) playlistEventTicker { return ticker }
	commands := make(chan PlaylistCommand, 8)
	ctx, cancel := context.WithCancel(context.Background())
	results := make(chan error, 2)
	go func() { results <- controller.Run(ctx, commands, events) }()
	go func() { results <- coordinator.Run(ctx) }()
	return &playlistPipelineHarness{
		controller: controller, commands: commands, sessions: sessions,
		session: session, statuses: statuses, ticker: ticker,
		cancel: cancel, results: results,
	}
}

func (h *playlistPipelineHarness) setSession(desired SessionConfig, generation uint64) {
	plan, err := BuildSessionPlan(desired, func(FeedConfig, FeedConfig) bool { return false })
	if err != nil {
		panic(err)
	}
	h.session.set(SessionSnapshot{Desired: desired, Plan: plan, Generation: generation}, true)
}

func (h *playlistPipelineHarness) stop(t *testing.T) {
	t.Helper()
	h.cancel()
	for range 2 {
		select {
		case err := <-h.results:
			if !errors.Is(err, context.Canceled) {
				t.Fatalf("pipeline Run() error = %v", err)
			}
		case <-time.After(time.Second):
			t.Fatal("playlist pipeline did not stop")
		}
	}
}