package playback import ( "context" "errors" "sync" "testing" "time" ) type slotVideoFactory struct { opened chan FeedConfig mu sync.Mutex active int maxActive int closeCount int } func newSlotVideoFactory() *slotVideoFactory { return &slotVideoFactory{opened: make(chan FeedConfig, 8)} } func (f *slotVideoFactory) OpenVideo( _ context.Context, config FeedConfig, ) (VideoReader, error) { f.mu.Lock() f.active++ if f.active > f.maxActive { f.maxActive = f.active } f.mu.Unlock() f.opened <- config return &slotVideoReader{factory: f}, nil } func (f *slotVideoFactory) counts() (active, maxActive, closeCount int) { f.mu.Lock() defer f.mu.Unlock() return f.active, f.maxActive, f.closeCount } type slotVideoReader struct { factory *slotVideoFactory } func (r *slotVideoReader) ReadVideo(ctx context.Context) (VideoFrame, error) { <-ctx.Done() return VideoFrame{}, ctx.Err() } func (r *slotVideoReader) Close() error { r.factory.mu.Lock() defer r.factory.mu.Unlock() r.factory.active-- r.factory.closeCount++ return nil } func newSlotTestWorker(t *testing.T, factory VideoReaderFactory) *VideoWorker { t.Helper() worker, err := NewVideoWorker( factory, &fakeVideoSink{}, testRetryPolicy(1), func(error) bool { return false }, nil, ) if err != nil { t.Fatalf("NewVideoWorker() error = %v", err) } return worker } func receiveSlotOpen(t *testing.T, opened <-chan FeedConfig) FeedConfig { t.Helper() select { case config := <-opened: return config case <-time.After(time.Second): t.Fatal("video worker did not open") return FeedConfig{} } } func TestNewVideoSlotRequiresWorker(t *testing.T) { slot, err := NewVideoSlot(nil) if slot != nil { t.Fatalf("NewVideoSlot(nil) slot = %#v, want nil", slot) } if !errors.Is(err, ErrVideoWorkerRequired) { t.Fatalf("NewVideoSlot(nil) error = %v, want %v", err, ErrVideoWorkerRequired) } } func TestVideoSlotStartsInitialActiveConfig(t *testing.T) { factory := newSlotVideoFactory() slot, err := NewVideoSlot(newSlotTestWorker(t, factory)) if err != nil { t.Fatalf("NewVideoSlot() error = %v", err) } ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) want := FeedConfig{Domain: "/video", UUID: "first", Active: true} go func() { done <- slot.Run(ctx, want, make(chan FeedConfig)) }() if got := receiveSlotOpen(t, factory.opened); got != want { t.Fatalf("opened config = %#v, want %#v", got, want) } cancel() select { case err := <-done: if !errors.Is(err, context.Canceled) { t.Fatalf("Run() error = %v, want %v", err, context.Canceled) } case <-time.After(time.Second): t.Fatal("Run() did not stop after cancellation") } active, _, closeCount := factory.counts() if active != 0 || closeCount != 1 { t.Fatalf("reader counts = active %d, closed %d; want 0, 1", active, closeCount) } } func TestVideoSlotReplacesWithoutOverlappingWorkers(t *testing.T) { factory := newSlotVideoFactory() slot, err := NewVideoSlot(newSlotTestWorker(t, factory)) if err != nil { t.Fatalf("NewVideoSlot() error = %v", err) } ctx, cancel := context.WithCancel(context.Background()) commands := make(chan FeedConfig) done := make(chan error, 1) first := FeedConfig{Domain: "/video", UUID: "first", Active: true} second := FeedConfig{Domain: "/video", UUID: "second", Active: true} go func() { done <- slot.Run(ctx, first, commands) }() receiveSlotOpen(t, factory.opened) commands <- second if got := receiveSlotOpen(t, factory.opened); got != second { t.Fatalf("replacement config = %#v, want %#v", got, second) } cancel() select { case <-done: case <-time.After(time.Second): t.Fatal("Run() did not stop") } active, maxActive, closeCount := factory.counts() if active != 0 || maxActive != 1 || closeCount != 2 { t.Fatalf( "reader counts = active %d, maximum %d, closed %d; want 0, 1, 2", active, maxActive, closeCount, ) } } func TestVideoSlotIgnoresInvalidCommand(t *testing.T) { factory := newSlotVideoFactory() slot, err := NewVideoSlot(newSlotTestWorker(t, factory)) if err != nil { t.Fatalf("NewVideoSlot() error = %v", err) } ctx, cancel := context.WithCancel(context.Background()) commands := make(chan FeedConfig) done := make(chan error, 1) initial := FeedConfig{Domain: "/video", UUID: "first", Active: true} go func() { done <- slot.Run(ctx, initial, commands) }() receiveSlotOpen(t, factory.opened) commands <- FeedConfig{UUID: "invalid", Active: true} select { case config := <-factory.opened: t.Fatalf("invalid command opened config %#v", config) case <-time.After(20 * time.Millisecond): } active, _, closeCount := factory.counts() if active != 1 || closeCount != 0 { t.Fatalf("invalid command disturbed reader: active %d, closed %d", active, closeCount) } cancel() <-done } func TestVideoSlotInactiveCommandStopsWithoutRestart(t *testing.T) { factory := newSlotVideoFactory() slot, err := NewVideoSlot(newSlotTestWorker(t, factory)) if err != nil { t.Fatalf("NewVideoSlot() error = %v", err) } ctx, cancel := context.WithCancel(context.Background()) defer cancel() commands := make(chan FeedConfig) done := make(chan error, 1) initial := FeedConfig{Domain: "/video", UUID: "first", Active: true} go func() { done <- slot.Run(ctx, initial, commands) }() receiveSlotOpen(t, factory.opened) commands <- FeedConfig{Domain: "/video", UUID: "first", Active: false} deadline := time.Now().Add(time.Second) for { active, _, closeCount := factory.counts() if active == 0 && closeCount == 1 { break } if time.Now().After(deadline) { t.Fatal("inactive command did not stop reader") } time.Sleep(time.Millisecond) } select { case config := <-factory.opened: t.Fatalf("inactive command restarted config %#v", config) 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 after commands closed") } }