test(core): cover video source buffer ownership, loop and memfd

commit: e43eec44cdcc6ac06e5498d65372987f8ea4b0e8
author: brice · 8d ago · 2026-10-14 19:28 UTC
parent: 6d3622a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +613 -0

A internal/core/shm_test.go +38 -0

@@ -0,0 +1,38 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"os"
	5	+	"strconv"
	6	+	"syscall"
	7	+	"testing"
	8	+	)
	9	+
	10	+	func TestMemfdShmRoundTrip(t *testing.T) {
	11	+	fd, data, err := memfdShm{}.NewSharedMem("pw-video-test", 4096)
	12	+	if err != nil {
	13	+	t.Skipf("memfd_create unavailable: %v", err)
	14	+	}
	15	+	if len(data) != 4096 {
	16	+	t.Fatalf("len = %d", len(data))
	17	+	}
	18	+	data[0], data[4095] = 1, 2
	19	+	// Re-map through the fd: the memory is shared and sealed against resizing.
	20	+	again, err := syscall.Mmap(fd, 0, 4096, syscall.PROT_READ, syscall.MAP_SHARED)
	21	+	if err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	if again[0] != 1 \|\| again[4095] != 2 {
	25	+	t.Fatal("memfd is not shared")
	26	+	}
	27	+	if err := syscall.Ftruncate(fd, 8192); err == nil {
	28	+	t.Fatal("sealed memfd was resized")
	29	+	}
	30	+	_ = syscall.Munmap(again)
	31	+	memfdShm{}.FreeSharedMem(fd, data)
	32	+	if _, err := os.Stat("/proc/self/fd/" + strconv.Itoa(fd)); err == nil {
	33	+	t.Fatal("fd still open after FreeSharedMem")
	34	+	}
	35	+	if _, _, err := (memfdShm{}).NewSharedMem("x", 0); err == nil {
	36	+	t.Fatal("zero size accepted")
	37	+	}
	38	+	}

@@ -0,0 +1,38 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"os"
	5	+	"strconv"
	6	+	"syscall"
	7	+	"testing"
	8	+	)
	9	+
	10	+	func TestMemfdShmRoundTrip(t *testing.T) {
	11	+	fd, data, err := memfdShm{}.NewSharedMem("pw-video-test", 4096)
	12	+	if err != nil {
	13	+	t.Skipf("memfd_create unavailable: %v", err)
	14	+	}
	15	+	if len(data) != 4096 {
	16	+	t.Fatalf("len = %d", len(data))
	17	+	}
	18	+	data[0], data[4095] = 1, 2
	19	+	// Re-map through the fd: the memory is shared and sealed against resizing.
	20	+	again, err := syscall.Mmap(fd, 0, 4096, syscall.PROT_READ, syscall.MAP_SHARED)
	21	+	if err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	if again[0] != 1 \|\| again[4095] != 2 {
	25	+	t.Fatal("memfd is not shared")
	26	+	}
	27	+	if err := syscall.Ftruncate(fd, 8192); err == nil {
	28	+	t.Fatal("sealed memfd was resized")
	29	+	}
	30	+	_ = syscall.Munmap(again)
	31	+	memfdShm{}.FreeSharedMem(fd, data)
	32	+	if _, err := os.Stat("/proc/self/fd/" + strconv.Itoa(fd)); err == nil {
	33	+	t.Fatal("fd still open after FreeSharedMem")
	34	+	}
	35	+	if _, _, err := (memfdShm{}).NewSharedMem("x", 0); err == nil {
	36	+	t.Fatal("zero size accepted")
	37	+	}
	38	+	}

A internal/core/video_test.go +575 -0

