From f3f53fc7ad4f7d8412ff6a77c0a0ed8975182c1a Mon Sep 17 00:00:00 2001 From: Dmitry Sergeev Date: Wed, 2 Sep 2026 00:06:02 +0300 Subject: [PATCH] end-to-end controller tests. --- feeds-uuid.txt | 8 + internal/playback/playlist_pipeline_test.go | 175 +++++++++++++++++++ internal/playback/session_controller_test.go | 32 ++++ 3 files changed, 215 insertions(+) create mode 100644 feeds-uuid.txt create mode 100644 internal/playback/playlist_pipeline_test.go diff --git a/feeds-uuid.txt b/feeds-uuid.txt new file mode 100644 index 0000000..01abc95 --- /dev/null +++ b/feeds-uuid.txt @@ -0,0 +1,8 @@ +5fbec3b1-1b0f-417d-9059-8b94a47197ed +5fbec3b1-1b0f-417d-9059-8b94a47197ec + +5fbec3b1-1b0f-417d-9059-8b94a47197ef +5fbec3b1-1b0f-417d-9059-8b94a47197eb + +2618979d-76a5-45e0-83cb-0f192978d1cd +9d2a041b-01cf-4ee4-bffa-188fe093c99b diff --git a/internal/playback/playlist_pipeline_test.go b/internal/playback/playlist_pipeline_test.go new file mode 100644 index 0000000..db5ca03 --- /dev/null +++ b/internal/playback/playlist_pipeline_test.go @@ -0,0 +1,175 @@ +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 *fakePlaylistReadinessTicker + 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 PlaylistReadiness, 8) + session := &fakeSessionSnapshotSource{} + statuses := &fakePlaybackStatusSnapshotSource{} + coordinator, err := NewPlaylistReadinessCoordinator( + controller, session, statuses, events, time.Millisecond, + ) + if err != nil { + t.Fatal(err) + } + ticker := newFakePlaylistReadinessTicker() + coordinator.newTicker = func(time.Duration) playlistReadinessTicker { 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") + } + } +} diff --git a/internal/playback/session_controller_test.go b/internal/playback/session_controller_test.go index f0468fc..e2619db 100644 --- a/internal/playback/session_controller_test.go +++ b/internal/playback/session_controller_test.go @@ -344,6 +344,38 @@ func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) { } } +func TestSessionControllerStopAllWhileSessionIsActive(t *testing.T) { + events := make(chan controllerEvent, 32) + controller := newRecordingController(t, events) + initial := validCommandSession() + initial.SyncRequested = false + commands := make(chan SessionCommand) + done := make(chan error, 1) + go func() { done <- controller.Run(context.Background(), initial, commands) }() + receiveIndependentStarts(t, events) + + commands <- SessionCommand{Kind: CommandStopAll} + stopped := map[Unit]bool{} + for len(stopped) < 2 { + event := receiveControllerEvent(t, events) + if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) { + t.Fatalf("unexpected stop event: %+v", event) + } + stopped[event.unit] = true + } + snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool { + return snapshot.Plan.Topology == TopologyIdle + }) + if snapshot.Desired.Video.Active || snapshot.Desired.Audio.Active { + t.Fatalf("stopped desired session = %#v", snapshot.Desired) + } + + close(commands) + if err := <-done; err != nil { + t.Fatalf("Run() error = %v", err) + } +} + func TestSessionControllerSnapshotTracksDesiredPlanAndGeneration(t *testing.T) { events := make(chan controllerEvent, 64) controller := newRecordingController(t, events)