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") } } }