342 lines
8.3 KiB
Go
342 lines
8.3 KiB
Go
package playback
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type videoOpenResult struct {
|
|
reader VideoReader
|
|
err error
|
|
}
|
|
|
|
type scriptedVideoFactory struct {
|
|
results []videoOpenResult
|
|
calls int
|
|
}
|
|
|
|
func (f *scriptedVideoFactory) OpenVideo(
|
|
context.Context,
|
|
FeedConfig,
|
|
) (VideoReader, error) {
|
|
if f.calls >= len(f.results) {
|
|
return nil, errors.New("unexpected video open attempt")
|
|
}
|
|
result := f.results[f.calls]
|
|
f.calls++
|
|
return result.reader, result.err
|
|
}
|
|
|
|
func activeVideoConfig() FeedConfig {
|
|
return FeedConfig{
|
|
Domain: "/dev/shm/mxl",
|
|
UUID: "video-uuid",
|
|
Active: true,
|
|
}
|
|
}
|
|
|
|
func newTestVideoWorker(
|
|
t *testing.T,
|
|
factory VideoReaderFactory,
|
|
sink VideoSink,
|
|
maxAttempts int,
|
|
shouldRetry func(error) bool,
|
|
observer StatusObserver,
|
|
) *VideoWorker {
|
|
t.Helper()
|
|
|
|
worker, err := NewVideoWorker(
|
|
factory,
|
|
sink,
|
|
testRetryPolicy(maxAttempts),
|
|
shouldRetry,
|
|
observer,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("NewVideoWorker() error = %v", err)
|
|
}
|
|
worker.wait = func(context.Context, time.Duration) error { return nil }
|
|
return worker
|
|
}
|
|
|
|
func TestNewVideoWorkerValidatesDependencies(t *testing.T) {
|
|
factory := &scriptedVideoFactory{}
|
|
sink := &fakeVideoSink{}
|
|
retry := testRetryPolicy(3)
|
|
decide := func(error) bool { return true }
|
|
|
|
tests := []struct {
|
|
name string
|
|
factory VideoReaderFactory
|
|
sink VideoSink
|
|
retry RetryPolicy
|
|
shouldRetry func(error) bool
|
|
wantErr error
|
|
}{
|
|
{
|
|
name: "missing factory",
|
|
sink: sink,
|
|
retry: retry,
|
|
shouldRetry: decide,
|
|
wantErr: ErrVideoFactoryRequired,
|
|
},
|
|
{
|
|
name: "missing sink",
|
|
factory: factory,
|
|
retry: retry,
|
|
shouldRetry: decide,
|
|
wantErr: ErrVideoSinkRequired,
|
|
},
|
|
{
|
|
name: "missing retry decider",
|
|
factory: factory,
|
|
sink: sink,
|
|
retry: retry,
|
|
wantErr: ErrVideoRetryDeciderRequired,
|
|
},
|
|
{
|
|
name: "invalid retry policy",
|
|
factory: factory,
|
|
sink: sink,
|
|
retry: RetryPolicy{},
|
|
shouldRetry: decide,
|
|
wantErr: ErrInvalidRetryDelay,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
worker, err := NewVideoWorker(
|
|
tt.factory,
|
|
tt.sink,
|
|
tt.retry,
|
|
tt.shouldRetry,
|
|
nil,
|
|
)
|
|
if worker != nil {
|
|
t.Fatal("NewVideoWorker() worker is not nil")
|
|
}
|
|
if !errors.Is(err, tt.wantErr) {
|
|
t.Fatalf("NewVideoWorker() error = %v, want %v", err, tt.wantErr)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestVideoWorkerRejectsInactiveFeed(t *testing.T) {
|
|
factory := &scriptedVideoFactory{}
|
|
var statuses []Status
|
|
worker := newTestVideoWorker(
|
|
t,
|
|
factory,
|
|
&fakeVideoSink{},
|
|
3,
|
|
func(error) bool { return true },
|
|
func(status Status) { statuses = append(statuses, status) },
|
|
)
|
|
config := activeVideoConfig()
|
|
config.Active = false
|
|
|
|
err := worker.Run(context.Background(), config)
|
|
if !errors.Is(err, ErrVideoFeedInactive) {
|
|
t.Fatalf("Run() error = %v, want %v", err, ErrVideoFeedInactive)
|
|
}
|
|
if factory.calls != 0 {
|
|
t.Errorf("factory calls = %d, want 0", factory.calls)
|
|
}
|
|
if len(statuses) != 0 {
|
|
t.Errorf("status count = %d, want 0", len(statuses))
|
|
}
|
|
}
|
|
|
|
func TestVideoWorkerStatusesInheritGeneration(t *testing.T) {
|
|
openErr := errors.New("unavailable")
|
|
var statuses []Status
|
|
worker := newTestVideoWorker(
|
|
t,
|
|
&scriptedVideoFactory{results: []videoOpenResult{{err: openErr}}},
|
|
&fakeVideoSink{},
|
|
1,
|
|
func(error) bool { return true },
|
|
func(status Status) { statuses = append(statuses, status) },
|
|
)
|
|
_ = worker.Run(withGeneration(context.Background(), 7), activeVideoConfig())
|
|
if len(statuses) == 0 {
|
|
t.Fatal("no statuses emitted")
|
|
}
|
|
for _, status := range statuses {
|
|
if status.Generation != 7 {
|
|
t.Fatalf("status generation = %d, want 7: %+v", status.Generation, status)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestVideoWorkerExhaustsOpenRetries(t *testing.T) {
|
|
openErr := errors.New("producer unavailable")
|
|
factory := &scriptedVideoFactory{
|
|
results: []videoOpenResult{
|
|
{err: openErr},
|
|
{err: openErr},
|
|
},
|
|
}
|
|
var statuses []Status
|
|
worker := newTestVideoWorker(
|
|
t,
|
|
factory,
|
|
&fakeVideoSink{},
|
|
2,
|
|
func(error) bool { return true },
|
|
func(status Status) { statuses = append(statuses, status) },
|
|
)
|
|
|
|
err := worker.Run(context.Background(), activeVideoConfig())
|
|
if !errors.Is(err, openErr) {
|
|
t.Fatalf("Run() error = %v, want %v", err, openErr)
|
|
}
|
|
if factory.calls != 2 {
|
|
t.Errorf("factory calls = %d, want 2", factory.calls)
|
|
}
|
|
|
|
wantStates := []State{
|
|
StateConnecting,
|
|
StateReconnecting,
|
|
StateReconnecting,
|
|
StateFailed,
|
|
}
|
|
if len(statuses) != len(wantStates) {
|
|
t.Fatalf("status count = %d, want %d: %+v", len(statuses), len(wantStates), statuses)
|
|
}
|
|
for i, want := range wantStates {
|
|
if statuses[i].State != want {
|
|
t.Errorf("status %d state = %v, want %v", i, statuses[i].State, want)
|
|
}
|
|
if statuses[i].Unit != UnitVideo {
|
|
t.Errorf("status %d unit = %v, want %v", i, statuses[i].Unit, UnitVideo)
|
|
}
|
|
}
|
|
final := statuses[len(statuses)-1]
|
|
if final.Attempt != 2 || final.FailedAttempts != 2 {
|
|
t.Errorf("final status = %+v, want attempt=2 failedAttempts=2", final)
|
|
}
|
|
if !errors.Is(final.Err, openErr) {
|
|
t.Errorf("final error = %v, want %v", final.Err, openErr)
|
|
}
|
|
}
|
|
|
|
func TestVideoWorkerStablePlaybackResetsRetryCounter(t *testing.T) {
|
|
readErr := errors.New("video disconnected")
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
first := &fakeVideoReader{
|
|
frames: []VideoFrame{{Index: 1, Payload: []byte{1}}},
|
|
readErr: readErr,
|
|
}
|
|
second := &fakeVideoReader{
|
|
frames: []VideoFrame{{Index: 2, Payload: []byte{2}}},
|
|
readErr: readErr,
|
|
}
|
|
third := &fakeVideoReader{
|
|
read: func(ctx context.Context) (VideoFrame, error) {
|
|
cancel()
|
|
return VideoFrame{}, ctx.Err()
|
|
},
|
|
}
|
|
factory := &scriptedVideoFactory{
|
|
results: []videoOpenResult{
|
|
{reader: first},
|
|
{reader: second},
|
|
{reader: third},
|
|
},
|
|
}
|
|
var statuses []Status
|
|
worker := newTestVideoWorker(
|
|
t,
|
|
factory,
|
|
&fakeVideoSink{},
|
|
2,
|
|
func(error) bool { return true },
|
|
func(status Status) { statuses = append(statuses, status) },
|
|
)
|
|
|
|
err := worker.Run(ctx, activeVideoConfig())
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("Run() error = %v, want context.Canceled", err)
|
|
}
|
|
if factory.calls != 3 {
|
|
t.Fatalf("factory calls = %d, want 3", factory.calls)
|
|
}
|
|
|
|
var playingAttempts []int
|
|
var retryFailures []int
|
|
for _, status := range statuses {
|
|
switch status.State {
|
|
case StatePlaying:
|
|
playingAttempts = append(playingAttempts, status.Attempt)
|
|
case StateReconnecting:
|
|
if status.RetryIn > 0 {
|
|
retryFailures = append(retryFailures, status.FailedAttempts)
|
|
}
|
|
}
|
|
}
|
|
if len(playingAttempts) != 2 || playingAttempts[0] != 1 || playingAttempts[1] != 2 {
|
|
t.Errorf("playing attempts = %v, want [1 2]", playingAttempts)
|
|
}
|
|
if len(retryFailures) != 2 || retryFailures[0] != 1 || retryFailures[1] != 1 {
|
|
t.Errorf("retry failure counts = %v, want [1 1]", retryFailures)
|
|
}
|
|
wantEnding := []State{StateStopping, StateIdle}
|
|
if len(statuses) < 2 {
|
|
t.Fatalf("status count = %d, want at least 2", len(statuses))
|
|
}
|
|
ending := statuses[len(statuses)-2:]
|
|
for i, want := range wantEnding {
|
|
if ending[i].State != want {
|
|
t.Errorf("ending status %d = %v, want %v", i, ending[i].State, want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestVideoWorkerDoesNotRetrySinkFailure(t *testing.T) {
|
|
sinkErr := errors.New("renderer failed")
|
|
reader := &fakeVideoReader{
|
|
frames: []VideoFrame{{Index: 1, Payload: []byte{1}}},
|
|
}
|
|
factory := &scriptedVideoFactory{
|
|
results: []videoOpenResult{{reader: reader}},
|
|
}
|
|
deciderCalls := 0
|
|
var statuses []Status
|
|
worker := newTestVideoWorker(
|
|
t,
|
|
factory,
|
|
&fakeVideoSink{err: sinkErr},
|
|
0,
|
|
func(error) bool {
|
|
deciderCalls++
|
|
return true
|
|
},
|
|
func(status Status) { statuses = append(statuses, status) },
|
|
)
|
|
|
|
err := worker.Run(context.Background(), activeVideoConfig())
|
|
if !errors.Is(err, sinkErr) {
|
|
t.Fatalf("Run() error = %v, want %v", err, sinkErr)
|
|
}
|
|
if factory.calls != 1 {
|
|
t.Errorf("factory calls = %d, want 1", factory.calls)
|
|
}
|
|
if deciderCalls != 0 {
|
|
t.Errorf("source retry decider calls = %d, want 0", deciderCalls)
|
|
}
|
|
if !reader.closed {
|
|
t.Fatal("reader was not closed")
|
|
}
|
|
if len(statuses) != 2 {
|
|
t.Fatalf("status count = %d, want 2: %+v", len(statuses), statuses)
|
|
}
|
|
if statuses[0].State != StateConnecting || statuses[1].State != StateFailed {
|
|
t.Errorf("status states = [%v %v], want [Connecting Failed]", statuses[0].State, statuses[1].State)
|
|
}
|
|
}
|