From e5472680f62ffbb53060c7b1740501b66b1e9b78 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 25 Oct 2026 13:01:23 -0700 Subject: [PATCH 1/2] Test vision IPC bindings --- msgq/visionipc/tests/test_visionipc.py | 57 ++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/msgq/visionipc/tests/test_visionipc.py b/msgq/visionipc/tests/test_visionipc.py index b938df447..d3788d6a6 100644 --- a/msgq/visionipc/tests/test_visionipc.py +++ b/msgq/visionipc/tests/test_visionipc.py @@ -1,3 +1,4 @@ +import gc import struct import unittest from typing import Optional @@ -101,3 +102,59 @@ def test_conflate(self): recv_buf = self.client.recv(timeout_ms=5) assert recv_buf is None + + @unittest.expectedFailure # Cython copies strided input as contiguous bytes. + def test_send_buffer_types(self): + self.setup_vipc("camerad", 0) + assert self.server is not None + assert self.client is not None + assert self.client.buffer_len is not None + data = bytes(i % 256 for i in range(self.client.buffer_len)) + strided = bytearray(len(data) * 2) + strided[::2] = data + for payload in (data, bytearray(data), memoryview(data), memoryview(strided)[::2], memoryview(data[::-1])[::-1]): + with self.subTest(type=type(payload)): + self.server.send(0, payload, frame_id=42, timestamp_sof=123, timestamp_eof=456) + buf = self.client.recv() + assert buf is not None + self.assertEqual(bytes(buf.data), data) + self.assertEqual(buf.frame_id, 42) + self.assertEqual(self.client.timestamp_sof, 123) + self.assertEqual(self.client.timestamp_eof, 456) + self.assertFalse(self.client.valid) + + def test_send_wrong_size(self): + self.setup_vipc("camerad", 0) + assert self.server is not None + with self.assertRaises(AssertionError): + self.server.send(0, b"short") + + @unittest.skip("Cython buffers do not retain their client; accessing freed buffers segfaults.") + def test_buffer_keeps_client_alive(self): + self.setup_vipc("camerad", 0) + assert self.server is not None + assert self.client is not None + assert self.client.buffer_len is not None + data = b"a" * self.client.buffer_len + self.server.send(0, data, frame_id=42) + buf = self.client.recv() + assert buf is not None + self.client = None + gc.collect() + self.assertEqual(buf.frame_id, 42) + view = buf.data + del buf + gc.collect() + self.assertEqual(bytes(view), data) + self.assertFalse(view.readonly) + view[0] = ord("b") + self.assertEqual(view[0], ord("b")) + + def test_unconnected_properties(self): + client = VisionIpcClient("unconnected", 0, False) + for value in (client.width, client.height, client.stride, client.uv_offset, client.buffer_len): + self.assertIsNone(value) + self.assertEqual(client.num_buffers, 0) + self.assertFalse(client.is_connected()) + self.assertFalse(client.connect(False)) + self.assertIsNone(client.recv(timeout_ms=0)) From 46a0ddd602c19f683b612299091066905614a28b Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 25 Oct 2026 13:04:57 -0700 Subject: [PATCH 2/2] Simplify vision IPC test setup --- msgq/visionipc/tests/test_visionipc.py | 134 +++++++++++-------------- 1 file changed, 59 insertions(+), 75 deletions(-) diff --git a/msgq/visionipc/tests/test_visionipc.py b/msgq/visionipc/tests/test_visionipc.py index d3788d6a6..4b1ad7cce 100644 --- a/msgq/visionipc/tests/test_visionipc.py +++ b/msgq/visionipc/tests/test_visionipc.py @@ -17,24 +17,24 @@ def tearDown(self): self.client = None self.server = None - def setup_vipc(self, name, *stream_types, num_buffers=1, width=100, height=100, conflate=False): - self.server = VisionIpcServer(name) - for stream_type in stream_types: - self.server.create_buffers(stream_type, num_buffers, width, height) - self.server.start_listener() - - if len(stream_types): - self.client = VisionIpcClient(name, stream_types[0], conflate) - assert self.client.connect(True) - else: - self.client = None - - return self.server, self.client + def setup_vipc( + self, name: str, stream_type: int, *other_stream_types: int, + num_buffers: int = 1, width: int = 100, height: int = 100, conflate: bool = False, + ) -> tuple[VisionIpcServer, VisionIpcClient, int]: + self.server = server = VisionIpcServer(name) + for stream in (stream_type, *other_stream_types): + server.create_buffers(stream, num_buffers, width, height) + server.start_listener() + + self.client = client = VisionIpcClient(name, stream_type, conflate) + assert client.connect(True) + buffer_len = client.buffer_len + assert buffer_len is not None + return server, client, buffer_len def test_connect(self): - self.setup_vipc("camerad", 0) - assert self.client is not None - assert self.client.is_connected() + _, client, _ = self.setup_vipc("camerad", 0) + assert client.is_connected() def test_available_streams(self): stream_types = (0, 2) @@ -44,102 +44,86 @@ def test_available_streams(self): def test_buffers(self): width, height, num_buffers = 100, 200, 5 - self.setup_vipc("camerad", 0, num_buffers=num_buffers, width=width, height=height) - assert self.client is not None - assert self.client.width == width - assert self.client.height == height - assert self.client.buffer_len is not None and self.client.buffer_len > 0 - assert self.client.num_buffers == num_buffers + _, client, buffer_len = self.setup_vipc("camerad", 0, num_buffers=num_buffers, width=width, height=height) + assert client.width == width + assert client.height == height + assert buffer_len > 0 + assert client.num_buffers == num_buffers def test_send_single_buffer(self): - self.setup_vipc("camerad", 0) - assert self.server is not None - assert self.client is not None - assert self.client.buffer_len is not None - buf = bytearray(self.client.buffer_len) + server, client, buffer_len = self.setup_vipc("camerad", 0) + buf = bytearray(buffer_len) struct.pack_into("