package playback import ( "context" "errors" "sync" "testing" "time" ) type stubVideoSlot struct{} func (stubVideoSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error { return nil } type stubAudioSlot struct{} func (stubAudioSlot) Run(context.Context, FeedConfig, <-chan FeedConfig) error { return nil } type stubSyncSlot struct{} func (stubSyncSlot) Run(context.Context, SyncPairConfig, <-chan SyncPairConfig) error { return nil } func TestNewSessionControllerValidatesSlots(t *testing.T) { video := stubVideoSlot{} audio := stubAudioSlot{} sync := stubSyncSlot{} tests := []struct { name string video VideoSlotRunner audio AudioSlotRunner sync SyncSlotRunner want error }{ {"missing video", nil, audio, sync, ErrVideoSlotRequired}, {"missing audio", video, nil, sync, ErrAudioSlotRequired}, {"missing sync", video, audio, nil, ErrSyncSlotRequired}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { controller, err := NewSessionController(tt.video, tt.audio, tt.sync, nil) if controller != nil { t.Fatalf("NewSessionController() controller = %#v, want nil", controller) } if !errors.Is(err, tt.want) { t.Fatalf("NewSessionController() error = %v, want %v", err, tt.want) } }) } } func TestNewSessionControllerAllowsUnavailableSynchronization(t *testing.T) { video := stubVideoSlot{} audio := stubAudioSlot{} sync := stubSyncSlot{} controller, err := NewSessionController(video, audio, sync, nil) if err != nil { t.Fatalf("NewSessionController() error = %v", err) } if controller == nil { t.Fatal("NewSessionController() controller is nil") } if controller.videoSlot != video || controller.audioSlot != audio || controller.syncSlot != sync { t.Fatalf("NewSessionController() = %#v", controller) } if controller.canSync != nil { t.Fatal("nil sync predicate was not preserved") } } func TestNewSessionControllerStoresSyncPredicate(t *testing.T) { predicate := func(FeedConfig, FeedConfig) bool { return true } controller, err := NewSessionController( stubVideoSlot{}, stubAudioSlot{}, stubSyncSlot{}, predicate, ) if err != nil { t.Fatalf("NewSessionController() error = %v", err) } if controller.canSync == nil || !controller.canSync(FeedConfig{}, FeedConfig{}) { t.Fatal("sync predicate was not stored") } } type controllerEvent struct { unit Unit action string generation uint64 feed FeedConfig pair SyncPairConfig } type recordingVideoSlot struct{ events chan<- controllerEvent } func (s recordingVideoSlot) Run( ctx context.Context, initial FeedConfig, commands <-chan FeedConfig, ) error { s.events <- controllerEvent{ unit: UnitVideo, action: "start", generation: generationFromContext(ctx), feed: initial, } for { select { case config := <-commands: s.events <- controllerEvent{unit: UnitVideo, action: "command", feed: config} case <-ctx.Done(): s.events <- controllerEvent{unit: UnitVideo, action: "stop"} return ctx.Err() } } } type recordingAudioSlot struct{ events chan<- controllerEvent } func (s recordingAudioSlot) Run( ctx context.Context, initial FeedConfig, commands <-chan FeedConfig, ) error { s.events <- controllerEvent{ unit: UnitAudio, action: "start", generation: generationFromContext(ctx), feed: initial, } for { select { case config := <-commands: s.events <- controllerEvent{unit: UnitAudio, action: "command", feed: config} case <-ctx.Done(): s.events <- controllerEvent{unit: UnitAudio, action: "stop"} return ctx.Err() } } } type recordingSyncSlot struct{ events chan<- controllerEvent } func (s recordingSyncSlot) Run( ctx context.Context, initial SyncPairConfig, commands <-chan SyncPairConfig, ) error { s.events <- controllerEvent{ unit: UnitSync, action: "start", generation: generationFromContext(ctx), pair: initial, } for { select { case config := <-commands: s.events <- controllerEvent{unit: UnitSync, action: "command", pair: config} case <-ctx.Done(): s.events <- controllerEvent{unit: UnitSync, action: "stop"} return ctx.Err() } } } func newRecordingController(t *testing.T, events chan<- controllerEvent) *SessionController { t.Helper() controller, err := NewSessionController( recordingVideoSlot{events}, recordingAudioSlot{events}, recordingSyncSlot{events}, func(FeedConfig, FeedConfig) bool { return true }, ) if err != nil { t.Fatal(err) } return controller } func receiveControllerEvent(t *testing.T, events <-chan controllerEvent) controllerEvent { t.Helper() select { case event := <-events: return event case <-time.After(time.Second): t.Fatal("controller event timed out") return controllerEvent{} } } func receiveIndependentStarts(t *testing.T, events <-chan controllerEvent) { t.Helper() seen := map[Unit]bool{} for len(seen) < 2 { event := receiveControllerEvent(t, events) if event.action != "start" || (event.unit != UnitVideo && event.unit != UnitAudio) { t.Fatalf("unexpected initial event: %+v", event) } seen[event.unit] = true } } func waitControllerSnapshot( t *testing.T, controller *SessionController, match func(SessionSnapshot) bool, ) SessionSnapshot { t.Helper() deadline := time.Now().Add(time.Second) for { if snapshot, ok := controller.Snapshot(); ok && match(snapshot) { return snapshot } if time.Now().After(deadline) { snapshot, ok := controller.Snapshot() t.Fatalf("snapshot timed out: %#v, available=%t", snapshot, ok) } time.Sleep(time.Millisecond) } } func TestSessionControllerUpdatesOnlyChangedIndependentSlot(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) want := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true} commands <- SessionCommand{Kind: CommandSetVideo, Config: want} event := receiveControllerEvent(t, events) if event.unit != UnitVideo || event.action != "command" || event.feed != want { t.Fatalf("replacement event = %+v", event) } select { case event := <-events: t.Fatalf("unchanged audio slot was disturbed: %+v", event) case <-time.After(20 * time.Millisecond): } close(commands) select { case err := <-done: if err != nil { t.Fatalf("Run() error = %v, want nil", err) } case <-time.After(time.Second): t.Fatal("Run() did not stop") } } func TestSessionControllerReplacesSynchronizedPairWithOneCommand(t *testing.T) { events := make(chan controllerEvent, 32) controller := newRecordingController(t, events) initial := validCommandSession() commands := make(chan SessionCommand) done := make(chan error, 1) go func() { done <- controller.Run(context.Background(), initial, commands) }() start := receiveControllerEvent(t, events) if start.unit != UnitSync || start.action != "start" { t.Fatalf("initial event = %+v, want sync start", start) } next := initial next.Video = FeedConfig{Domain: "/next", UUID: "next-video", Active: true} next.Audio = FeedConfig{Domain: "/next", UUID: "next-audio", Active: true} commands <- SessionCommand{Kind: CommandSetSession, Session: next} event := receiveControllerEvent(t, events) wantPair := SyncPairConfig{Video: next.Video, Audio: next.Audio} if event.unit != UnitSync || event.action != "command" || event.pair != wantPair { t.Fatalf("replacement event = %+v, want one sync command for %#v", event, wantPair) } select { case event := <-events: t.Fatalf("atomic replacement emitted an extra event: %+v", event) case <-time.After(20 * time.Millisecond): } close(commands) if err := <-done; err != nil { t.Fatalf("Run() error = %v", err) } } func TestSessionControllerStopsIndependentSlotsBeforeStartingSync(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: CommandEnableSync} stopped := map[Unit]bool{} for { event := receiveControllerEvent(t, events) if event.unit == UnitSync && event.action == "start" { if !stopped[UnitVideo] || !stopped[UnitAudio] { t.Fatalf("sync started before both independent slots stopped: %v", stopped) } if event.generation != 2 { t.Fatalf("sync runtime generation = %d, want 2", event.generation) } break } if event.action != "stop" || (event.unit != UnitVideo && event.unit != UnitAudio) { t.Fatalf("unexpected transition event: %+v", event) } stopped[event.unit] = true } close(commands) select { case err := <-done: if err != nil { t.Fatalf("Run() error = %v, want nil", err) } case <-time.After(time.Second): t.Fatal("Run() did not stop") } } func TestSessionControllerCancellationStopsAndJoinsRuntime(t *testing.T) { events := make(chan controllerEvent, 32) controller := newRecordingController(t, events) initial := validCommandSession() initial.SyncRequested = false ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { done <- controller.Run(ctx, initial, make(chan SessionCommand)) }() receiveIndependentStarts(t, events) cancel() select { case err := <-done: if !errors.Is(err, context.Canceled) { t.Fatalf("Run() error = %v, want context.Canceled", err) } case <-time.After(time.Second): t.Fatal("Run() did not stop after cancellation") } } func TestSessionControllerSnapshotTracksDesiredPlanAndGeneration(t *testing.T) { events := make(chan controllerEvent, 64) controller := newRecordingController(t, events) if snapshot, ok := controller.Snapshot(); ok { t.Fatalf("Snapshot() before Run = %#v, true; want unavailable", snapshot) } 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) snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool { return snapshot.Desired == initial }) if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent { t.Fatalf("initial snapshot = %#v", snapshot) } newVideo := FeedConfig{Domain: "/new-video", UUID: "new-video", Active: true} commands <- SessionCommand{Kind: CommandSetVideo, Config: newVideo} receiveControllerEvent(t, events) snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool { return snapshot.Desired.Video == newVideo }) if snapshot.Generation != 1 || snapshot.Plan.Topology != TopologyIndependent { t.Fatalf("same-topology snapshot = %#v", snapshot) } commands <- SessionCommand{Kind: CommandEnableSync} for { if event := receiveControllerEvent(t, events); event.unit == UnitSync && event.action == "start" { break } } snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool { return snapshot.Plan.Topology == TopologySynchronized }) if snapshot.Generation != 2 || !snapshot.Desired.SyncRequested { t.Fatalf("sync snapshot = %#v", snapshot) } commands <- SessionCommand{Kind: CommandDisableSync} for { event := receiveControllerEvent(t, events) if event.action == "start" && (event.unit == UnitVideo || event.unit == UnitAudio) { break } } snapshot = waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool { return snapshot.Plan.Topology == TopologyIndependent && !snapshot.Desired.SyncRequested }) if snapshot.Generation != 3 { t.Fatalf("independent snapshot generation = %d, want 3", snapshot.Generation) } close(commands) if err := <-done; err != nil { t.Fatalf("Run() error = %v", err) } } func TestSessionControllerSnapshotRetainsUnavailableSyncRequest(t *testing.T) { events := make(chan controllerEvent, 32) controller, err := NewSessionController( recordingVideoSlot{events}, recordingAudioSlot{events}, recordingSyncSlot{events}, nil, ) if err != nil { t.Fatal(err) } 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: CommandEnableSync} snapshot := waitControllerSnapshot(t, controller, func(snapshot SessionSnapshot) bool { return snapshot.Desired.SyncRequested }) if snapshot.Plan.Topology != TopologyIndependent || snapshot.Generation != 1 { t.Fatalf("unsupported-sync snapshot = %#v", snapshot) } close(commands) if err := <-done; err != nil { t.Fatal(err) } } func TestSessionControllerSnapshotConcurrentReads(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) waitControllerSnapshot(t, controller, func(SessionSnapshot) bool { return true }) var readers sync.WaitGroup for range 8 { readers.Add(1) go func() { defer readers.Done() for range 1_000 { controller.Snapshot() } }() } readers.Wait() close(commands) if err := <-done; err != nil { t.Fatal(err) } }