package app import ( "bytes" "context" "encoding/json" "fmt" "net/http" "net/http/httptest" "net/url" "os/exec" "strings" "sync" "testing" "time" ) func embyScanFixture(t *testing.T, optimized bool) (*UIApp, func(int) string) { t.Helper() app, fixture := nativePlaybackFixture(t, "4", "160x90") app.closePlayback(fixture.id) app.mediaResources().settings.Enabled = optimized app.mediaResources().settings.MaxSessions = 1 for index := 1; index <= 8; index++ { task := fixture.tasks[0] task.Index, task.Total = index, 8 task.Chapter.ID = fmt.Sprintf("%s:%d", task.DramaID, index) task.Chapter.CurrentEpisode = rawEpisode(index) id := fmt.Sprintf("scan-%d", index) app.tasks[id] = &UITask{ID: id, DramaID: task.DramaID, Status: uiStatusSuccess, Path: task.OutPath, Source: task} app.taskOrder = append(app.taskOrder, id) } key, err := app.embySigningKey(true) if err != nil { t.Fatal(err) } return app, func(index int) string { task := app.tasks[fmt.Sprintf("scan-%d", index)].Source query := url.Values{"id": {task.DramaID}, "chapter": {task.Chapter.ID}, "key": {embyToken(key, task.DramaID, task.Chapter.ID)}} return "/api/emby/stream.m3u8?" + query.Encode() } } func embyRequest(app *UIApp, method, address string) *httptest.ResponseRecorder { writer := httptest.NewRecorder() app.routes().ServeHTTP(writer, httptest.NewRequest(method, "http://localhost"+address, nil)) return writer } func embyFirstSegment(t *testing.T, playlist string) string { t.Helper() for _, line := range strings.Split(playlist, "\n") { if strings.HasPrefix(line, "segment.ts?") { return "/api/emby/" + line } } t.Fatal("missing signed HLS segment") return "" } func TestEmbySerialProbesReleaseSlotsAndReusePlayback(t *testing.T) { probe, err := exec.LookPath("ffprobe") if err != nil { t.Skip("ffprobe unavailable") } for _, optimized := range []bool{false, true} { t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) { app, address := embyScanFixture(t, optimized) server := httptest.NewServer(app.routes()) defer server.Close() first := embyRequest(app, http.MethodGet, address(1)) firstBody, firstLocation := first.Body.String(), first.Header().Get("Location") for index := 1; index <= 8; index++ { if head := embyRequest(app, http.MethodHead, address(index)); head.Code != 200 || head.Body.Len() != 0 { t.Fatal("HEAD failed", head.Code) } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) output, err := exec.CommandContext(ctx, probe, "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", server.URL+address(index)).CombinedOutput() cancel() if err != nil || !strings.Contains(string(output), "4.000000") { t.Fatalf("serial probe %d failed: %v %s", index, err, output) } app.playbackMu.Lock() active := app.activePlaybackSessionsLocked() cached := 0 for _, session := range app.playbacks { if session.native != nil { cached++ } } app.playbackMu.Unlock() if active != 0 || cached > 1 { t.Fatal("finished probe retained a slot or unbounded media cache", active, cached) } } if !optimized { part := embyRequest(app, http.MethodGet, embyFirstSegment(t, firstBody)) if part.Code != 200 || part.Body.Len() < 188 || part.Body.Bytes()[0] != 0x47 { t.Fatal("evicting an idle cache broke an existing HLS URL", part.Code, part.Body.String()) } } repeated := embyRequest(app, http.MethodGet, address(1)) if repeated.Code != first.Code || repeated.Body.String() != firstBody || repeated.Header().Get("Location") != firstLocation { t.Fatal("repeated episode did not reuse its prepared session", repeated.Code) } web := historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"}) if web.Code != 200 { t.Fatal("completed Emby scans blocked a new browser playback", web.Code, web.Body.String()) } }) } } type embyBlockingWriter struct { *httptest.ResponseRecorder entered chan struct{} resume chan struct{} once sync.Once } func (writer *embyBlockingWriter) Write(body []byte) (int, error) { writer.once.Do(func() { close(writer.entered) <-writer.resume }) return writer.ResponseRecorder.Write(body) } func TestEmbyActiveRequestsShareLimitAndReleaseOnCompletion(t *testing.T) { for _, optimized := range []bool{false, true} { t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) { app, address := embyScanFixture(t, optimized) first := embyRequest(app, http.MethodGet, address(1)) asset := first.Header().Get("Location") if !optimized { asset = embyFirstSegment(t, first.Body.String()) } writer := &embyBlockingWriter{ResponseRecorder: httptest.NewRecorder(), entered: make(chan struct{}), resume: make(chan struct{})} done := make(chan struct{}) go func() { defer close(done) app.routes().ServeHTTP(writer, httptest.NewRequest(http.MethodGet, "http://localhost"+asset, nil)) }() var resume sync.Once defer resume.Do(func() { close(writer.resume); <-done }) select { case <-writer.entered: case <-time.After(10 * time.Second): t.Fatal("media request never reached the response writer") } if second := embyRequest(app, http.MethodGet, address(2)); second.Code != 429 || second.Header().Get("Retry-After") == "" { t.Fatal("active requests bypassed the configured limit", second.Code) } if same := embyRequest(app, http.MethodGet, address(1)); same.Code != first.Code { t.Fatal("concurrent requests for the same episode were counted twice", same.Code) } web := historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"}) if web.Code != 429 { t.Fatal("browser playback bypassed an active Emby request", web.Code) } resume.Do(func() { close(writer.resume); <-done }) if writer.Code != 200 { t.Fatal("the original media request was interrupted", writer.Code) } if second := embyRequest(app, http.MethodGet, address(2)); second.Code != first.Code { t.Fatal("completed request did not release its slot", second.Code) } web = historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"}) if web.Code != 200 { t.Fatal("released Emby slot was not available to the browser", web.Code, web.Body.String()) } var opened struct{ Session string } if err := json.Unmarshal(web.Body.Bytes(), &opened); err != nil { t.Fatal(err) } app.closePlayback(opened.Session) }) } } func TestEmbyCanceledProbeReleasesPendingPreparation(t *testing.T) { started, stopped := make(chan struct{}), make(chan struct{}) var startOnce, stopOnce sync.Once d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) { startOnce.Do(func() { close(started) }) <-request.Context().Done() stopOnce.Do(func() { close(stopped) }) return nil, request.Context().Err() }) app := &UIApp{downloader: d, cfg: d.cfg} t.Cleanup(app.closePlaybacks) key, err := app.embySigningKey(true) if err != nil { t.Fatal(err) } id, chapter := "hongguo:7000000000000000001", "hongguo:7000000000000000001:1" query := url.Values{"id": {id}, "chapter": {chapter}, "key": {embyToken(key, id, chapter)}} ctx, cancel := context.WithCancel(context.Background()) defer cancel() done := make(chan struct{}) go func() { defer close(done) request := httptest.NewRequest(http.MethodGet, "http://localhost/api/emby/stream.m3u8?"+query.Encode(), nil).WithContext(ctx) app.handleEmbyStream(httptest.NewRecorder(), request) }() select { case <-started: case <-time.After(5 * time.Second): t.Fatal("preparation did not start") } other, cancelOther := context.WithCancel(context.Background()) defer cancelOther() otherDone := make(chan struct{}) go func() { defer close(otherDone) request := httptest.NewRequest(http.MethodGet, "http://localhost/api/emby/stream.m3u8?"+query.Encode(), nil).WithContext(other) app.handleEmbyStream(httptest.NewRecorder(), request) }() deadline := time.Now().Add(5 * time.Second) for { app.playbackMu.Lock() waiters := 0 for _, session := range app.playbacks { waiters += session.mediaWaiters } app.playbackMu.Unlock() if waiters == 2 { break } if time.Now().After(deadline) { t.Fatal("second request did not join the preparation") } time.Sleep(time.Millisecond) } cancel() select { case <-done: case <-time.After(5 * time.Second): t.Fatal("first canceled request did not finish") } select { case <-stopped: t.Fatal("one canceled client stopped another client's preparation") default: } cancelOther() for _, signal := range []chan struct{}{otherDone, stopped} { select { case <-signal: case <-time.After(5 * time.Second): t.Fatal("canceling the probe left its preparation running") } } app.playbackMu.Lock() defer app.playbackMu.Unlock() if len(app.playbacks) != 0 { t.Fatal("canceled probe retained a playback session") } } func TestEmbyEvictedRemuxRestoresExistingSegmentURLs(t *testing.T) { app, address := embyScanFixture(t, true) first := embyRequest(app, http.MethodGet, address(1)) if first.Code != http.StatusFound { t.Fatal(first.Code, first.Body.String()) } parsed, _ := url.Parse(first.Header().Get("Location")) base := strings.TrimSuffix(parsed.Path, "file.mp4") sessionID := strings.Split(base, "/")[4] app.playbackMu.Lock() session := app.playbacks[sessionID] media := session.media app.playbackMu.Unlock() ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() if err := app.generatePlayback(ctx, media, "remux", base, parsed.RawQuery); err != nil { t.Fatal(err) } segment := media.generated.assetURL(media.generated.segments[0].Name) before := embyRequest(app, http.MethodGet, segment) if before.Code != 200 { t.Fatal("prepared remux was unreadable", before.Code) } if next := embyRequest(app, http.MethodGet, address(2)); next.Code != http.StatusFound { t.Fatal("could not prepare another episode", next.Code) } if media.ctx.Err() == nil { t.Fatal("idle remux cache was not released") } after := embyRequest(app, http.MethodGet, segment) if after.Code != 200 || !bytes.Equal(before.Body.Bytes(), after.Body.Bytes()) { t.Fatal("reclaiming cached files changed or invalidated an existing segment", after.Code) } } func TestEmbyPlaybackCacheRemainsAccountScoped(t *testing.T) { for _, optimized := range []bool{false, true} { t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) { app, address := embyScanFixture(t, optimized) store := app.browserViewers().accountStore() store.mu.Lock() for _, owner := range []string{"owner-a", "owner-b"} { store.state.Accounts[owner] = accountRecord{ID: owner, Sources: []string{sourceHongguo}} } store.mu.Unlock() key, err := app.embySigningKey(false) if err != nil { t.Fatal(err) } sign := func(path, owner string) string { parsed, _ := url.Parse(path) query := parsed.Query() query.Set("account", owner) query.Set("key", embyToken(key, query.Get("id"), query.Get("chapter"), owner)) parsed.RawQuery = query.Encode() return parsed.String() } first := embyRequest(app, http.MethodGet, sign(address(1), "owner-a")) asset := first.Header().Get("Location") if !optimized { asset = embyFirstSegment(t, first.Body.String()) } if forged := embyRequest(app, http.MethodGet, sign(asset, "owner-b")); forged.Code != http.StatusGone { t.Fatal("valid credentials for another account reused a private cache", forged.Code) } second := embyRequest(app, http.MethodGet, sign(address(1), "owner-b")) if second.Code != first.Code { t.Fatal("second account could not prepare its own session", second.Code) } app.playbackMu.Lock() count := len(app.playbacks) app.playbackMu.Unlock() if count != 2 { t.Fatal("accounts shared a playback session", count) } }) } }