Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
3.1 kB · 104 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105package api
import ( "bytes" "context" "fmt" "io" "testing" "testing/synctest" "time"
"github.com/stretchr/testify/require")
func TestShadowIdleReaderAllowsConsumerWork(t *testing.T) { synctest.Test(t, func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() reader := newShadowIdleReader(bytes.NewReader([]byte{3}), cancel) defer reader.stop() var buf [1]byte n, err := reader.Read(buf[:]) require.NoError(t, err) require.Equal(t, 1, n)
// A caller processing bytes is not waiting on idle publisher input. time.Sleep(2 * RTMPTimeout) require.NoError(t, ctx.Err()) n, err = reader.Read(buf[:]) require.ErrorIs(t, err, io.EOF) require.Zero(t, n) time.Sleep(2 * RTMPTimeout) require.NoError(t, ctx.Err(), "publisher EOF must allow downstream drain") })}
func TestShadowIdleReaderStaleCallbackHonorsPendingRead(t *testing.T) { synctest.Test(t, func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() input, publisher := io.Pipe() defer input.Close() defer publisher.Close() stopInput := context.AfterFunc(ctx, func() { _ = input.Close() }) defer stopInput() reader := newShadowIdleReader(input, cancel) defer reader.stop() result := make(chan error, 1) go func() { var buf [1]byte _, err := reader.Read(buf[:]) result <- err }() synctest.Wait() time.Sleep(RTMPTimeout / 2) // A previous timer callback can already be queued when a new Read starts. reader.expire() require.NoError(t, ctx.Err(), "an old callback must not shorten this Read") time.Sleep(RTMPTimeout / 2) require.ErrorIs(t, <-result, errShadowIdleTimeout) require.ErrorIs(t, ctx.Err(), context.Canceled) })}
func TestDuplicateMistIdleHandshakeClosesOnlyShadowInput(t *testing.T) { input, publisher := io.Pipe() defer input.Close() defer publisher.Close() result := make(chan error, 1) go func() { result <- RunDuplicateMistWorker(t.Context(), input, io.Discard) }() select { case err := <-result: require.ErrorIs(t, err, errShadowIdleTimeout) case <-time.After(RTMPTimeout + 5*time.Second): t.Fatal("idle shadow handshake did not exit") }}
func TestDuplicateMistRejectsTruncatedHandshake(t *testing.T) { for name, tc := range map[string]struct { input []byte want error }{ "no handshake": {nil, io.EOF}, "truncated plain handshake": {[]byte{3, 0, 0}, io.ErrUnexpectedEOF}, } { t.Run(name, func(t *testing.T) { err := RunDuplicateMistWorker(t.Context(), bytes.NewReader(tc.input), io.Discard) require.ErrorIs(t, err, tc.want) }) }}
// Upstream parser errors can print entire AMF connect/publish commands.func TestShadowProtocolErrorsNeverExposeStreamKey(t *testing.T) { const key = "zSECRET-STREAM-KEY" for _, err := range []error{ fmt.Errorf("invalid publish arguments: %s", key), fmt.Errorf("publish %s: %w", key, io.EOF), fmt.Errorf("publish %s: %w", key, io.ErrUnexpectedEOF), } { require.NotContains(t, shadowProtocolError("accept", err).Error(), key) }}