datarhei-dragonfork-core/app/webrtc/handler_m3_test.go

package webrtc

import (
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"

	"github.com/labstack/echo/v4"

	corewebrtc "github.com/datarhei/core/v16/core/webrtc"
)

// minimalH264OpusOffer returns an SDP offer that includes both H264
// and Opus rtpmap lines — passes requireH264AndOpus but is otherwise
// nonsense, so CreatePeerFromSources will fail downstream when this
// is wired through. Use it only in tests that don't reach the
// PeerConnection path.
func minimalH264OpusOffer() string {
	return "v=0\r\n" +
		"o=- 0 0 IN IP4 0.0.0.0\r\ns=-\r\nt=0 0\r\n" +
		"m=video 9 UDP/TLS/RTP/SAVPF 102\r\n" +
		"a=rtpmap:102 H264/90000\r\n" +
		"m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n" +
		"a=rtpmap:111 opus/48000/2\r\n"
}

// nonH264Offer is missing H264 entirely. Triggers requireH264AndOpus.
func nonH264Offer() string {
	return "v=0\r\n" +
		"m=video 9 UDP/TLS/RTP/SAVPF 96\r\n" +
		"a=rtpmap:96 VP8/90000\r\n" +
		"m=audio 9 UDP/TLS/RTP/SAVPF 111\r\n" +
		"a=rtpmap:111 opus/48000/2\r\n"
}

// TestHandler_Subscribe_406OnCodecMismatch verifies an offer that
// doesn't include H264 yields 406, per the design's error matrix.
func TestHandler_Subscribe_406OnCodecMismatch(t *testing.T) {
	sub := newTestSubsystem(t)
	sub.mu.Lock()
	sub.streams["s"] = &processStream{id: "s"}
	sub.mu.Unlock()
	h := NewHandler(sub, 0)

	e := echo.New()
	req := httptest.NewRequest(http.MethodPost, "/whep/s", strings.NewReader(nonH264Offer()))
	rec := httptest.NewRecorder()
	c := e.NewContext(req, rec)
	c.SetParamNames("id")
	c.SetParamValues("s")

	if err := h.Subscribe(c); err != nil {
		t.Fatalf("Subscribe: %v", err)
	}
	if rec.Code != http.StatusNotAcceptable {
		t.Fatalf("expected 406, got %d: %s", rec.Code, rec.Body.String())
	}
	if !strings.Contains(rec.Body.String(), "H264") {
		t.Errorf("body should mention missing codec: %q", rec.Body.String())
	}
}

// TestHandler_Subscribe_503OnTotalCap simulates the total cap being
// exhausted by another subscriber. We don't actually create real peers
// (would need a real PeerConnection); instead we pre-load the atomic
// counter so the cap check fires.
func TestHandler_Subscribe_503OnTotalCap(t *testing.T) {
	sub := newTestSubsystem(t)
	sub.mu.Lock()
	sub.streams["s"] = &processStream{id: "s"}
	sub.mu.Unlock()
	h := NewHandlerWithCaps(sub, 1, 100)
	atomic.StoreInt64(&h.count, 1) // simulate one in-flight peer

	e := echo.New()
	req := httptest.NewRequest(http.MethodPost, "/whep/s", strings.NewReader(minimalH264OpusOffer()))
	rec := httptest.NewRecorder()
	c := e.NewContext(req, rec)
	c.SetParamNames("id")
	c.SetParamValues("s")
	_ = h.Subscribe(c)
	if rec.Code != http.StatusServiceUnavailable {
		t.Fatalf("expected 503, got %d: %s", rec.Code, rec.Body.String())
	}
	if !strings.Contains(rec.Body.String(), corewebrtc.ErrPeerCapReached.Error()) {
		t.Errorf("body should mention peer cap: %q", rec.Body.String())
	}
}

// TestHandler_Subscribe_503OnPerStreamCap simulates the per-stream cap
// being exhausted. Same trick as above but populating the per-stream
// index directly.
func TestHandler_Subscribe_503OnPerStreamCap(t *testing.T) {
	sub := newTestSubsystem(t)
	sub.mu.Lock()
	sub.streams["s"] = &processStream{id: "s"}
	sub.mu.Unlock()
	h := NewHandlerWithCaps(sub, 100, 1)
	// Drop a placeholder peer into the per-stream bucket so the cap
	// arithmetic trips on the next subscribe.
	h.mu.Lock()
	h.peersByStream["s"] = map[string]*corewebrtc.Peer{"existing": nil}
	h.mu.Unlock()

	e := echo.New()
	req := httptest.NewRequest(http.MethodPost, "/whep/s", strings.NewReader(minimalH264OpusOffer()))
	rec := httptest.NewRecorder()
	c := e.NewContext(req, rec)
	c.SetParamNames("id")
	c.SetParamValues("s")
	_ = h.Subscribe(c)
	if rec.Code != http.StatusServiceUnavailable {
		t.Fatalf("expected 503, got %d: %s", rec.Code, rec.Body.String())
	}
	if !strings.Contains(rec.Body.String(), "per-stream") {
		t.Errorf("body should mention per-stream cap: %q", rec.Body.String())
	}
}

