go-mxl-player/internal/playback/video_bridge_test.go

package playback

import (
	"context"
	"errors"
	"testing"
	"time"
)

const videoBridgeTestTimeout = time.Second

func TestVideoBridgeDeliversFrameAndCompletionResult(t *testing.T) {
	bridge := NewVideoBridge()
	wantErr := errors.New("stage frame")
	wantFrame := VideoFrame{
		Index:   42,
		Width:   1920,
		Height:  1080,
		Stride:  7680,
		Payload: []byte{1, 2, 3},
	}
	consumeResult := make(chan error, 1)

	wantSource := FeedConfig{Domain: "/video", UUID: "video", Active: true}
	go func() {
		ctx := withGeneration(context.Background(), 17)
		consumeResult <- bridge.ConsumeVideo(withVideoSource(ctx, wantSource), wantFrame)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
	defer cancel()
	pending, err := bridge.Next(ctx)
	if err != nil {
		t.Fatalf("Next() error = %v", err)
	}
	if pending.Frame.Index != wantFrame.Index {
		t.Fatalf("Next() frame index = %d, want %d", pending.Frame.Index, wantFrame.Index)
	}
	if pending.Generation != 17 {
		t.Fatalf("Next() generation = %d, want 17", pending.Generation)
	}
	if pending.Source != wantSource {
		t.Fatalf("Next() source = %#v, want %#v", pending.Source, wantSource)
	}
	if &pending.Frame.Payload[0] != &wantFrame.Payload[0] {
		t.Fatal("Next() copied the borrowed payload")
	}

	pending.Complete(wantErr)
	select {
	case err := <-consumeResult:
		if !errors.Is(err, wantErr) {
			t.Fatalf("ConsumeVideo() error = %v, want %v", err, wantErr)
		}
	case <-time.After(videoBridgeTestTimeout):
		t.Fatal("ConsumeVideo() did not return after completion")
	}
}

func TestVideoBridgeConsumeHonorsCancellationBeforeDelivery(t *testing.T) {
	bridge := NewVideoBridge()
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err := bridge.ConsumeVideo(ctx, VideoFrame{})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("ConsumeVideo() error = %v, want %v", err, context.Canceled)
	}
}

func TestVideoBridgeConsumeWaitsForCompletionAfterDelivery(t *testing.T) {
	bridge := NewVideoBridge()
	ctx, cancel := context.WithCancel(context.Background())
	consumeResult := make(chan error, 1)

	go func() {
		consumeResult <- bridge.ConsumeVideo(ctx, VideoFrame{Index: 7})
	}()

	nextCtx, nextCancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
	defer nextCancel()
	pending, err := bridge.Next(nextCtx)
	if err != nil {
		t.Fatalf("Next() error = %v", err)
	}
	cancel()

	select {
	case err := <-consumeResult:
		t.Fatalf("ConsumeVideo() returned before completion: %v", err)
	case <-time.After(20 * time.Millisecond):
	}

	pending.Complete(nil)
	select {
	case err := <-consumeResult:
		if err != nil {
			t.Fatalf("ConsumeVideo() error = %v, want nil", err)
		}
	case <-time.After(videoBridgeTestTimeout):
		t.Fatal("ConsumeVideo() did not return after completion")
	}
}

func TestPendingVideoFrameCompleteIsIdempotent(t *testing.T) {
	bridge := NewVideoBridge()
	consumeResult := make(chan error, 1)

	go func() {
		consumeResult <- bridge.ConsumeVideo(context.Background(), VideoFrame{})
	}()

	ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
	defer cancel()
	pending, err := bridge.Next(ctx)
	if err != nil {
		t.Fatalf("Next() error = %v", err)
	}

	firstErr := errors.New("first")
	pending.Complete(firstErr)
	pending.Complete(errors.New("second"))

	select {
	case err := <-consumeResult:
		if !errors.Is(err, firstErr) {
			t.Fatalf("ConsumeVideo() error = %v, want %v", err, firstErr)
		}
	case <-time.After(videoBridgeTestTimeout):
		t.Fatal("ConsumeVideo() did not return")
	}
}

func TestVideoBridgeNextHonorsCancellation(t *testing.T) {
	bridge := NewVideoBridge()
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	pending, err := bridge.Next(ctx)
	if pending != nil {
		t.Fatalf("Next() pending = %#v, want nil", pending)
	}
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("Next() error = %v, want %v", err, context.Canceled)
	}
}

func TestVideoBridgeTryNextReturnsImmediatelyWhenEmpty(t *testing.T) {
	bridge := NewVideoBridge()
	if pending, ok := bridge.TryNext(); ok || pending != nil {
		t.Fatalf("TryNext() = %#v, %t; want nil, false", pending, ok)
	}
}

func TestVideoBridgeTryNextDeliversWithoutCopyAndRequiresCompletion(t *testing.T) {
	bridge := NewVideoBridge()
	frame := VideoFrame{Index: 9, Payload: []byte{1, 2, 3}}
	consumeResult := make(chan error, 1)
	started := make(chan struct{})
	go func() {
		close(started)
		consumeResult <- bridge.ConsumeVideo(context.Background(), frame)
	}()
	<-started

	deadline := time.Now().Add(videoBridgeTestTimeout)
	var pending *PendingVideoFrame
	for pending == nil && time.Now().Before(deadline) {
		pending, _ = bridge.TryNext()
		if pending == nil {
			time.Sleep(time.Millisecond)
		}
	}
	if pending == nil {
		t.Fatal("TryNext() did not receive waiting producer")
	}
	if &pending.Frame.Payload[0] != &frame.Payload[0] {
		t.Fatal("TryNext() copied borrowed payload")
	}
	select {
	case err := <-consumeResult:
		t.Fatalf("ConsumeVideo() returned before completion: %v", err)
	default:
	}

	pending.Complete(nil)
	select {
	case err := <-consumeResult:
		if err != nil {
			t.Fatalf("ConsumeVideo() error = %v", err)
		}
	case <-time.After(videoBridgeTestTimeout):
		t.Fatal("ConsumeVideo() did not return after completion")
	}
}