Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
4.2 kB · 121 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122package media
import ( "bytes" "context" "encoding/json" "testing" "time"
"github.com/stretchr/testify/require" "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/comatproto")
// TestStreamKickMarshalsAsPlaceStreamError locks the dashboard wire contract: a// StreamKick must marshal to the place.stream.error frame the client turns into// a "problem" (js/.../websocket-consumer.tsx keys on $type/code/message).func TestStreamKickMarshalsAsPlaceStreamError(t *testing.T) { bs, err := json.Marshal(NewStreamKick("bitrate", "too high")) require.NoError(t, err)
var got map[string]any require.NoError(t, json.Unmarshal(bs, &got)) require.Equal(t, "place.stream.error", got["$type"]) require.Equal(t, "bitrate", got["code"]) require.Equal(t, "too high", got["message"])}
// TestWatchKeyRevocationBan checks the shared detection core: a banned label// published to the streamer's bus channel fires onRevoked. (The bus only// delivers to already-registered subscribers, so we re-publish on a tick until// the watcher's subscription is live.)func TestWatchKeyRevocationBan(t *testing.T) { mm, _ := getStaticTestMediaManager(t) ms := newBareSegmentSigner(t)
ctx, cancel := context.WithCancel(context.Background()) defer cancel()
revoked := make(chan string, 1) go mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { revoked <- reason })
banned := &comatproto.LabelDefs_Label{Val: atproto.LabelDMCAViolation, Uri: "did:plc:test-streamer"} tick := time.NewTicker(50 * time.Millisecond) defer tick.Stop() deadline := time.After(5 * time.Second) for { select { case reason := <-revoked: require.Contains(t, reason, "user banned") return case <-deadline: t.Fatal("ban label did not trigger key revocation") case <-tick.C: mm.bus.Publish(ms.Streamer(), banned) } }}
// TestWatchKeyRevocationStreamKick checks that a StreamKick published to the// streamer's bus channel fires onRevoked with its message — the path the max// live bitrate enforcement uses to tear a stream down across every ingest path.func TestWatchKeyRevocationStreamKick(t *testing.T) { mm, _ := getStaticTestMediaManager(t) ms := newBareSegmentSigner(t)
ctx, cancel := context.WithCancel(context.Background()) defer cancel()
revoked := make(chan string, 1) go mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { revoked <- reason })
kick := NewStreamKick("bitrate", "bitrate too high") tick := time.NewTicker(50 * time.Millisecond) defer tick.Stop() deadline := time.After(5 * time.Second) for { select { case reason := <-revoked: require.Equal(t, "bitrate too high", reason) return case <-deadline: t.Fatal("StreamKick did not trigger teardown") case <-tick.C: mm.bus.Publish(ms.Streamer(), kick) } }}
// TestMP4IngestIsolatedBanContained proves the fix end to end: banning a streamer// mid-ingest tears their isolated worker down. The watchdog is set generously// (60s) and the input is a wedging audio-only fMP4 that never ends on its own —// so a timely return can only be the ban kill, not the watchdog or a natural// EOS. (The 4-audio sample-stream.mp4 previously used here now ingests to// completion in a few seconds, which would race the ban.)func TestMP4IngestIsolatedBanContained(t *testing.T) { old := ingestWorkerWatchdog ingestWorkerWatchdog = 60 * time.Second defer func() { ingestWorkerWatchdog = old }()
mm, _ := getStaticTestMediaManager(t) ms := newBareSegmentSigner(t) wedge := makeAudioOnlyAACFMP4(t, context.Background(), 5)
// Ban the streamer once the worker is up and the watcher has subscribed. go func() { time.Sleep(3 * time.Second) mm.bus.Publish(ms.Streamer(), &comatproto.LabelDefs_Label{ Val: atproto.LabelDMCAViolation, Uri: "did:plc:test-streamer", }) }()
start := time.Now() err := mm.MP4IngestIsolated(context.Background(), bytes.NewReader(wedge), ms) elapsed := time.Since(start)
require.Error(t, err, "a banned stream is torn down, surfaced as an error") require.Less(t, elapsed, 30*time.Second, "the ban killed the worker well before the 60s watchdog") t.Logf("banned worker contained in %s: %v", elapsed.Round(time.Second), err)}