Test cleanup: split oversized test files, fix three weak tests (tests only)

pkg/moqt/session/datastream_subgroup_delta_test.go

@@ -0,0 +1,157 @@
+package session_test
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+
+	"github.com/floatdrop/moq-go/pkg/moqt/message"
+	"github.com/floatdrop/moq-go/pkg/moqt/session"
+)
+
+// TestWriteObjectAt: WriteObjectAt is the exact encoding inverse of
+// ReadDecoded — absolute Object IDs become the §11.4.2 deltas on the wire
+// (first object's delta = absolute ID; later = currentID-prevID-1).
+func TestWriteObjectAt(t *testing.T) {
+	cli, srv := openPair(t)
+	hdr := message.SubgroupHeader{TrackAlias: 42, GroupID: 7, SubgroupIDMode: message.SubgroupIDImplicitZero}
+
+	writeIDs := []uint64{4, 5, 9}
+	wantDeltas := []uint64{4, 0, 3}
+
+	in, writeErr := sendSubgroup(t, cli, srv, hdr, func(out *session.OutgoingSubgroupStream) error {
+		for i, id := range writeIDs {
+			if err := out.WriteObjectAt(id, &message.SubgroupObject{Payload: []byte{byte('a' + i)}}); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+
+	// Read the RAW objects: the exact wire deltas are the strongest check that
+	// the absolute→delta mapping is right.
+	for i := range writeIDs {
+		raw, err := in.ReadObject()
+		if err != nil {
+			t.Fatalf("ReadObject #%d: %v", i, err)
+		}
+		if raw.ObjectIDDelta != wantDeltas[i] {
+			t.Errorf("obj #%d: ObjectIDDelta got %d, want %d", i, raw.ObjectIDDelta, wantDeltas[i])
+		}
+		if string(raw.Payload) != string(byte('a'+i)) {
+			t.Errorf("obj #%d: payload got %q, want %q", i, raw.Payload, string(byte('a'+i)))
+		}
+	}
+	if _, err := in.ReadObject(); !errors.Is(err, io.EOF) {
+		t.Errorf("trailing ReadObject: got %v, want io.EOF", err)
+	}
+	if err := <-writeErr; err != nil {
+		t.Errorf("writer: %v", err)
+	}
+}
+
+// TestWriteObjectAtRejectsNonIncreasing: an Object ID not greater than the
+// previous one is rejected with ErrObjectIDNotIncreasing, nothing is written,
+// and the stream stays usable for a subsequent in-order write.
+func TestWriteObjectAtRejectsNonIncreasing(t *testing.T) {
+	cli, srv := openPair(t)
+	hdr := message.SubgroupHeader{TrackAlias: 1, GroupID: 0, SubgroupIDMode: message.SubgroupIDImplicitZero}
+
+	in, writeErr := sendSubgroup(t, cli, srv, hdr, func(out *session.OutgoingSubgroupStream) error {
+		if err := out.WriteObjectAt(5, &message.SubgroupObject{Payload: []byte("a")}); err != nil {
+			return err
+		}
+		for _, id := range []uint64{5, 3} { // equal, then lower
+			err := out.WriteObjectAt(id, &message.SubgroupObject{Payload: []byte("x")})
+			if !errors.Is(err, session.ErrObjectIDNotIncreasing) {
+				return fmt.Errorf("WriteObjectAt(%d) = %w, want ErrObjectIDNotIncreasing", id, err)
+			}
+		}
+		return out.WriteObjectAt(6, &message.SubgroupObject{Payload: []byte("b")})
+	})
+
+	// Only the two accepted objects reach the wire.
+	wantIDs := []uint64{5, 6}
+	wantPayloads := []string{"a", "b"}
+	for i := range wantIDs {
+		got, err := in.ReadDecoded()
+		if err != nil {
+			t.Fatalf("ReadDecoded #%d: %v", i, err)
+		}
+		if got.ObjectID != wantIDs[i] {
+			t.Errorf("obj #%d: ObjectID got %d, want %d", i, got.ObjectID, wantIDs[i])
+		}
+		if string(got.Payload) != wantPayloads[i] {
+			t.Errorf("obj #%d: payload got %q, want %q", i, got.Payload, wantPayloads[i])
+		}
+	}
+	if _, err := in.ReadDecoded(); !errors.Is(err, io.EOF) {
+		t.Errorf("trailing ReadDecoded: got %v, want io.EOF", err)
+	}
+	if err := <-writeErr; err != nil {
+		t.Errorf("writer: %v", err)
+	}
+}
+
+// TestIncomingSubgroupStream_ReadDecoded covers absolute ObjectID
+// reconstruction (first object's delta is the absolute ID; subsequent deltas
+// encode currentID - prevID - 1) and the three §11.4.2 SubgroupID modes.
+func TestIncomingSubgroupStream_ReadDecoded(t *testing.T) {
+	cases := []struct {
+		name       string
+		mode       message.SubgroupIDMode
+		explicitID uint64 // Explicit mode only
+		wantSubID  uint64
+	}{
+		{"ImplicitZero", message.SubgroupIDImplicitZero, 0, 0},
+		{"ImplicitFirstObject", message.SubgroupIDImplicitFirstObject, 0, 4 /* first abs ObjectID */},
+		{"Explicit", message.SubgroupIDExplicit, 99, 99},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			cli, srv := openPair(t)
+			hdr := message.SubgroupHeader{
+				TrackAlias:     42,
+				GroupID:        7,
+				SubgroupIDMode: tc.mode,
+				SubgroupID:     tc.explicitID,
+			}
+			// Absolute IDs 4, 5, 9: the first delta carries the absolute ID,
+			// the second is consecutive, the third skips 6/7/8.
+			in, writeErr := sendSubgroup(t, cli, srv, hdr, writeObjects(
+				&message.SubgroupObject{ObjectIDDelta: 4, Payload: []byte("a")},
+				&message.SubgroupObject{ObjectIDDelta: 0, Payload: []byte("b")},
+				&message.SubgroupObject{ObjectIDDelta: 3, Payload: []byte("c")},
+			))
+
+			wantIDs := []uint64{4, 5, 9}
+			wantPayloads := []string{"a", "b", "c"}
+			for i := range wantIDs {
+				got, err := in.ReadDecoded()
+				if err != nil {
+					t.Fatalf("ReadDecoded #%d: %v", i, err)
+				}
+				if got.GroupID != 7 {
+					t.Errorf("obj #%d: GroupID got %d, want 7", i, got.GroupID)
+				}
+				if got.ObjectID != wantIDs[i] {
+					t.Errorf("obj #%d: ObjectID got %d, want %d", i, got.ObjectID, wantIDs[i])
+				}
+				if got.SubgroupID != tc.wantSubID {
+					t.Errorf("obj #%d: SubgroupID got %d, want %d", i, got.SubgroupID, tc.wantSubID)
+				}
+				if string(got.Payload) != wantPayloads[i] {
+					t.Errorf("obj #%d: payload got %q, want %q", i, got.Payload, wantPayloads[i])
+				}
+			}
+			if _, err := in.ReadDecoded(); !errors.Is(err, io.EOF) {
+				t.Errorf("trailing ReadDecoded: got %v, want io.EOF", err)
+			}
+			if err := <-writeErr; err != nil {
+				t.Errorf("writer: %v", err)
+			}
+		})
+	}
+}

pkg/moqt/session/datastream_subgroup_test.go

@@ -0,0 +1,107 @@
+package session_test
+
+import (
+	"errors"
+	"io"
+	"testing"
+
+	"github.com/floatdrop/moq-go/pkg/moqt/message"
+)
+
+// TestSubgroupObjectReadRejectsInvalidStatus: ReadObject validates each decoded
+// object, so an empty payload with a status that is not
+// Normal/EndOfGroup/EndOfTrack is a protocol violation (§11.2.1.1), not a valid object.
+func TestSubgroupObjectReadRejectsInvalidStatus(t *testing.T) {
+	cli, srv := openPair(t)
+	hdr := message.SubgroupHeader{TrackAlias: 42, GroupID: 7, SubgroupIDMode: message.SubgroupIDImplicitZero}
+	// 0x2 is not a defined Object Status.
+	in, writeErr := sendSubgroup(t, cli, srv, hdr, writeObjects(&message.SubgroupObject{ObjectStatus: 0x2}))
+
+	if _, err := in.ReadObject(); err == nil {
+		t.Fatal("ReadObject must reject an object with an invalid status")
+	}
+	// The rejection closes the session under the writer, so its result is not checked.
+	<-writeErr
+}
+
+// TestSubgroupObjectRoundTrip: objects written with WriteObject read back
+// unchanged, the header's Properties flag reaching both ends without the caller
+// tracking it, and the closed stream then reads as io.EOF.
+func TestSubgroupObjectRoundTrip(t *testing.T) {
+	tests := []struct {
+		name string
+		hdr  message.SubgroupHeader
+		objs []*message.SubgroupObject
+	}{
+		{
+			name: "without properties",
+			hdr:  message.SubgroupHeader{TrackAlias: 42, GroupID: 7, SubgroupIDMode: message.SubgroupIDImplicitZero},
+			objs: []*message.SubgroupObject{
+				{ObjectIDDelta: 0, Payload: []byte("hello")},
+				{ObjectIDDelta: 0, Payload: []byte("world")},
+			},
+		},
+		{
+			name: "with properties",
+			hdr:  message.SubgroupHeader{TrackAlias: 1, GroupID: 0, Properties: true},
+			// One property: Type 2 (a varint value), value 7.
+			objs: []*message.SubgroupObject{{Properties: []byte{0x02, 0x07}, Payload: []byte("data")}},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			cli, srv := openPair(t)
+			in, writeErr := sendSubgroup(t, cli, srv, tc.hdr, writeObjects(tc.objs...))
+
+			if in.Header.TrackAlias != tc.hdr.TrackAlias || in.Header.GroupID != tc.hdr.GroupID {
+				t.Errorf("header mismatch: got %+v, want %+v", in.Header, tc.hdr)
+			}
+			for i, want := range tc.objs {
+				got, err := in.ReadObject()
+				if err != nil {
+					t.Fatalf("ReadObject(%d): %v", i, err)
+				}
+				if got.ObjectIDDelta != want.ObjectIDDelta {
+					t.Errorf("object %d delta: got %d, want %d", i, got.ObjectIDDelta, want.ObjectIDDelta)
+				}
+				if string(got.Properties) != string(want.Properties) {
+					t.Errorf("object %d properties: got %x, want %x", i, got.Properties, want.Properties)
+				}
+				if string(got.Payload) != string(want.Payload) {
+					t.Errorf("object %d payload: got %q, want %q", i, got.Payload, want.Payload)
+				}
+			}
+			if _, err := in.ReadObject(); !errors.Is(err, io.EOF) {
+				t.Errorf("ReadObject after close: got %v, want io.EOF", err)
+			}
+			if err := <-writeErr; err != nil {
+				t.Errorf("writer: %v", err)
+			}
+		})
+	}
+}
+
+// TestWriteObjectWrongType: the wrong object type is a compile error (distinct
+// WriteObject signatures); at runtime a zero-value object must not panic.
+func TestWriteObjectWrongType(t *testing.T) {
+	cli, srv := openPair(t)
+	ctx := t.Context()
+
+	// Drain the server side so the synchronous pipe doesn't deadlock the write.
+	go func() {
+		if ds, err := srv.AcceptDataStream(ctx); err == nil {
+			io.Copy(io.Discard, ds)
+		}
+	}()
+
+	outStream, err := cli.OpenSubgroup(message.SubgroupHeader{TrackAlias: 1, GroupID: 0})
+	if err != nil {
+		t.Fatalf("OpenSubgroup: %v", err)
+	}
+	defer outStream.Cancel(0)
+
+	// A nil payload is valid on the wire.
+	if err := outStream.WriteObject(&message.SubgroupObject{}); err != nil {
+		t.Errorf("WriteObject(zero SubgroupObject): unexpected error: %v", err)
+	}
+}

