1package core
2
3import (
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
16func 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.
64type 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
74const rigW, rigH = 4, 2 // stride 16, size 32
75
76func 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.
128func (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.
157type fakeBuf struct {
158 pw videoPWBuffer
159 sb spaBuffer
160 data spaData
161 chunk spaChunk
162 meta spaMeta
163 hdr spaMetaHeader
164}
165
166func 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
176func (f *fakeBuf) ptr() unsafe.Pointer { return unsafe.Pointer(&f.pw) }
177
178// add runs add_buffer with a memfd that the shm mock "creates".
179func (r *videoRig) add(f *fakeBuf, fd int) []byte {
180 r.t.Helper()
181 mem := make([]byte, rigW*4*rigH)
182 r.shm.EXPECT().NewSharedMem("pw-video", rigW*4*rigH).Return(fd, mem, nil).Once()
183 r.handler.EXPECT().BuffersChanged().Once()
184 r.src.onAddBuffer(f.ptr())
185 return mem
186}
187
188func 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.
245func (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
253func 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", rigW*4*rigH).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
293func 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
339func 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
360func 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
372func 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
397func 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
417func 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 // A borrowed buffer holds a slot the replacements need.
431 r.negotiate()
432 f := newFakeBuf(1 << spa.DataMemFd)
433 r.add(f, 42)
434 r.api.EXPECT().PWStreamDequeueBuffer(r.stream).Return(f.ptr()).Once()
435 b := r.src.Dequeue()
436 if err := r.src.Resize(8, 6); !errors.Is(err, ErrVideoBorrowed) {
437 t.Fatalf("Resize with a borrowed buffer: %v, want ErrVideoBorrowed", err)
438 }
439 r.api.EXPECT().PWStreamReturnBuffer(r.stream, f.ptr()).Return(0).Once()
440 r.src.Release(b)
441 r.api.EXPECT().PWStreamUpdateParams(r.stream, mock.Anything, uint32(1)).Return(-22).Once()
442 if err := r.src.Resize(8, 6); err == nil {
443 t.Fatal("Resize accepted a failing update_params")
444 }
445 r.expectDestroy(1)
446 r.src.Destroy()
447 if err := r.src.Resize(8, 6); err == nil {
448 t.Fatal("Resize after Destroy accepted")
449 }
450}
451
452func TestNewVideoSourceErrors(t *testing.T) {
453 api := portmocks.NewMockVideoStreamAPI(t)
454 shm := portmocks.NewMockSharedMemAPI(t)
455 h := portinmocks.NewMockVideoHandler(t)
456 core := &Core{ptr: unsafe.Pointer(new(byte)), capi: api, shm: shm}
457
458 if _, err := core.NewVideoSource(VideoConfig{Width: 0, Height: 10}, h); !errors.Is(err, ErrVideoConfig) {
459 t.Errorf("zero width: %v", err)
460 }
461 if _, err := core.NewVideoSource(VideoConfig{Width: 1 << 15, Height: 1 << 14}, h); !errors.Is(err, ErrVideoConfig) {
462 t.Errorf("frame size over int32: %v", err)
463 }
464 if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10, Formats: []spa.VideoFormat{spa.VideoFormatBGRx, 18}}, h); !errors.Is(err, ErrVideoConfig) {
465 t.Errorf("unsupported format: %v", err)
466 }
467 if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, nil); !errors.Is(err, ErrVideoConfig) {
468 t.Errorf("nil handler: %v", err)
469 }
470 if _, err := core.NewVideoSource(VideoConfig{Name: `a"b`, Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoConfig) {
471 t.Errorf("quote in name: %v", err)
472 }
473 if _, err := (&Core{}).NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
474 t.Errorf("disconnected core: %v", err)
475 }
476
477 // properties allocation failure
478 api.EXPECT().PWPropertiesNewString(mock.Anything).Return(nil).Once()
479 if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
480 t.Errorf("props failure: %v", err)
481 }
482
483 // pw_stream_new failure
484 props := unsafe.Pointer(new(byte))
485 api.EXPECT().PWPropertiesNewString(mock.Anything).Return(props).Once()
486 api.EXPECT().PWStreamNew(core.ptr, mock.Anything, props).Return(nil).Once()
487 if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoStreamCreate) {
488 t.Errorf("stream_new failure: %v", err)
489 }
490
491 // pw_stream_connect failure tears the stream down and frees the slot
492 stream := unsafe.Pointer(new(byte))
493 api.EXPECT().PWPropertiesNewString(mock.Anything).Return(props).Once()
494 api.EXPECT().PWStreamNew(core.ptr, mock.Anything, props).Return(stream).Once()
495 api.EXPECT().PWStreamAddListener(stream, mock.Anything, mock.Anything, mock.Anything).Return(0).Once()
496 api.EXPECT().PWStreamConnect(stream, int32(1), uint32(spa.IDInvalid), mock.Anything, mock.Anything, uint32(1)).Return(-1).Once()
497 api.EXPECT().PWStreamDisconnect(stream).Return(0).Once()
498 api.EXPECT().PWStreamDestroy(stream).Once()
499 if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoConnect) {
500 t.Errorf("connect failure: %v", err)
501 }
502 for i := range videoTable {
503 if videoTable[i].Load() != nil {
504 t.Errorf("callback slot %d leaked", i)
505 }
506 }
507}
508
509func TestNewVideoSourceTableFull(t *testing.T) {
510 api := portmocks.NewMockVideoStreamAPI(t)
511 h := portinmocks.NewMockVideoHandler(t)
512 core := &Core{ptr: unsafe.Pointer(new(byte)), capi: api, shm: portmocks.NewMockSharedMemAPI(t)}
513 dummy := &VideoSource{}
514 for i := range videoTable {
515 videoTable[i].Store(dummy)
516 }
517 defer func() {
518 for i := range videoTable {
519 videoTable[i].Store(nil)
520 }
521 }()
522 if _, err := core.NewVideoSource(VideoConfig{Width: 10, Height: 10}, h); !errors.Is(err, ErrVideoTableFull) {
523 t.Fatalf("err = %v, want ErrVideoTableFull", err)
524 }
525}
526
527func TestLoopContextCore(t *testing.T) {
528 api := portmocks.NewMockCAPI(t)
529 r := NewRuntime(api)
530
531 loopPtr := unsafe.Pointer(new(byte))
532 api.EXPECT().PWLoopNew(unsafe.Pointer(nil)).Return(nil).Once()
533 if _, err := r.NewLoop(); !errors.Is(err, ErrLoopCreate) {
534 t.Fatalf("NewLoop nil: %v", err)
535 }
536 api.EXPECT().PWLoopNew(unsafe.Pointer(nil)).Return(loopPtr).Once()
537 loop, err := r.NewLoop()
538 if err != nil {
539 t.Fatal(err)
540 }
541
542 api.EXPECT().PWLoopGetFd(loopPtr).Return(9).Once()
543 if loop.Fd() != 9 {
544 t.Fatal("Fd")
545 }
546 // Iterate: enter, iterate(0), leave in that order; a negative result is not fatal.
547 var order []string
548 api.EXPECT().PWLoopEnter(loopPtr).Run(func(unsafe.Pointer) { order = append(order, "enter") }).Once()
549 api.EXPECT().PWLoopIterate(loopPtr, int32(0)).Run(func(unsafe.Pointer, int32) { order = append(order, "iterate") }).Return(-4).Once()
550 api.EXPECT().PWLoopLeave(loopPtr).Run(func(unsafe.Pointer) { order = append(order, "leave") }).Once()
551 loop.Iterate()
552 if len(order) != 3 || order[0] != "enter" || order[1] != "iterate" || order[2] != "leave" {
553 t.Fatalf("Iterate order = %v", order)
554 }
555
556 if _, err := r.NewContext(nil); !errors.Is(err, ErrContextCreate) {
557 t.Fatalf("NewContext(nil loop): %v", err)
558 }
559 api.EXPECT().PWContextNew(loopPtr, unsafe.Pointer(nil), uintptr(0)).Return(nil).Once()
560 if _, err := r.NewContext(loop); !errors.Is(err, ErrContextCreate) {
561 t.Fatalf("NewContext nil ctx: %v", err)
562 }
563 ctxPtr := unsafe.Pointer(new(byte))
564 api.EXPECT().PWContextNew(loopPtr, unsafe.Pointer(nil), uintptr(0)).Return(ctxPtr).Once()
565 ctx, err := r.NewContext(loop)
566 if err != nil {
567 t.Fatal(err)
568 }
569
570 api.EXPECT().PWContextConnect(ctxPtr, unsafe.Pointer(nil), uintptr(0)).Return(nil).Once()
571 if _, err := ctx.Connect(); !errors.Is(err, ErrCoreConnect) {
572 t.Fatalf("Connect nil: %v", err)
573 }
574 corePtr := unsafe.Pointer(new(byte))
575 api.EXPECT().PWContextConnect(ctxPtr, unsafe.Pointer(nil), uintptr(0)).Return(corePtr).Once()
576 c, err := ctx.Connect()
577 if err != nil {
578 t.Fatal(err)
579 }
580 api.EXPECT().PWCoreDisconnect(corePtr).Return(0).Once()
581 c.Disconnect()
582 c.Disconnect() // idempotent
583 api.EXPECT().PWContextDestroy(ctxPtr).Once()
584 ctx.Destroy()
585 ctx.Destroy()
586 if _, err := ctx.Connect(); err == nil {
587 t.Fatal("Connect after Destroy accepted")
588 }
589 api.EXPECT().PWLoopDestroy(loopPtr).Once()
590 loop.Destroy()
591 loop.Destroy()
592}