// TestHandler_Trickle_404WhenUnknown verifies a PATCH for an unknown
// resource returns 404 (we still treat the resource as authoritative
// here; only DELETE is idempotent per spec).
func TestHandler_Trickle_404WhenUnknown(t *testing.T) {
	h := NewHandler(newTestSubsystem(t), 0)

	e := echo.New()
	req := httptest.NewRequest(http.MethodPatch, "/whep/id/unknown", strings.NewReader(""))
	rec := httptest.NewRecorder()
	c := e.NewContext(req, rec)
	c.SetParamNames("id", "resource")
	c.SetParamValues("id", "unknown")

	if err := h.Trickle(c); err != nil {
		t.Fatalf("Trickle: %v", err)
	}
	if rec.Code != http.StatusNotFound {
		t.Fatalf("expected 404, got %d", rec.Code)
	}
}

// TestHandler_PreflightCORS verifies OPTIONS returns 204 with the
// browser-friendly CORS headers.
func TestHandler_PreflightCORS(t *testing.T) {
	h := NewHandler(newTestSubsystem(t), 0)

	e := echo.New()
	req := httptest.NewRequest(http.MethodOptions, "/whep/x", nil)
	rec := httptest.NewRecorder()
	c := e.NewContext(req, rec)
	c.SetParamNames("id")
	c.SetParamValues("x")

	if err := h.preflight(c); err != nil {
		t.Fatalf("preflight: %v", err)
	}
	if rec.Code != http.StatusNoContent {
		t.Fatalf("expected 204, got %d", rec.Code)
	}
	hh := rec.Header()
	for _, k := range []string{
		"Access-Control-Allow-Origin",
		"Access-Control-Allow-Methods",
		"Access-Control-Allow-Headers",
		"Access-Control-Expose-Headers",
	} {
		if hh.Get(k) == "" {
			t.Errorf("missing CORS header %q", k)
		}
	}
}

// TestHandler_RegisterMountsAllRoutes is a sanity check that
// Handler.Register installs OPTIONS / POST / DELETE / PATCH on the
// expected paths. Echo's Group has no public route enumerator, so we
// dispatch synthetic requests and assert the right methods are
// reachable.
func TestHandler_RegisterMountsAllRoutes(t *testing.T) {
	h := NewHandler(newTestSubsystem(t), 0)
	e := echo.New()
	g := e.Group("")
	h.Register(g)

	cases := []struct {
		method, path string
		want         int
	}{
		{http.MethodOptions, "/whep/foo", http.StatusNoContent},
		{http.MethodOptions, "/whep/foo/bar", http.StatusNoContent},
		{http.MethodPost, "/whep/foo", http.StatusNotFound}, // stream missing -> 404
		{http.MethodDelete, "/whep/foo/bar", http.StatusNoContent},
		{http.MethodPatch, "/whep/foo/bar", http.StatusNotFound},
	}
	for _, tc := range cases {
		req := httptest.NewRequest(tc.method, tc.path, strings.NewReader(""))
		rec := httptest.NewRecorder()
		e.ServeHTTP(rec, req)
		if rec.Code != tc.want {
			t.Errorf("%s %s: got %d want %d (%s)", tc.method, tc.path, rec.Code, tc.want, rec.Body.String())
		}
	}
}

// TestHandler_Close_DrainsPeers seeds a fake peer into the index and
// verifies Close clears it without panicking.
func TestHandler_Close_DrainsPeers(t *testing.T) {
	h := NewHandler(newTestSubsystem(t), 0)
	h.mu.Lock()
	h.peersByStream["s"] = map[string]*corewebrtc.Peer{"r1": nil}
	h.peerStream["r1"] = "s"
	atomic.StoreInt64(&h.count, 1)
	h.mu.Unlock()

	h.Close()
	if got := atomic.LoadInt64(&h.count); got != 0 {
		t.Errorf("count after Close = %d, want 0", got)
	}
	h.mu.Lock()
	if len(h.peersByStream) != 0 || len(h.peerStream) != 0 {
		t.Errorf("indexes not cleared")
	}
	h.mu.Unlock()
}

// TestRequireH264AndOpus covers the SDP scanner's positive +
// negative cases.
func TestRequireH264AndOpus(t *testing.T) {
	cases := []struct {
		name string
		sdp  string
		ok   bool
	}{
		{"both", minimalH264OpusOffer(), true},
		{"missing h264", nonH264Offer(), false},
		{"missing opus", "m=video 9 UDP/TLS/RTP/SAVPF 102\r\na=rtpmap:102 H264/90000\r\n", false},
		{"capitalized", "a=rtpmap:111 OPUS/48000\r\na=rtpmap:102 H264/90000", true},
		{"empty", "", false},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			err := requireH264AndOpus(c.sdp)
			if c.ok && err != nil {
				t.Errorf("expected ok, got %v", err)
			}
			if !c.ok && err == nil {
				t.Errorf("expected error")
			}
		})
	}
}