pkg/moqt/session/helpers_test.go

@@ -292,3 +292,48 @@ func drainOneSubgroup(t *testing.T, client *session.Session) {
 		}
 	}
 }
+
+// sendSubgroup opens a subgroup stream on from, runs write on it and closes it, and returns to's end with the
+// writer's result. The writer runs in a goroutine because the test pipe is synchronous: Write blocks until read.
+func sendSubgroup(
+	t *testing.T,
+	from, to *session.Session,
+	hdr message.SubgroupHeader,
+	write func(*session.OutgoingSubgroupStream) error,
+) (*session.IncomingSubgroupStream, <-chan error) {
+	t.Helper()
+	writeErr := make(chan error, 1)
+	go func() {
+		out, err := from.OpenSubgroup(hdr)
+		if err != nil {
+			writeErr <- err
+			return
+		}
+		if err := write(out); err != nil {
+			writeErr <- err
+			return
+		}
+		writeErr <- out.Close()
+	}()
+	ds, err := to.AcceptDataStream(t.Context())
+	if err != nil {
+		t.Fatalf("AcceptDataStream: %v", err)
+	}
+	in, ok := ds.(*session.IncomingSubgroupStream)
+	if !ok {
+		t.Fatalf("AcceptDataStream returned %T, want *session.IncomingSubgroupStream", ds)
+	}
+	return in, writeErr
+}
+
+// writeObjects is a sendSubgroup write func that writes objs in order.
+func writeObjects(objs ...*message.SubgroupObject) func(*session.OutgoingSubgroupStream) error {
+	return func(out *session.OutgoingSubgroupStream) error {
+		for _, o := range objs {
+			if err := out.WriteObject(o); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}