package app import ( "bytes" "context" "errors" "fmt" "io" "net/http" "net/http/httptest" "strconv" "strings" "sync/atomic" "testing" "time" ) func mediaProxyFixture(t *testing.T, handler http.HandlerFunc) (*hlsProxy, *http.Client) { t.Helper() server := httptest.NewServer(handler) t.Cleanup(server.Close) downloader := &Downloader{cfg: Config{Retries: 3}, client: server.Client()} proxy, err := downloader.newHLSProxy(context.Background(), providerMedia{URL: server.URL + "/fixture.mp4", Referer: "https://example.invalid/"}, nil) if err != nil { t.Fatal(err) } t.Cleanup(proxy.Close) client := &http.Client{Transport: &http.Transport{Proxy: nil}, Timeout: 10 * time.Second} t.Cleanup(client.CloseIdleConnections) return proxy, client } func TestMediaProxyResumesTruncatedResponse(t *testing.T) { payload := bytes.Repeat([]byte("0123456789abcdef"), 64) for _, test := range []struct { name string rangeHeader string start int end int }{ {name: "whole", start: 0, end: 1023}, {name: "bounded seek", rangeHeader: "bytes=100-699", start: 100, end: 699}, {name: "suffix seek", rangeHeader: "bytes=-129", start: 895, end: 1023}, } { t.Run(test.name, func(t *testing.T) { var calls atomic.Int32 proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { attempt := calls.Add(1) if request.Header.Get("Accept-Encoding") != "identity" || request.Header.Get("Referer") != "https://example.invalid/" { t.Error("media request lost source headers") } start, end, status := test.start, test.end, http.StatusOK if test.rangeHeader != "" { status = http.StatusPartialContent } if attempt == 1 { if request.Header.Get("Range") != test.rangeHeader { t.Error("initial seek range was changed") } } else { start += 13 status = http.StatusPartialContent want := fmt.Sprintf("bytes=%d-%d", start, end) if request.Header.Get("Range") != want || request.Header.Get("If-Range") != `"fixture-v1"` { t.Errorf("unsafe continuation: Range=%q If-Range=%q, want %s", request.Header.Get("Range"), request.Header.Get("If-Range"), want) } } writer.Header().Set("ETag", `"fixture-v1"`) writer.Header().Set("Content-Type", "video/mp4") writer.Header().Set("Content-Length", strconv.Itoa(end-start+1)) if status == http.StatusPartialContent { writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(payload))) } writer.WriteHeader(status) if attempt == 1 { _, _ = writer.Write(payload[start : start+13]) } else { _, _ = writer.Write(payload[start : end+1]) } }) request, _ := http.NewRequest(http.MethodGet, proxy.root, nil) request.Header.Set("Range", test.rangeHeader) response, err := client.Do(request) if err != nil { t.Fatal(err) } defer response.Body.Close() body, err := io.ReadAll(response.Body) if err != nil || !bytes.Equal(body, payload[test.start:test.end+1]) || calls.Load() != 2 || proxy.Err() != nil { t.Fatalf("interrupted media did not resume exactly: bytes=%d calls=%d read=%v proxy=%v", len(body), calls.Load(), err, proxy.Err()) } wantStatus := http.StatusOK if test.rangeHeader != "" { wantStatus = http.StatusPartialContent } if response.StatusCode != wantStatus || response.ContentLength != int64(test.end-test.start+1) { t.Fatal("continuation changed the local response headers") } }) } } func TestMediaProxyExhaustsBodyRetries(t *testing.T) { payload := bytes.Repeat([]byte("0123456789"), 10) var calls atomic.Int32 proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { attempt := int(calls.Add(1)) start := (attempt - 1) * 13 writer.Header().Set("Content-Length", strconv.Itoa(len(payload)-start)) writer.Header().Set("ETag", `"fixture-v1"`) if attempt > 1 { if request.Header.Get("Range") != fmt.Sprintf("bytes=%d-99", start) { t.Error("retry did not preserve its cumulative offset") } writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-99/100", start)) writer.WriteHeader(http.StatusPartialContent) } _, _ = writer.Write(payload[start : start+13]) }) response, err := client.Get(proxy.root) if err != nil { t.Fatal(err) } defer response.Body.Close() body, err := io.ReadAll(response.Body) if err == nil || calls.Load() != 3 || !bytes.Equal(body, payload[:39]) || proxy.Err() == nil { t.Fatalf("exhausted transfer was not bounded or reported: bytes=%d calls=%d read=%v proxy=%v", len(body), calls.Load(), err, proxy.Err()) } } func TestMediaProxyRejectsUnsafeContinuation(t *testing.T) { for _, test := range []struct { name, contentRange, etag, encoding string status, length int }{ {name: "range ignored", status: 200, length: 100, etag: `"fixture-v1"`}, {name: "wrong start", status: 206, length: 90, contentRange: "bytes 9-98/100", etag: `"fixture-v1"`}, {name: "wrong end", status: 206, length: 89, contentRange: "bytes 10-98/100", etag: `"fixture-v1"`}, {name: "changed length", status: 206, length: 90, contentRange: "bytes 10-99/101", etag: `"fixture-v1"`}, {name: "changed object", status: 206, length: 90, contentRange: "bytes 10-99/100", etag: `"fixture-v2"`}, {name: "missing validator", status: 206, length: 90, contentRange: "bytes 10-99/100"}, {name: "encoded bytes", status: 206, length: 90, contentRange: "bytes 10-99/100", etag: `"fixture-v1"`, encoding: "gzip"}, {name: "conflicting length", status: 206, length: 80, contentRange: "bytes 10-99/100", etag: `"fixture-v1"`}, } { t.Run(test.name, func(t *testing.T) { var calls atomic.Int32 proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { if calls.Add(1) == 1 { writer.Header().Set("Content-Length", "100") writer.Header().Set("ETag", `"fixture-v1"`) _, _ = io.WriteString(writer, "0123456789") return } writer.Header().Set("Content-Length", strconv.Itoa(test.length)) writer.Header().Set("Content-Range", test.contentRange) writer.Header().Set("ETag", test.etag) writer.Header().Set("Content-Encoding", test.encoding) writer.WriteHeader(test.status) _, _ = io.WriteString(writer, strings.Repeat("x", test.length)) }) response, err := client.Get(proxy.root) if err != nil { t.Fatal(err) } defer response.Body.Close() body, err := io.ReadAll(response.Body) if err == nil || string(body) != "0123456789" || calls.Load() != 2 || proxy.Err() == nil { t.Fatalf("unverified bytes were appended or failure was lost: bytes=%d calls=%d read=%v proxy=%v", len(body), calls.Load(), err, proxy.Err()) } }) } } func TestMediaProxyCanceledRequestDoesNotRecordFailure(t *testing.T) { proxy, _ := mediaProxyFixture(t, func(http.ResponseWriter, *http.Request) { t.Error("canceled request reached upstream") }) ctx, cancel := context.WithCancel(context.Background()) cancel() request := httptest.NewRequest(http.MethodGet, proxy.root, nil).WithContext(ctx) proxy.ServeHTTP(httptest.NewRecorder(), request) if proxy.Err() != nil { t.Fatal("stopping playback poisoned its media proxy", proxy.Err()) } } func TestMediaProxyCancelsDuringContinuation(t *testing.T) { started := make(chan struct{}) var calls atomic.Int32 proxy, _ := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { if calls.Add(1) == 1 { writer.Header().Set("Content-Length", "100") _, _ = io.WriteString(writer, "0123456789") return } close(started) <-request.Context().Done() }) ctx, cancel := context.WithCancel(context.Background()) defer cancel() done := make(chan struct{}) go func() { defer close(done) proxy.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, proxy.root, nil).WithContext(ctx)) }() select { case <-started: case <-time.After(3 * time.Second): t.Fatal("continuation never started") } cancel() select { case <-done: case <-time.After(time.Second): t.Fatal("continuation blocked playback cancellation") } if proxy.Err() != nil || calls.Load() != 2 { t.Fatal("cancellation was recorded as an upstream failure", proxy.Err()) } } func TestMediaProxyRetriesTemporaryContinuationStatus(t *testing.T) { const modified = "Mon, 14 Sep 2026 12:00:00 GMT" var calls atomic.Int32 proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { attempt := calls.Add(1) if attempt == 2 { writer.WriteHeader(http.StatusServiceUnavailable) return } writer.Header().Set("Last-Modified", modified) if attempt == 1 { writer.Header().Set("Content-Length", "20") _, _ = io.WriteString(writer, "0123456789") return } if request.Header.Get("Range") != "bytes=10-19" || request.Header.Get("If-Range") != modified { t.Error("date validator or retry offset was lost") } writer.Header().Set("Content-Length", "10") writer.Header().Set("Content-Range", "bytes 10-19/20") writer.WriteHeader(http.StatusPartialContent) _, _ = io.WriteString(writer, "abcdefghij") }) response, err := client.Get(proxy.root) if err != nil { t.Fatal(err) } defer response.Body.Close() body, err := io.ReadAll(response.Body) if err != nil || string(body) != "0123456789abcdefghij" || calls.Load() != 3 || proxy.Err() != nil { t.Fatal("temporary continuation failure was latched", err, proxy.Err()) } } func TestMediaProxyHeadAndUnknownLength(t *testing.T) { for _, mode := range []string{"head", "chunked", "broken chunked"} { t.Run(mode, func(t *testing.T) { var calls atomic.Int32 proxy, client := mediaProxyFixture(t, func(writer http.ResponseWriter, request *http.Request) { calls.Add(1) if mode == "head" { writer.Header().Set("Content-Length", "100") return } writer.WriteHeader(http.StatusOK) writer.(http.Flusher).Flush() _, _ = io.WriteString(writer, "unknown-length-fixture") writer.(http.Flusher).Flush() if mode == "broken chunked" { connection, _, err := writer.(http.Hijacker).Hijack() if err != nil { t.Error(err) return } _ = connection.Close() } }) method := http.MethodGet if mode == "head" { method = http.MethodHead } request, _ := http.NewRequest(method, proxy.root, nil) response, err := client.Do(request) if err != nil { t.Fatal(err) } defer response.Body.Close() body, _ := io.ReadAll(response.Body) if mode == "head" && len(body) != 0 || mode != "head" && string(body) != "unknown-length-fixture" || calls.Load() != 1 { t.Fatal("HEAD or unknown-length response changed") } if (proxy.Err() != nil) != (mode == "broken chunked") { t.Fatal("unverifiable truncation was hidden or a complete body was rejected", proxy.Err()) } }) } } type mediaFailureBody struct { data []byte } func (body *mediaFailureBody) Read(buffer []byte) (int, error) { count := copy(buffer, body.data) body.data = body.data[count:] return count, io.ErrUnexpectedEOF } func (*mediaFailureBody) Close() error { return nil } type mediaStoppedWriter struct{ header http.Header } func (writer *mediaStoppedWriter) Header() http.Header { return writer.header } func (*mediaStoppedWriter) WriteHeader(int) {} func (*mediaStoppedWriter) Write([]byte) (int, error) { return 0, errors.New("consumer stopped") } func TestMediaProxyConsumerStopDoesNotLatchPendingReadError(t *testing.T) { proxy, _ := mediaProxyFixture(t, func(http.ResponseWriter, *http.Request) { t.Error("fixture transport was bypassed") }) var calls int proxy.client.Transport = rankingTransport(func(request *http.Request) (*http.Response, error) { calls++ return &http.Response{StatusCode: 200, Header: http.Header{"Content-Length": {"100"}}, ContentLength: 100, Body: &mediaFailureBody{data: []byte("0123456789")}, Request: request}, nil }) proxy.ServeHTTP(&mediaStoppedWriter{header: make(http.Header)}, httptest.NewRequest(http.MethodGet, proxy.root, nil)) if calls != 1 || proxy.Err() != nil { t.Fatal("unused media bytes triggered a retry or false failure", proxy.Err()) } } func TestMediaReaderAcceptsExactlyCompletedBody(t *testing.T) { request := httptest.NewRequest(http.MethodGet, "http://fixture.invalid/media", nil) proxy := &hlsProxy{retries: 3} reader := proxy.mediaReader(request, &http.Response{StatusCode: 200, ContentLength: 10, Header: make(http.Header), Body: &mediaFailureBody{data: []byte("0123456789")}}) defer reader.Close() body, err := io.ReadAll(reader) if err != nil || string(body) != "0123456789" || reader.attempts != 0 { t.Fatal("completed response was retried", err) } }