@@ -0,0 +1,575 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"time"
	7	+	"unsafe"
	8	+
	9	+	"github.com/stretchr/testify/mock"
	10	+
	11	+	portinmocks "github.com/bnema/purego-pipewire/internal/ports/in/mocks"
	12	+	portmocks "github.com/bnema/purego-pipewire/internal/ports/out/mocks"
	13	+	"github.com/bnema/purego-pipewire/spa"
	14	+	)
	15	+
	16	+	func TestVideoStructLayouts(t *testing.T) {
	17	+	var pb videoPWBuffer
	18	+	if unsafe.Offsetof(pb.Buffer) != 0 \|\| unsafe.Offsetof(pb.UserData) != 8 \|\| unsafe.Offsetof(pb.Size) != 16 {
	19	+	t.Errorf("pw_buffer layout: user_data@%d size@%d", unsafe.Offsetof(pb.UserData), unsafe.Offsetof(pb.Size))
	20	+	}
	21	+	var sb spaBuffer
	22	+	if unsafe.Offsetof(sb.NMetas) != 0 \|\| unsafe.Offsetof(sb.NDatas) != 4 \|\| unsafe.Offsetof(sb.Metas) != 8 \|\| unsafe.Offsetof(sb.Datas) != 16 {
	23	+	t.Errorf("spa_buffer layout wrong")
	24	+	}
	25	+	var m spaMeta
	26	+	if unsafe.Sizeof(m) != 16 \|\| unsafe.Offsetof(m.Data) != 8 {
	27	+	t.Errorf("spa_meta layout: size %d", unsafe.Sizeof(m))
	28	+	}
	29	+	var h spaMetaHeader
	30	+	if unsafe.Sizeof(h) != 32 \|\| unsafe.Offsetof(h.PTS) != 8 \|\| unsafe.Offsetof(h.DTSOffset) != 16 \|\| unsafe.Offsetof(h.Seq) != 24 {
	31	+	t.Errorf("spa_meta_header layout: size %d", unsafe.Sizeof(h))
	32	+	}
	33	+	var c spaChunk
	34	+	if unsafe.Sizeof(c) != 16 \|\| unsafe.Offsetof(c.Stride) != 8 \|\| unsafe.Offsetof(c.Flags) != 12 {
	35	+	t.Errorf("spa_chunk layout wrong")
	36	+	}
	37	+	var d spaData
	38	+	if unsafe.Sizeof(d) != 40 \|\| unsafe.Offsetof(d.Fd) != 8 \|\| unsafe.Offsetof(d.MapOffset) != 16 \|\|
	39	+	unsafe.Offsetof(d.Maxsize) != 20 \|\| unsafe.Offsetof(d.Data) != 24 \|\| unsafe.Offsetof(d.Chunk) != 32 {
	40	+	t.Errorf("spa_data layout: size %d", unsafe.Sizeof(d))
	41	+	}
	42	+	var e videoEvents
	43	+	for name, want := range map[string][2]uintptr{
	44	+	"state_changed": {unsafe.Offsetof(e.StateChanged), 16},
	45	+	"param_changed": {unsafe.Offsetof(e.ParamChanged), 40},
	46	+	"add_buffer": {unsafe.Offsetof(e.AddBuffer), 48},
	47	+	"remove_buffer": {unsafe.Offsetof(e.RemoveBuf), 56},
	48	+	"process": {unsafe.Offsetof(e.Process), 64},
	49	+	"trigger_done": {unsafe.Offsetof(e.TriggerDone), 88},
	50	+	} {
	51	+	if want[0] != want[1] {
	52	+	t.Errorf("pw_stream_events.%s at %d, want %d", name, want[0], want[1])
	53	+	}
	54	+	}
	55	+	var s VideoSource
	56	+	if unsafe.Sizeof(s.hook) < 48 {
	57	+	t.Errorf("spa_hook storage %d bytes, need 48", unsafe.Sizeof(s.hook))
	58	+	}
	59	+	}
	60	+
	61	+	// videoRig is a VideoSource wired to mocked C calls and mocked shared memory,
	62	+	// with a fake negotiated format. C-side structures are Go memory, which is
	63	+	// enough because the production code only reads them through pointers.
	64	+	type videoRig struct {
	65	+	t *testing.T
	66	+	api *portmocks.MockVideoStreamAPI
	67	+	shm *portmocks.MockSharedMemAPI
	68	+	handler *portinmocks.MockVideoHandler
	69	+	src *VideoSource
	70	+	stream unsafe.Pointer
	71	+	streamB [8]byte
	72	+	}
	73	+
	74	+	const rigW, rigH = 4, 2 // stride 16, size 32
	75	+
	76	+	func newRig(t testing.T) videoRig {
	77	+	t.Helper()
	78	+	r := &videoRig{
	79	+	t: t,
	80	+	api: portmocks.NewMockVideoStreamAPI(t),
	81	+	shm: portmocks.NewMockSharedMemAPI(t),
	82	+	handler: portinmocks.NewMockVideoHandler(t),
	83	+	}
	84	+	r.stream = unsafe.Pointer(&r.streamB[0])
	85	+	core := &Core{ptr: unsafe.Pointer(&r.streamB[4]), capi: r.api, shm: r.shm}
	86	+
	87	+	var gotProps *byte
	88	+	r.api.EXPECT().PWPropertiesNewString(mock.Anything).Run(func(args *byte) { gotProps = args }).Return(unsafe.Pointer(&r.streamB[1])).Once()
	89	+	r.api.EXPECT().PWStreamNew(core.ptr, mock.Anything, unsafe.Pointer(&r.streamB[1])).Return(r.stream).Once()
	90	+	var events *videoEvents
	91	+	r.api.EXPECT().PWStreamAddListener(r.stream, mock.Anything, mock.Anything, mock.Anything).
	92	+	Run(func(_, _, ev, _ unsafe.Pointer) { events = (*videoEvents)(ev) }).Return(0).Once()
	93	+	var connectFlags uint32
	94	+	r.api.EXPECT().PWStreamConnect(r.stream, int32(1), uint32(spa.IDInvalid), mock.Anything, mock.Anything, uint32(1)).
	95	+	Run(func(_ unsafe.Pointer, _ int32, _ uint32, flags uint32, _ unsafe.Pointer, _ uint32) {
	96	+	connectFlags = flags
	97	+	}).
	98	+	Return(0).Once()
	99	+
	100	+	src, err := core.NewVideoSource(VideoConfig{Name: "test-src", Width: rigW, Height: rigH}, r.handler)
	101	+	if err != nil {
	102	+	t.Fatal(err)
	103	+	}
	104	+	r.src = src
	105	+	t.Cleanup(func() {
	106	+	if r.src.table >= 0 {
	107	+	videoTable[r.src.table].Store(nil)
	108	+	}
	109	+	})
	110	+
	111	+	if connectFlags != pwStreamFlagDriver\|pwStreamFlagAllocBuffers {
	112	+	t.Fatalf("connect flags = %#x, want DRIVER\|ALLOC_BUFFERS", connectFlags)
	113	+	}
	114	+	if events == nil \|\| events.Version != 2 \|\| events.StateChanged == 0 \|\| events.ParamChanged == 0 \|\|
	115	+	events.AddBuffer == 0 \|\| events.RemoveBuf == 0 \|\| events.Process == 0 \|\| events.TriggerDone == 0 {
	116	+	t.Fatalf("events not fully registered: %+v", events)
	117	+	}
	118	+	if got := cString(unsafe.Pointer(gotProps)); got != `media.class=Video/Source node.name="test-src"` {
	119	+	t.Fatalf("props = %q", got)
	120	+	}
	121	+	if videoTable[src.table].Load() != src {
	122	+	t.Fatal("source not in callback slot table")
	123	+	}
	124	+	return r
	125	+	}
	126	+
	127	+	// negotiate feeds a Format param to the source, as PipeWire would.
	128	+	func (r *videoRig) negotiate() {
	129	+	r.t.Helper()
	130	+	var dst [256]byte
	131	+	b := spa.NewPodBuilder(dst[:])
	132	+	b.Object(spa.TypeObjectFormat, spa.ParamFormat)
	133	+	b.Prop(spa.FormatKeyMediaType)
	134	+	b.Id(spa.MediaTypeVideo)
	135	+	b.Prop(spa.FormatKeyMediaSubtype)
	136	+	b.Id(spa.MediaSubtypeRaw)
	137	+	b.Prop(spa.FormatKeyVideoFormat)
	138	+	b.Id(uint32(spa.VideoFormatBGRx))
	139	+	b.Prop(spa.FormatKeyVideoSize)
	140	+	b.Rectangle(rigW, rigH)
	141	+	b.End()
	142	+	pod, err := b.Bytes()
	143	+	if err != nil {
	144	+	r.t.Fatal(err)
	145	+	}
	146	+	var params unsafe.Pointer
	147	+	r.api.EXPECT().PWStreamUpdateParams(r.stream, mock.Anything, uint32(2)).
	148	+	Run(func(_, p unsafe.Pointer, _ uint32) { params = p }).Return(0).Once()
	149	+	r.handler.EXPECT().FormatChanged(spa.VideoFormatBGRx, rigW, rigH, rigW*4).Once()
	150	+	r.src.onParamChanged(spa.ParamFormat, unsafe.Pointer(&pod[0]))
	151	+	if params != unsafe.Pointer(&r.src.params[0]) {
	152	+	r.t.Fatal("update_params not given the slot-owned params array")
	153	+	}
	154	+	}
	155	+
	156	+	// fakeBuf is a PipeWire-side buffer: pw_buffer + spa_buffer + data + chunk + header meta.
	157	+	type fakeBuf struct {
	158	+	pw videoPWBuffer
	159	+	sb spaBuffer
	160	+	data spaData
	161	+	chunk spaChunk
	162	+	meta spaMeta
	163	+	hdr spaMetaHeader
	164	+	}
	165	+
	166	+	func newFakeBuf(dataType uint32) *fakeBuf {
	167	+	f := &fakeBuf{}
	168	+	f.data.Type = dataType
	169	+	f.data.Chunk = &f.chunk
	170	+	f.meta = spaMeta{Type: spa.MetaTypeHeader, Size: spa.MetaHeaderSize, Data: unsafe.Pointer(&f.hdr)}
	171	+	f.sb = spaBuffer{NMetas: 1, NDatas: 1, Metas: unsafe.Pointer(&f.meta), Datas: unsafe.Pointer(&f.data)}
	172	+	f.pw.Buffer = &f.sb
	173	+	return f
	174	+	}
	175	+
	176	+	func (f *fakeBuf) ptr() unsafe.Pointer { return unsafe.Pointer(&f.pw) }
	177	+
	178	+	// add runs add_buffer with a memfd that the shm mock "creates".
	179	+	func (r videoRig) add(f fakeBuf, fd int) []byte {
	180	+	r.t.Helper()
	181	+	mem := make([]byte, rigW4rigH)
	182	+	r.shm.EXPECT().NewSharedMem("pw-video", rigW4rigH).Return(fd, mem, nil).Once()
	183	+	r.handler.EXPECT().BuffersChanged().Once()
	184	+	r.src.onAddBuffer(f.ptr())
	185	+	return mem
	186	+	}
	187	+
	188	+	func TestVideoSourceNegotiationAndAddBuffer(t *testing.T) {
	189	+	r := newRig(t)
	190	+	r.negotiate()
	191	+	f := newFakeBuf(1 << spa.DataMemFd)
	192	+	mem := r.add(f, 42)
	193	+
	194	+	if f.data.Type != spa.DataMemFd \|\| f.data.Flags != spa.DataFlagReadWrite \|\| f.data.Fd != 42 \|\|
	195	+	f.data.MapOffset != 0 \|\| f.data.Maxsize != uint32(len(mem)) \|\| f.data.Data != unsafe.Pointer(&mem[0]) {
	196	+	t.Fatalf("spa_data not filled: %+v", f.data)
	197	+	}
	198	+	if f.chunk.Offset != 0 \|\| f.chunk.Size != uint32(len(mem)) \|\| f.chunk.Stride != rigW*4 {
	199	+	t.Fatalf("chunk = %+v", f.chunk)
	200	+	}
	201	+	if f.pw.UserData != 1 {
	202	+	t.Fatalf("user_data = %d, want slot 0 + 1", f.pw.UserData)
	203	+	}
	204	+
	205	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	206	+	b := r.src.Dequeue()
	207	+	if b == nil \|\| len(b.Data) != len(mem) \|\| &b.Data[0] != &mem[0] \|\| b.Stride != rigW*4 \|\| b.Width != rigW \|\| b.Height != rigH \|\| b.Format != spa.VideoFormatBGRx {
	208	+	t.Fatalf("Dequeue = %+v", b)
	209	+	}
	210	+
	211	+	var queued unsafe.Pointer
	212	+	r.api.EXPECT().PWStreamQueueBuffer(r.stream, f.ptr()).Run(func(_, p unsafe.Pointer) { queued = p }).Return(0).Once()
	213	+	r.src.Queue(b, 7, 1500*time.Millisecond, false)
	214	+	if queued != f.ptr() {
	215	+	t.Fatal("buffer not queued")
	216	+	}
	217	+	if f.chunk.Size != uint32(len(mem)) \|\| f.chunk.Stride != rigW*4 \|\| f.chunk.Flags != 0 {
	218	+	t.Fatalf("chunk after Queue = %+v", f.chunk)
	219	+	}
	220	+	if f.hdr.Seq != 7 \|\| f.hdr.PTS != int64(1500*time.Millisecond) \|\| f.hdr.Flags != 0 {
	221	+	t.Fatalf("header after Queue = %+v", f.hdr)
	222	+	}
	223	+
	224	+	// Second round, corrupted.
	225	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	226	+	b = r.src.Dequeue()
	227	+	if b == nil {
	228	+	t.Fatal("buffer not re-dequeued")
	229	+	}
	230	+	r.api.EXPECT().PWStreamQueueBuffer(r.stream, f.ptr()).Return(0).Once()
	231	+	r.src.Queue(b, 8, 0, true)
	232	+	if f.chunk.Size != 0 \|\| f.chunk.Flags != spa.ChunkFlagCorrupted \|\| f.hdr.Flags != spa.MetaHeaderFlagCorrupted \|\| f.hdr.Seq != 8 {
	233	+	t.Fatalf("corrupted frame: chunk %+v hdr %+v", f.chunk, f.hdr)
	234	+	}
	235	+
	236	+	// Trigger is separate and goes straight to the C call.
	237	+	r.api.EXPECT().PWStreamTriggerProcess(r.stream).Return(0).Once()
	238	+	r.src.Trigger()
	239	+
	240	+	r.expectDestroy(1)
	241	+	r.src.Destroy()
	242	+	}
	243	+
	244	+	// expectDestroy expects the disconnect/destroy sequence and the free of n live slots.
	245	+	func (r *videoRig) expectDestroy(n int) {
	246	+	r.api.EXPECT().PWStreamDisconnect(r.stream).Return(0).Once()
	247	+	r.api.EXPECT().PWStreamDestroy(r.stream).Once()
	248	+	if n > 0 {
	249	+	r.shm.EXPECT().FreeSharedMem(mock.Anything, mock.Anything).Times(n)
	250	+	}
	251	+	}
	252	+
	253	+	func TestVideoSourceMemfdFailureMakesBufferUnusable(t *testing.T) {
	254	+	r := newRig(t)
	255	+	r.negotiate()
	256	+
	257	+	bad := newFakeBuf(1 << spa.DataMemFd)
	258	+	r.shm.EXPECT().NewSharedMem("pw-video", rigW4rigH).Return(-1, nil, errors.New("memfd_create: boom")).Once()
	259	+	r.src.onAddBuffer(bad.ptr())
	260	+	if bad.pw.UserData != 0 {
	261	+	t.Fatalf("failed buffer got user_data %d", bad.pw.UserData)
	262	+	}
	263	+
	264	+	good := newFakeBuf(1 << spa.DataMemFd)
	265	+	mem := r.add(good, 7)
	266	+
	267	+	// Dequeue skips the unusable buffer and returns the good one.
	268	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(bad.ptr()).Once()
	269	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(good.ptr()).Once()
	270	+	b := r.src.Dequeue()
	271	+	if b == nil \|\| &b.Data[0] != &mem[0] {
	272	+	t.Fatalf("Dequeue = %+v, want the good buffer", b)
	273	+	}
	274	+	r.api.EXPECT().PWStreamReturnBuffer(r.stream, good.ptr()).Return(0).Once()
	275	+	r.src.Release(b)
	276	+
	277	+	// A DMA-BUF-only buffer is also unusable and allocates nothing.
	278	+	dma := newFakeBuf(1 << spa.DataDmaBuf)
	279	+	r.src.onAddBuffer(dma.ptr())
	280	+	if dma.pw.UserData != 0 {
	281	+	t.Fatalf("dmabuf buffer got user_data %d", dma.pw.UserData)
	282	+	}
	283	+
	284	+	// Nothing is dequeueable and no buffer is available: nil, no panic.
	285	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(nil).Once()
	286	+	if b := r.src.Dequeue(); b != nil {
	287	+	t.Fatalf("Dequeue = %+v, want nil", b)
	288	+	}
	289	+	r.expectDestroy(1)
	290	+	r.src.Destroy()
	291	+	}
	292	+
	293	+	func TestVideoSourceRemoveBorrowedKeepsMappingUntilRelease(t *testing.T) {
	294	+	r := newRig(t)
	295	+	r.negotiate()
	296	+	f := newFakeBuf(1 << spa.DataMemFd)
	297	+	mem := r.add(f, 9)
	298	+
	299	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	300	+	b := r.src.Dequeue()
	301	+	if b == nil {
	302	+	t.Fatal("no buffer")
	303	+	}
	304	+
	305	+	// PipeWire removes the buffer while the owner still holds it: no unmap yet
	306	+	// (FreeSharedMem has no expectation, so a call would fail the test).
	307	+	r.handler.EXPECT().BuffersChanged().Once()
	308	+	r.src.onRemoveBuffer(f.ptr())
	309	+	b.Data[0] = 0xAB // still writable
	310	+	if mem[0] != 0xAB \|\| f.pw.UserData != 0 {
	311	+	t.Fatalf("mapping not kept: mem[0]=%x user_data=%d", mem[0], f.pw.UserData)
	312	+	}
	313	+
	314	+	// A removed slot is never returned by Dequeue (PipeWire no longer owns it,
	315	+	// so even if it showed up the source would not hand it out).
	316	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	317	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(nil).Once()
	318	+	if got := r.src.Dequeue(); got != nil {
	319	+	t.Fatalf("removed buffer dequeued: %+v", got)
	320	+	}
	321	+
	322	+	// Queue on a removed slot must not call pw_stream_queue_buffer, and
	323	+	// releases the mapping.
	324	+	r.shm.EXPECT().FreeSharedMem(9, mem).Once()
	325	+	r.api.AssertNotCalled(t, "PWStreamQueueBuffer", mock.Anything, mock.Anything)
	326	+	r.src.Queue(b, 1, 0, false)
	327	+	r.api.AssertNotCalled(t, "PWStreamQueueBuffer", mock.Anything, mock.Anything)
	328	+
	329	+	// The slot is reusable afterwards.
	330	+	f2 := newFakeBuf(1 << spa.DataMemFd)
	331	+	r.add(f2, 10)
	332	+	if f2.pw.UserData != 1 {
	333	+	t.Fatalf("slot not reused, user_data = %d", f2.pw.UserData)
	334	+	}
	335	+	r.expectDestroy(1)
	336	+	r.src.Destroy()
	337	+	}
	338	+
	339	+	func TestVideoSourceRemoveBorrowedThenRelease(t *testing.T) {
	340	+	r := newRig(t)
	341	+	r.negotiate()
	342	+	f := newFakeBuf(1 << spa.DataMemFd)
	343	+	mem := r.add(f, 11)
	344	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	345	+	b := r.src.Dequeue()
	346	+
	347	+	r.handler.EXPECT().BuffersChanged().Once()
	348	+	r.src.onRemoveBuffer(f.ptr())
	349	+	r.shm.EXPECT().FreeSharedMem(11, mem).Once()
	350	+	r.src.Release(b)
	351	+	r.api.AssertNotCalled(t, "PWStreamReturnBuffer", mock.Anything, mock.Anything)
	352	+
	353	+	// A second Release/Queue of the same buffer is ignored.
	354	+	r.src.Release(b)
	355	+	r.src.Queue(b, 0, 0, false)
	356	+	r.expectDestroy(0)
	357	+	r.src.Destroy()
	358	+	}
	359	+
	360	+	func TestVideoSourceRemoveIdleUnmapsImmediately(t *testing.T) {
	361	+	r := newRig(t)
	362	+	r.negotiate()
	363	+	f := newFakeBuf(1 << spa.DataMemFd)
	364	+	mem := r.add(f, 12)
	365	+	r.shm.EXPECT().FreeSharedMem(12, mem).Once()
	366	+	r.handler.EXPECT().BuffersChanged().Once()
	367	+	r.src.onRemoveBuffer(f.ptr())
	368	+	r.expectDestroy(0)
	369	+	r.src.Destroy()
	370	+	}
	371	+
	372	+	func TestVideoSourceDestroyKeepsBorrowedMappingUntilRelease(t *testing.T) {
	373	+	r := newRig(t)
	374	+	r.negotiate()
	375	+	f := newFakeBuf(1 << spa.DataMemFd)
	376	+	mem := r.add(f, 13)
	377	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	378	+	b := r.src.Dequeue()
	379	+
	380	+	r.expectDestroy(0)
	381	+	r.src.Destroy()
	382	+	b.Data[1] = 1 // still writable after Destroy
	383	+	if mem[1] != 1 {
	384	+	t.Fatal("mapping lost on Destroy")
	385	+	}
	386	+	r.shm.EXPECT().FreeSharedMem(13, mem).Once()
	387	+	r.src.Release(b)
	388	+
	389	+	// Destroy is idempotent and Dequeue/Trigger after it are no-ops.
	390	+	r.src.Destroy()
	391	+	r.src.Trigger()
	392	+	if r.src.Dequeue() != nil {
	393	+	t.Fatal("Dequeue after Destroy returned a buffer")
	394	+	}
	395	+	}
	396	+
	397	+	func TestVideoSourceStateChanged(t *testing.T) {
	398	+	r := newRig(t)
	399	+	r.api.EXPECT().PWStreamGetNodeId(r.stream).Return(77).Twice()
	400	+	r.api.EXPECT().PWStreamIsDriving(r.stream).Return(false).Once()
	401	+	r.handler.EXPECT().StateChanged(int32(2), uint32(77), "").Once()
	402	+	r.src.onStateChanged(2, nil)
	403	+	r.handler.EXPECT().StateChanged(int32(3), uint32(77), "").Once()
	404	+	r.src.onStateChanged(3, nil)
	405	+	if r.src.NodeID() != 77 \|\| r.src.State() != 3 {
	406	+	t.Fatalf("node %d state %d", r.src.NodeID(), r.src.State())
	407	+	}
	408	+
	409	+	msg := append([]byte("boom"), 0)
	410	+	r.handler.EXPECT().StateChanged(int32(-1), uint32(77), "boom").Once()
	411	+	r.src.onStateChanged(-1, unsafe.Pointer(&msg[0]))
	412	+
	413	+	r.expectDestroy(0)
	414	+	r.src.Destroy()
	415	+	}
	416	+
	417	+	func TestVideoSourceResize(t *testing.T) {
	418	+	r := newRig(t)
	419	+	var p unsafe.Pointer
	420	+	r.api.EXPECT().PWStreamUpdateParams(r.stream, mock.Anything, uint32(1)).Run(func(_, params unsafe.Pointer, _ uint32) { p = params }).Return(0).Once()
	421	+	if err := r.src.Resize(8, 6); err != nil {
	422	+	t.Fatal(err)
	423	+	}
	424	+	if p != unsafe.Pointer(&r.src.params[0]) {
	425	+	t.Fatal("params not slot-owned")
	426	+	}
	427	+	if err := r.src.Resize(0, 6); err == nil {
	428	+	t.Fatal("Resize(0, 6) accepted")
	429	+	}
	430	+	r.api.EXPECT().PWStreamUpdateParams(r.stream, mock.Anything, uint32(1)).Return(-22).Once()
	431	+	if err := r.src.Resize(8, 6); err == nil {
	432	+	t.Fatal("Resize accepted a failing update_params")
	433	+	}
	434	+	r.expectDestroy(0)
	435	+	r.src.Destroy()
	436	+	if err := r.src.Resize(8, 6); err == nil {
	437	+	t.Fatal("Resize after Destroy accepted")
	438	+	}
	439	+	}
	440	+
	441	+	func TestNewVideoSourceErrors(t *testing.T) {
	442	+	api := portmocks.NewMockVideoStreamAPI(t)
	443	+	shm := portmocks.NewMockSharedMemAPI(t)
	444	+	h := portinmocks.NewMockVideoHandler(t)
	445	+	core := &Core{ptr: unsafe.Pointer(new(byte)), capi: api, shm: shm}
	446	+
	447	+	if _, err := core.NewVideoSource(VideoConfig{Width: 0, Height: 10}, h); !errors.Is(err, ErrVideoConfig) {
	448	+	t.Errorf("zero width: %v", err)
	449	+	}
	450	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, nil); !errors.Is(err, ErrVideoConfig) {
	451	+	t.Errorf("nil handler: %v", err)
	452	+	}
	453	+	if _, err := core.NewVideoSource(VideoConfig{Name: `a"b`, Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoConfig) {
	454	+	t.Errorf("quote in name: %v", err)
	455	+	}
	456	+	if _, err := (&Core{}).NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
	457	+	t.Errorf("disconnected core: %v", err)
	458	+	}
	459	+
	460	+	// properties allocation failure
	461	+	api.EXPECT().PWPropertiesNewString(mock.Anything).Return(nil).Once()
	462	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
	463	+	t.Errorf("props failure: %v", err)
	464	+	}
	465	+
	466	+	// pw_stream_new failure
	467	+	props := unsafe.Pointer(new(byte))
	468	+	api.EXPECT().PWPropertiesNewString(mock.Anything).Return(props).Once()
	469	+	api.EXPECT().PWStreamNew(core.ptr, mock.Anything, props).Return(nil).Once()
	470	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
	471	+	t.Errorf("stream_new failure: %v", err)
	472	+	}
	473	+
	474	+	// pw_stream_connect failure tears the stream down and frees the slot
	475	+	stream := unsafe.Pointer(new(byte))
	476	+	api.EXPECT().PWPropertiesNewString(mock.Anything).Return(props).Once()
	477	+	api.EXPECT().PWStreamNew(core.ptr, mock.Anything, props).Return(stream).Once()
	478	+	api.EXPECT().PWStreamAddListener(stream, mock.Anything, mock.Anything, mock.Anything).Return(0).Once()
	479	+	api.EXPECT().PWStreamConnect(stream, int32(1), uint32(spa.IDInvalid), mock.Anything, mock.Anything, uint32(1)).Return(-1).Once()
	480	+	api.EXPECT().PWStreamDisconnect(stream).Return(0).Once()
	481	+	api.EXPECT().PWStreamDestroy(stream).Once()
	482	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoConnect) {
	483	+	t.Errorf("connect failure: %v", err)
	484	+	}
	485	+	for i := range videoTable {
	486	+	if videoTable[i].Load() != nil {
	487	+	t.Errorf("callback slot %d leaked", i)
	488	+	}
	489	+	}
	490	+	}
	491	+
	492	+	func TestNewVideoSourceTableFull(t *testing.T) {
	493	+	api := portmocks.NewMockVideoStreamAPI(t)
	494	+	h := portinmocks.NewMockVideoHandler(t)
	495	+	core := &Core{ptr: unsafe.Pointer(new(byte)), capi: api, shm: portmocks.NewMockSharedMemAPI(t)}
	496	+	dummy := &VideoSource{}
	497	+	for i := range videoTable {
	498	+	videoTable[i].Store(dummy)
	499	+	}
	500	+	defer func() {
	501	+	for i := range videoTable {
	502	+	videoTable[i].Store(nil)
	503	+	}
	504	+	}()
	505	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoTableFull) {
	506	+	t.Fatalf("err = %v, want ErrVideoTableFull", err)
	507	+	}
	508	+	}
	509	+
	510	+	func TestLoopContextCore(t *testing.T) {
	511	+	api := portmocks.NewMockCAPI(t)
	512	+	r := NewRuntime(api)
	513	+
	514	+	loopPtr := unsafe.Pointer(new(byte))
	515	+	api.EXPECT().PWLoopNew(unsafe.Pointer(nil)).Return(nil).Once()
	516	+	if _, err := r.NewLoop(); !errors.Is(err, ErrLoopCreate) {
	517	+	t.Fatalf("NewLoop nil: %v", err)
	518	+	}
	519	+	api.EXPECT().PWLoopNew(unsafe.Pointer(nil)).Return(loopPtr).Once()
	520	+	loop, err := r.NewLoop()
	521	+	if err != nil {
	522	+	t.Fatal(err)
	523	+	}
	524	+
	525	+	api.EXPECT().PWLoopGetFd(loopPtr).Return(9).Once()
	526	+	if loop.Fd() != 9 {
	527	+	t.Fatal("Fd")
	528	+	}
	529	+	// Iterate: enter, iterate(0), leave in that order; a negative result is not fatal.
	530	+	var order []string
	531	+	api.EXPECT().PWLoopEnter(loopPtr).Run(func(unsafe.Pointer) { order = append(order, "enter") }).Once()
	532	+	api.EXPECT().PWLoopIterate(loopPtr, int32(0)).Run(func(unsafe.Pointer, int32) { order = append(order, "iterate") }).Return(-4).Once()
	533	+	api.EXPECT().PWLoopLeave(loopPtr).Run(func(unsafe.Pointer) { order = append(order, "leave") }).Once()
	534	+	loop.Iterate()
	535	+	if len(order) != 3 \|\| order[0] != "enter" \|\| order[1] != "iterate" \|\| order[2] != "leave" {
	536	+	t.Fatalf("Iterate order = %v", order)
	537	+	}
	538	+
	539	+	if _, err := r.NewContext(nil); !errors.Is(err, ErrContextCreate) {
	540	+	t.Fatalf("NewContext(nil loop): %v", err)
	541	+	}
	542	+	api.EXPECT().PWContextNew(loopPtr, unsafe.Pointer(nil), uintptr(0)).Return(nil).Once()
	543	+	if _, err := r.NewContext(loop); !errors.Is(err, ErrContextCreate) {
	544	+	t.Fatalf("NewContext nil ctx: %v", err)
	545	+	}
	546	+	ctxPtr := unsafe.Pointer(new(byte))
	547	+	api.EXPECT().PWContextNew(loopPtr, unsafe.Pointer(nil), uintptr(0)).Return(ctxPtr).Once()
	548	+	ctx, err := r.NewContext(loop)
	549	+	if err != nil {
	550	+	t.Fatal(err)
	551	+	}
	552	+
	553	+	api.EXPECT().PWContextConnect(ctxPtr, unsafe.Pointer(nil), uintptr(0)).Return(nil).Once()
	554	+	if _, err := ctx.Connect(); !errors.Is(err, ErrCoreConnect) {
	555	+	t.Fatalf("Connect nil: %v", err)
	556	+	}
	557	+	corePtr := unsafe.Pointer(new(byte))
	558	+	api.EXPECT().PWContextConnect(ctxPtr, unsafe.Pointer(nil), uintptr(0)).Return(corePtr).Once()
	559	+	c, err := ctx.Connect()
	560	+	if err != nil {
	561	+	t.Fatal(err)
	562	+	}
	563	+	api.EXPECT().PWCoreDisconnect(corePtr).Return(0).Once()
	564	+	c.Disconnect()
	565	+	c.Disconnect() // idempotent
	566	+	api.EXPECT().PWContextDestroy(ctxPtr).Once()
	567	+	ctx.Destroy()
	568	+	ctx.Destroy()
	569	+	if _, err := ctx.Connect(); err == nil {
	570	+	t.Fatal("Connect after Destroy accepted")
	571	+	}
	572	+	api.EXPECT().PWLoopDestroy(loopPtr).Once()
	573	+	loop.Destroy()
	574	+	loop.Destroy()
	575	+	}

@@ -0,0 +1,575 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"time"
	7	+	"unsafe"
	8	+
	9	+	"github.com/stretchr/testify/mock"
	10	+
	11	+	portinmocks "github.com/bnema/purego-pipewire/internal/ports/in/mocks"
	12	+	portmocks "github.com/bnema/purego-pipewire/internal/ports/out/mocks"
	13	+	"github.com/bnema/purego-pipewire/spa"
	14	+	)
	15	+
	16	+	func TestVideoStructLayouts(t *testing.T) {
	17	+	var pb videoPWBuffer
	18	+	if unsafe.Offsetof(pb.Buffer) != 0 \|\| unsafe.Offsetof(pb.UserData) != 8 \|\| unsafe.Offsetof(pb.Size) != 16 {
	19	+	t.Errorf("pw_buffer layout: user_data@%d size@%d", unsafe.Offsetof(pb.UserData), unsafe.Offsetof(pb.Size))
	20	+	}
	21	+	var sb spaBuffer
	22	+	if unsafe.Offsetof(sb.NMetas) != 0 \|\| unsafe.Offsetof(sb.NDatas) != 4 \|\| unsafe.Offsetof(sb.Metas) != 8 \|\| unsafe.Offsetof(sb.Datas) != 16 {
	23	+	t.Errorf("spa_buffer layout wrong")
	24	+	}
	25	+	var m spaMeta
	26	+	if unsafe.Sizeof(m) != 16 \|\| unsafe.Offsetof(m.Data) != 8 {
	27	+	t.Errorf("spa_meta layout: size %d", unsafe.Sizeof(m))
	28	+	}
	29	+	var h spaMetaHeader
	30	+	if unsafe.Sizeof(h) != 32 \|\| unsafe.Offsetof(h.PTS) != 8 \|\| unsafe.Offsetof(h.DTSOffset) != 16 \|\| unsafe.Offsetof(h.Seq) != 24 {
	31	+	t.Errorf("spa_meta_header layout: size %d", unsafe.Sizeof(h))
	32	+	}
	33	+	var c spaChunk
	34	+	if unsafe.Sizeof(c) != 16 \|\| unsafe.Offsetof(c.Stride) != 8 \|\| unsafe.Offsetof(c.Flags) != 12 {
	35	+	t.Errorf("spa_chunk layout wrong")
	36	+	}
	37	+	var d spaData
	38	+	if unsafe.Sizeof(d) != 40 \|\| unsafe.Offsetof(d.Fd) != 8 \|\| unsafe.Offsetof(d.MapOffset) != 16 \|\|
	39	+	unsafe.Offsetof(d.Maxsize) != 20 \|\| unsafe.Offsetof(d.Data) != 24 \|\| unsafe.Offsetof(d.Chunk) != 32 {
	40	+	t.Errorf("spa_data layout: size %d", unsafe.Sizeof(d))
	41	+	}
	42	+	var e videoEvents
	43	+	for name, want := range map[string][2]uintptr{
	44	+	"state_changed": {unsafe.Offsetof(e.StateChanged), 16},
	45	+	"param_changed": {unsafe.Offsetof(e.ParamChanged), 40},
	46	+	"add_buffer": {unsafe.Offsetof(e.AddBuffer), 48},
	47	+	"remove_buffer": {unsafe.Offsetof(e.RemoveBuf), 56},
	48	+	"process": {unsafe.Offsetof(e.Process), 64},
	49	+	"trigger_done": {unsafe.Offsetof(e.TriggerDone), 88},
	50	+	} {
	51	+	if want[0] != want[1] {
	52	+	t.Errorf("pw_stream_events.%s at %d, want %d", name, want[0], want[1])
	53	+	}
	54	+	}
	55	+	var s VideoSource
	56	+	if unsafe.Sizeof(s.hook) < 48 {
	57	+	t.Errorf("spa_hook storage %d bytes, need 48", unsafe.Sizeof(s.hook))
	58	+	}
	59	+	}
	60	+
	61	+	// videoRig is a VideoSource wired to mocked C calls and mocked shared memory,
	62	+	// with a fake negotiated format. C-side structures are Go memory, which is
	63	+	// enough because the production code only reads them through pointers.
	64	+	type videoRig struct {
	65	+	t *testing.T
	66	+	api *portmocks.MockVideoStreamAPI
	67	+	shm *portmocks.MockSharedMemAPI
	68	+	handler *portinmocks.MockVideoHandler
	69	+	src *VideoSource
	70	+	stream unsafe.Pointer
	71	+	streamB [8]byte
	72	+	}
	73	+
	74	+	const rigW, rigH = 4, 2 // stride 16, size 32
	75	+
	76	+	func newRig(t testing.T) videoRig {
	77	+	t.Helper()
	78	+	r := &videoRig{
	79	+	t: t,
	80	+	api: portmocks.NewMockVideoStreamAPI(t),
	81	+	shm: portmocks.NewMockSharedMemAPI(t),
	82	+	handler: portinmocks.NewMockVideoHandler(t),
	83	+	}
	84	+	r.stream = unsafe.Pointer(&r.streamB[0])
	85	+	core := &Core{ptr: unsafe.Pointer(&r.streamB[4]), capi: r.api, shm: r.shm}
	86	+
	87	+	var gotProps *byte
	88	+	r.api.EXPECT().PWPropertiesNewString(mock.Anything).Run(func(args *byte) { gotProps = args }).Return(unsafe.Pointer(&r.streamB[1])).Once()
	89	+	r.api.EXPECT().PWStreamNew(core.ptr, mock.Anything, unsafe.Pointer(&r.streamB[1])).Return(r.stream).Once()
	90	+	var events *videoEvents
	91	+	r.api.EXPECT().PWStreamAddListener(r.stream, mock.Anything, mock.Anything, mock.Anything).
	92	+	Run(func(_, _, ev, _ unsafe.Pointer) { events = (*videoEvents)(ev) }).Return(0).Once()
	93	+	var connectFlags uint32
	94	+	r.api.EXPECT().PWStreamConnect(r.stream, int32(1), uint32(spa.IDInvalid), mock.Anything, mock.Anything, uint32(1)).
	95	+	Run(func(_ unsafe.Pointer, _ int32, _ uint32, flags uint32, _ unsafe.Pointer, _ uint32) {
	96	+	connectFlags = flags
	97	+	}).
	98	+	Return(0).Once()
	99	+
	100	+	src, err := core.NewVideoSource(VideoConfig{Name: "test-src", Width: rigW, Height: rigH}, r.handler)
	101	+	if err != nil {
	102	+	t.Fatal(err)
	103	+	}
	104	+	r.src = src
	105	+	t.Cleanup(func() {
	106	+	if r.src.table >= 0 {
	107	+	videoTable[r.src.table].Store(nil)
	108	+	}
	109	+	})
	110	+
	111	+	if connectFlags != pwStreamFlagDriver\|pwStreamFlagAllocBuffers {
	112	+	t.Fatalf("connect flags = %#x, want DRIVER\|ALLOC_BUFFERS", connectFlags)
	113	+	}
	114	+	if events == nil \|\| events.Version != 2 \|\| events.StateChanged == 0 \|\| events.ParamChanged == 0 \|\|
	115	+	events.AddBuffer == 0 \|\| events.RemoveBuf == 0 \|\| events.Process == 0 \|\| events.TriggerDone == 0 {
	116	+	t.Fatalf("events not fully registered: %+v", events)
	117	+	}
	118	+	if got := cString(unsafe.Pointer(gotProps)); got != `media.class=Video/Source node.name="test-src"` {
	119	+	t.Fatalf("props = %q", got)
	120	+	}
	121	+	if videoTable[src.table].Load() != src {
	122	+	t.Fatal("source not in callback slot table")
	123	+	}
	124	+	return r
	125	+	}
	126	+
	127	+	// negotiate feeds a Format param to the source, as PipeWire would.
	128	+	func (r *videoRig) negotiate() {
	129	+	r.t.Helper()
	130	+	var dst [256]byte
	131	+	b := spa.NewPodBuilder(dst[:])
	132	+	b.Object(spa.TypeObjectFormat, spa.ParamFormat)
	133	+	b.Prop(spa.FormatKeyMediaType)
	134	+	b.Id(spa.MediaTypeVideo)
	135	+	b.Prop(spa.FormatKeyMediaSubtype)
	136	+	b.Id(spa.MediaSubtypeRaw)
	137	+	b.Prop(spa.FormatKeyVideoFormat)
	138	+	b.Id(uint32(spa.VideoFormatBGRx))
	139	+	b.Prop(spa.FormatKeyVideoSize)
	140	+	b.Rectangle(rigW, rigH)
	141	+	b.End()
	142	+	pod, err := b.Bytes()
	143	+	if err != nil {
	144	+	r.t.Fatal(err)
	145	+	}
	146	+	var params unsafe.Pointer
	147	+	r.api.EXPECT().PWStreamUpdateParams(r.stream, mock.Anything, uint32(2)).
	148	+	Run(func(_, p unsafe.Pointer, _ uint32) { params = p }).Return(0).Once()
	149	+	r.handler.EXPECT().FormatChanged(spa.VideoFormatBGRx, rigW, rigH, rigW*4).Once()
	150	+	r.src.onParamChanged(spa.ParamFormat, unsafe.Pointer(&pod[0]))
	151	+	if params != unsafe.Pointer(&r.src.params[0]) {
	152	+	r.t.Fatal("update_params not given the slot-owned params array")
	153	+	}
	154	+	}
	155	+
	156	+	// fakeBuf is a PipeWire-side buffer: pw_buffer + spa_buffer + data + chunk + header meta.
	157	+	type fakeBuf struct {
	158	+	pw videoPWBuffer
	159	+	sb spaBuffer
	160	+	data spaData
	161	+	chunk spaChunk
	162	+	meta spaMeta
	163	+	hdr spaMetaHeader
	164	+	}
	165	+
	166	+	func newFakeBuf(dataType uint32) *fakeBuf {
	167	+	f := &fakeBuf{}
	168	+	f.data.Type = dataType
	169	+	f.data.Chunk = &f.chunk
	170	+	f.meta = spaMeta{Type: spa.MetaTypeHeader, Size: spa.MetaHeaderSize, Data: unsafe.Pointer(&f.hdr)}
	171	+	f.sb = spaBuffer{NMetas: 1, NDatas: 1, Metas: unsafe.Pointer(&f.meta), Datas: unsafe.Pointer(&f.data)}
	172	+	f.pw.Buffer = &f.sb
	173	+	return f
	174	+	}
	175	+
	176	+	func (f *fakeBuf) ptr() unsafe.Pointer { return unsafe.Pointer(&f.pw) }
	177	+
	178	+	// add runs add_buffer with a memfd that the shm mock "creates".
	179	+	func (r videoRig) add(f fakeBuf, fd int) []byte {
	180	+	r.t.Helper()
	181	+	mem := make([]byte, rigW4rigH)
	182	+	r.shm.EXPECT().NewSharedMem("pw-video", rigW4rigH).Return(fd, mem, nil).Once()
	183	+	r.handler.EXPECT().BuffersChanged().Once()
	184	+	r.src.onAddBuffer(f.ptr())
	185	+	return mem
	186	+	}
	187	+
	188	+	func TestVideoSourceNegotiationAndAddBuffer(t *testing.T) {
	189	+	r := newRig(t)
	190	+	r.negotiate()
	191	+	f := newFakeBuf(1 << spa.DataMemFd)
	192	+	mem := r.add(f, 42)
	193	+
	194	+	if f.data.Type != spa.DataMemFd \|\| f.data.Flags != spa.DataFlagReadWrite \|\| f.data.Fd != 42 \|\|
	195	+	f.data.MapOffset != 0 \|\| f.data.Maxsize != uint32(len(mem)) \|\| f.data.Data != unsafe.Pointer(&mem[0]) {
	196	+	t.Fatalf("spa_data not filled: %+v", f.data)
	197	+	}
	198	+	if f.chunk.Offset != 0 \|\| f.chunk.Size != uint32(len(mem)) \|\| f.chunk.Stride != rigW*4 {
	199	+	t.Fatalf("chunk = %+v", f.chunk)
	200	+	}
	201	+	if f.pw.UserData != 1 {
	202	+	t.Fatalf("user_data = %d, want slot 0 + 1", f.pw.UserData)
	203	+	}
	204	+
	205	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	206	+	b := r.src.Dequeue()
	207	+	if b == nil \|\| len(b.Data) != len(mem) \|\| &b.Data[0] != &mem[0] \|\| b.Stride != rigW*4 \|\| b.Width != rigW \|\| b.Height != rigH \|\| b.Format != spa.VideoFormatBGRx {
	208	+	t.Fatalf("Dequeue = %+v", b)
	209	+	}
	210	+
	211	+	var queued unsafe.Pointer
	212	+	r.api.EXPECT().PWStreamQueueBuffer(r.stream, f.ptr()).Run(func(_, p unsafe.Pointer) { queued = p }).Return(0).Once()
	213	+	r.src.Queue(b, 7, 1500*time.Millisecond, false)
	214	+	if queued != f.ptr() {
	215	+	t.Fatal("buffer not queued")
	216	+	}
	217	+	if f.chunk.Size != uint32(len(mem)) \|\| f.chunk.Stride != rigW*4 \|\| f.chunk.Flags != 0 {
	218	+	t.Fatalf("chunk after Queue = %+v", f.chunk)
	219	+	}
	220	+	if f.hdr.Seq != 7 \|\| f.hdr.PTS != int64(1500*time.Millisecond) \|\| f.hdr.Flags != 0 {
	221	+	t.Fatalf("header after Queue = %+v", f.hdr)
	222	+	}
	223	+
	224	+	// Second round, corrupted.
	225	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	226	+	b = r.src.Dequeue()
	227	+	if b == nil {
	228	+	t.Fatal("buffer not re-dequeued")
	229	+	}
	230	+	r.api.EXPECT().PWStreamQueueBuffer(r.stream, f.ptr()).Return(0).Once()
	231	+	r.src.Queue(b, 8, 0, true)
	232	+	if f.chunk.Size != 0 \|\| f.chunk.Flags != spa.ChunkFlagCorrupted \|\| f.hdr.Flags != spa.MetaHeaderFlagCorrupted \|\| f.hdr.Seq != 8 {
	233	+	t.Fatalf("corrupted frame: chunk %+v hdr %+v", f.chunk, f.hdr)
	234	+	}
	235	+
	236	+	// Trigger is separate and goes straight to the C call.
	237	+	r.api.EXPECT().PWStreamTriggerProcess(r.stream).Return(0).Once()
	238	+	r.src.Trigger()
	239	+
	240	+	r.expectDestroy(1)
	241	+	r.src.Destroy()
	242	+	}
	243	+
	244	+	// expectDestroy expects the disconnect/destroy sequence and the free of n live slots.
	245	+	func (r *videoRig) expectDestroy(n int) {
	246	+	r.api.EXPECT().PWStreamDisconnect(r.stream).Return(0).Once()
	247	+	r.api.EXPECT().PWStreamDestroy(r.stream).Once()
	248	+	if n > 0 {
	249	+	r.shm.EXPECT().FreeSharedMem(mock.Anything, mock.Anything).Times(n)
	250	+	}
	251	+	}
	252	+
	253	+	func TestVideoSourceMemfdFailureMakesBufferUnusable(t *testing.T) {
	254	+	r := newRig(t)
	255	+	r.negotiate()
	256	+
	257	+	bad := newFakeBuf(1 << spa.DataMemFd)
	258	+	r.shm.EXPECT().NewSharedMem("pw-video", rigW4rigH).Return(-1, nil, errors.New("memfd_create: boom")).Once()
	259	+	r.src.onAddBuffer(bad.ptr())
	260	+	if bad.pw.UserData != 0 {
	261	+	t.Fatalf("failed buffer got user_data %d", bad.pw.UserData)
	262	+	}
	263	+
	264	+	good := newFakeBuf(1 << spa.DataMemFd)
	265	+	mem := r.add(good, 7)
	266	+
	267	+	// Dequeue skips the unusable buffer and returns the good one.
	268	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(bad.ptr()).Once()
	269	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(good.ptr()).Once()
	270	+	b := r.src.Dequeue()
	271	+	if b == nil \|\| &b.Data[0] != &mem[0] {
	272	+	t.Fatalf("Dequeue = %+v, want the good buffer", b)
	273	+	}
	274	+	r.api.EXPECT().PWStreamReturnBuffer(r.stream, good.ptr()).Return(0).Once()
	275	+	r.src.Release(b)
	276	+
	277	+	// A DMA-BUF-only buffer is also unusable and allocates nothing.
	278	+	dma := newFakeBuf(1 << spa.DataDmaBuf)
	279	+	r.src.onAddBuffer(dma.ptr())
	280	+	if dma.pw.UserData != 0 {
	281	+	t.Fatalf("dmabuf buffer got user_data %d", dma.pw.UserData)
	282	+	}
	283	+
	284	+	// Nothing is dequeueable and no buffer is available: nil, no panic.
	285	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(nil).Once()
	286	+	if b := r.src.Dequeue(); b != nil {
	287	+	t.Fatalf("Dequeue = %+v, want nil", b)
	288	+	}
	289	+	r.expectDestroy(1)
	290	+	r.src.Destroy()
	291	+	}
	292	+
	293	+	func TestVideoSourceRemoveBorrowedKeepsMappingUntilRelease(t *testing.T) {
	294	+	r := newRig(t)
	295	+	r.negotiate()
	296	+	f := newFakeBuf(1 << spa.DataMemFd)
	297	+	mem := r.add(f, 9)
	298	+
	299	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	300	+	b := r.src.Dequeue()
	301	+	if b == nil {
	302	+	t.Fatal("no buffer")
	303	+	}
	304	+
	305	+	// PipeWire removes the buffer while the owner still holds it: no unmap yet
	306	+	// (FreeSharedMem has no expectation, so a call would fail the test).
	307	+	r.handler.EXPECT().BuffersChanged().Once()
	308	+	r.src.onRemoveBuffer(f.ptr())
	309	+	b.Data[0] = 0xAB // still writable
	310	+	if mem[0] != 0xAB \|\| f.pw.UserData != 0 {
	311	+	t.Fatalf("mapping not kept: mem[0]=%x user_data=%d", mem[0], f.pw.UserData)
	312	+	}
	313	+
	314	+	// A removed slot is never returned by Dequeue (PipeWire no longer owns it,
	315	+	// so even if it showed up the source would not hand it out).
	316	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	317	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(nil).Once()
	318	+	if got := r.src.Dequeue(); got != nil {
	319	+	t.Fatalf("removed buffer dequeued: %+v", got)
	320	+	}
	321	+
	322	+	// Queue on a removed slot must not call pw_stream_queue_buffer, and
	323	+	// releases the mapping.
	324	+	r.shm.EXPECT().FreeSharedMem(9, mem).Once()
	325	+	r.api.AssertNotCalled(t, "PWStreamQueueBuffer", mock.Anything, mock.Anything)
	326	+	r.src.Queue(b, 1, 0, false)
	327	+	r.api.AssertNotCalled(t, "PWStreamQueueBuffer", mock.Anything, mock.Anything)
	328	+
	329	+	// The slot is reusable afterwards.
	330	+	f2 := newFakeBuf(1 << spa.DataMemFd)
	331	+	r.add(f2, 10)
	332	+	if f2.pw.UserData != 1 {
	333	+	t.Fatalf("slot not reused, user_data = %d", f2.pw.UserData)
	334	+	}
	335	+	r.expectDestroy(1)
	336	+	r.src.Destroy()
	337	+	}
	338	+
	339	+	func TestVideoSourceRemoveBorrowedThenRelease(t *testing.T) {
	340	+	r := newRig(t)
	341	+	r.negotiate()
	342	+	f := newFakeBuf(1 << spa.DataMemFd)
	343	+	mem := r.add(f, 11)
	344	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	345	+	b := r.src.Dequeue()
	346	+
	347	+	r.handler.EXPECT().BuffersChanged().Once()
	348	+	r.src.onRemoveBuffer(f.ptr())
	349	+	r.shm.EXPECT().FreeSharedMem(11, mem).Once()
	350	+	r.src.Release(b)
	351	+	r.api.AssertNotCalled(t, "PWStreamReturnBuffer", mock.Anything, mock.Anything)
	352	+
	353	+	// A second Release/Queue of the same buffer is ignored.
	354	+	r.src.Release(b)
	355	+	r.src.Queue(b, 0, 0, false)
	356	+	r.expectDestroy(0)
	357	+	r.src.Destroy()
	358	+	}
	359	+
	360	+	func TestVideoSourceRemoveIdleUnmapsImmediately(t *testing.T) {
	361	+	r := newRig(t)
	362	+	r.negotiate()
	363	+	f := newFakeBuf(1 << spa.DataMemFd)
	364	+	mem := r.add(f, 12)
	365	+	r.shm.EXPECT().FreeSharedMem(12, mem).Once()
	366	+	r.handler.EXPECT().BuffersChanged().Once()
	367	+	r.src.onRemoveBuffer(f.ptr())
	368	+	r.expectDestroy(0)
	369	+	r.src.Destroy()
	370	+	}
	371	+
	372	+	func TestVideoSourceDestroyKeepsBorrowedMappingUntilRelease(t *testing.T) {
	373	+	r := newRig(t)
	374	+	r.negotiate()
	375	+	f := newFakeBuf(1 << spa.DataMemFd)
	376	+	mem := r.add(f, 13)
	377	+	r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
	378	+	b := r.src.Dequeue()
	379	+
	380	+	r.expectDestroy(0)
	381	+	r.src.Destroy()
	382	+	b.Data[1] = 1 // still writable after Destroy
	383	+	if mem[1] != 1 {
	384	+	t.Fatal("mapping lost on Destroy")
	385	+	}
	386	+	r.shm.EXPECT().FreeSharedMem(13, mem).Once()
	387	+	r.src.Release(b)
	388	+
	389	+	// Destroy is idempotent and Dequeue/Trigger after it are no-ops.
	390	+	r.src.Destroy()
	391	+	r.src.Trigger()
	392	+	if r.src.Dequeue() != nil {
	393	+	t.Fatal("Dequeue after Destroy returned a buffer")
	394	+	}
	395	+	}
	396	+
	397	+	func TestVideoSourceStateChanged(t *testing.T) {
	398	+	r := newRig(t)
	399	+	r.api.EXPECT().PWStreamGetNodeId(r.stream).Return(77).Twice()
	400	+	r.api.EXPECT().PWStreamIsDriving(r.stream).Return(false).Once()
	401	+	r.handler.EXPECT().StateChanged(int32(2), uint32(77), "").Once()
	402	+	r.src.onStateChanged(2, nil)
	403	+	r.handler.EXPECT().StateChanged(int32(3), uint32(77), "").Once()
	404	+	r.src.onStateChanged(3, nil)
	405	+	if r.src.NodeID() != 77 \|\| r.src.State() != 3 {
	406	+	t.Fatalf("node %d state %d", r.src.NodeID(), r.src.State())
	407	+	}
	408	+
	409	+	msg := append([]byte("boom"), 0)
	410	+	r.handler.EXPECT().StateChanged(int32(-1), uint32(77), "boom").Once()
	411	+	r.src.onStateChanged(-1, unsafe.Pointer(&msg[0]))
	412	+
	413	+	r.expectDestroy(0)
	414	+	r.src.Destroy()
	415	+	}
	416	+
	417	+	func TestVideoSourceResize(t *testing.T) {
	418	+	r := newRig(t)
	419	+	var p unsafe.Pointer
	420	+	r.api.EXPECT().PWStreamUpdateParams(r.stream, mock.Anything, uint32(1)).Run(func(_, params unsafe.Pointer, _ uint32) { p = params }).Return(0).Once()
	421	+	if err := r.src.Resize(8, 6); err != nil {
	422	+	t.Fatal(err)
	423	+	}
	424	+	if p != unsafe.Pointer(&r.src.params[0]) {
	425	+	t.Fatal("params not slot-owned")
	426	+	}
	427	+	if err := r.src.Resize(0, 6); err == nil {
	428	+	t.Fatal("Resize(0, 6) accepted")
	429	+	}
	430	+	r.api.EXPECT().PWStreamUpdateParams(r.stream, mock.Anything, uint32(1)).Return(-22).Once()
	431	+	if err := r.src.Resize(8, 6); err == nil {
	432	+	t.Fatal("Resize accepted a failing update_params")
	433	+	}
	434	+	r.expectDestroy(0)
	435	+	r.src.Destroy()
	436	+	if err := r.src.Resize(8, 6); err == nil {
	437	+	t.Fatal("Resize after Destroy accepted")
	438	+	}
	439	+	}
	440	+
	441	+	func TestNewVideoSourceErrors(t *testing.T) {
	442	+	api := portmocks.NewMockVideoStreamAPI(t)
	443	+	shm := portmocks.NewMockSharedMemAPI(t)
	444	+	h := portinmocks.NewMockVideoHandler(t)
	445	+	core := &Core{ptr: unsafe.Pointer(new(byte)), capi: api, shm: shm}
	446	+
	447	+	if _, err := core.NewVideoSource(VideoConfig{Width: 0, Height: 10}, h); !errors.Is(err, ErrVideoConfig) {
	448	+	t.Errorf("zero width: %v", err)
	449	+	}
	450	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, nil); !errors.Is(err, ErrVideoConfig) {
	451	+	t.Errorf("nil handler: %v", err)
	452	+	}
	453	+	if _, err := core.NewVideoSource(VideoConfig{Name: `a"b`, Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoConfig) {
	454	+	t.Errorf("quote in name: %v", err)
	455	+	}
	456	+	if _, err := (&Core{}).NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
	457	+	t.Errorf("disconnected core: %v", err)
	458	+	}
	459	+
	460	+	// properties allocation failure
	461	+	api.EXPECT().PWPropertiesNewString(mock.Anything).Return(nil).Once()
	462	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
	463	+	t.Errorf("props failure: %v", err)
	464	+	}
	465	+
	466	+	// pw_stream_new failure
	467	+	props := unsafe.Pointer(new(byte))
	468	+	api.EXPECT().PWPropertiesNewString(mock.Anything).Return(props).Once()
	469	+	api.EXPECT().PWStreamNew(core.ptr, mock.Anything, props).Return(nil).Once()
	470	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
	471	+	t.Errorf("stream_new failure: %v", err)
	472	+	}
	473	+
	474	+	// pw_stream_connect failure tears the stream down and frees the slot
	475	+	stream := unsafe.Pointer(new(byte))
	476	+	api.EXPECT().PWPropertiesNewString(mock.Anything).Return(props).Once()
	477	+	api.EXPECT().PWStreamNew(core.ptr, mock.Anything, props).Return(stream).Once()
	478	+	api.EXPECT().PWStreamAddListener(stream, mock.Anything, mock.Anything, mock.Anything).Return(0).Once()
	479	+	api.EXPECT().PWStreamConnect(stream, int32(1), uint32(spa.IDInvalid), mock.Anything, mock.Anything, uint32(1)).Return(-1).Once()
	480	+	api.EXPECT().PWStreamDisconnect(stream).Return(0).Once()
	481	+	api.EXPECT().PWStreamDestroy(stream).Once()
	482	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoConnect) {
	483	+	t.Errorf("connect failure: %v", err)
	484	+	}
	485	+	for i := range videoTable {
	486	+	if videoTable[i].Load() != nil {
	487	+	t.Errorf("callback slot %d leaked", i)
	488	+	}
	489	+	}
	490	+	}
	491	+
	492	+	func TestNewVideoSourceTableFull(t *testing.T) {
	493	+	api := portmocks.NewMockVideoStreamAPI(t)
	494	+	h := portinmocks.NewMockVideoHandler(t)
	495	+	core := &Core{ptr: unsafe.Pointer(new(byte)), capi: api, shm: portmocks.NewMockSharedMemAPI(t)}
	496	+	dummy := &VideoSource{}
	497	+	for i := range videoTable {
	498	+	videoTable[i].Store(dummy)
	499	+	}
	500	+	defer func() {
	501	+	for i := range videoTable {
	502	+	videoTable[i].Store(nil)
	503	+	}
	504	+	}()
	505	+	if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoTableFull) {
	506	+	t.Fatalf("err = %v, want ErrVideoTableFull", err)
	507	+	}
	508	+	}
	509	+
	510	+	func TestLoopContextCore(t *testing.T) {
	511	+	api := portmocks.NewMockCAPI(t)
	512	+	r := NewRuntime(api)
	513	+
	514	+	loopPtr := unsafe.Pointer(new(byte))
	515	+	api.EXPECT().PWLoopNew(unsafe.Pointer(nil)).Return(nil).Once()
	516	+	if _, err := r.NewLoop(); !errors.Is(err, ErrLoopCreate) {
	517	+	t.Fatalf("NewLoop nil: %v", err)
	518	+	}
	519	+	api.EXPECT().PWLoopNew(unsafe.Pointer(nil)).Return(loopPtr).Once()
	520	+	loop, err := r.NewLoop()
	521	+	if err != nil {
	522	+	t.Fatal(err)
	523	+	}
	524	+
	525	+	api.EXPECT().PWLoopGetFd(loopPtr).Return(9).Once()
	526	+	if loop.Fd() != 9 {
	527	+	t.Fatal("Fd")
	528	+	}
	529	+	// Iterate: enter, iterate(0), leave in that order; a negative result is not fatal.
	530	+	var order []string
	531	+	api.EXPECT().PWLoopEnter(loopPtr).Run(func(unsafe.Pointer) { order = append(order, "enter") }).Once()
	532	+	api.EXPECT().PWLoopIterate(loopPtr, int32(0)).Run(func(unsafe.Pointer, int32) { order = append(order, "iterate") }).Return(-4).Once()
	533	+	api.EXPECT().PWLoopLeave(loopPtr).Run(func(unsafe.Pointer) { order = append(order, "leave") }).Once()
	534	+	loop.Iterate()
	535	+	if len(order) != 3 \|\| order[0] != "enter" \|\| order[1] != "iterate" \|\| order[2] != "leave" {
	536	+	t.Fatalf("Iterate order = %v", order)
	537	+	}
	538	+
	539	+	if _, err := r.NewContext(nil); !errors.Is(err, ErrContextCreate) {
	540	+	t.Fatalf("NewContext(nil loop): %v", err)
	541	+	}
	542	+	api.EXPECT().PWContextNew(loopPtr, unsafe.Pointer(nil), uintptr(0)).Return(nil).Once()
	543	+	if _, err := r.NewContext(loop); !errors.Is(err, ErrContextCreate) {
	544	+	t.Fatalf("NewContext nil ctx: %v", err)
	545	+	}
	546	+	ctxPtr := unsafe.Pointer(new(byte))
	547	+	api.EXPECT().PWContextNew(loopPtr, unsafe.Pointer(nil), uintptr(0)).Return(ctxPtr).Once()
	548	+	ctx, err := r.NewContext(loop)
	549	+	if err != nil {
	550	+	t.Fatal(err)
	551	+	}
	552	+
	553	+	api.EXPECT().PWContextConnect(ctxPtr, unsafe.Pointer(nil), uintptr(0)).Return(nil).Once()
	554	+	if _, err := ctx.Connect(); !errors.Is(err, ErrCoreConnect) {
	555	+	t.Fatalf("Connect nil: %v", err)
	556	+	}
	557	+	corePtr := unsafe.Pointer(new(byte))
	558	+	api.EXPECT().PWContextConnect(ctxPtr, unsafe.Pointer(nil), uintptr(0)).Return(corePtr).Once()
	559	+	c, err := ctx.Connect()
	560	+	if err != nil {
	561	+	t.Fatal(err)
	562	+	}
	563	+	api.EXPECT().PWCoreDisconnect(corePtr).Return(0).Once()
	564	+	c.Disconnect()
	565	+	c.Disconnect() // idempotent
	566	+	api.EXPECT().PWContextDestroy(ctxPtr).Once()
	567	+	ctx.Destroy()
	568	+	ctx.Destroy()
	569	+	if _, err := ctx.Connect(); err == nil {
	570	+	t.Fatal("Connect after Destroy accepted")
	571	+	}
	572	+	api.EXPECT().PWLoopDestroy(loopPtr).Once()
	573	+	loop.Destroy()
	574	+	loop.Destroy()
	575	+	}