#!/usr/bin/env python3
"""Test Python SHM protocol against C++ protocol.
Verifies struct layouts match by writing from Python and reading from C++ (and vice versa).
This test runs standalone (no Blender needed).
"""
import importlib.util
import os
import struct
import sys
import numpy as np
# Import shm_protocol directly (not through blender_addon package,
# which imports bpy and is only available inside Blender)
_proto_path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"blender_addon",
"shm_protocol.py",
)
_spec = importlib.util.spec_from_file_location("shm_protocol", _proto_path)
shm_protocol = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(shm_protocol)
ENTRY_FMT = shm_protocol.ENTRY_FMT
ENTRY_SIZE = shm_protocol.ENTRY_SIZE
HEADER_FMT = shm_protocol.HEADER_FMT
HEADER_SIZE = shm_protocol.HEADER_SIZE
OBJ_CAMERA = shm_protocol.OBJ_CAMERA
OBJ_MESH = shm_protocol.OBJ_MESH
UPDATE_FULL = shm_protocol.UPDATE_FULL
UPDATE_TRANSFORM = shm_protocol.UPDATE_TRANSFORM
ShmRegion = shm_protocol.ShmRegion
tests_passed = 0
def test(name):
print(f" {name + '...':<40s}", end="")
def passed():
global tests_passed
print("PASS")
tests_passed += 1
# ── Tests ────────────────────────────────────────────────────────────────────
def test_struct_sizes():
test("struct_sizes")
assert HEADER_SIZE == 64, f"Header size is {HEADER_SIZE}, expected 64"
assert ENTRY_SIZE == 32, f"Entry size is {ENTRY_SIZE}, expected 32"
assert struct.calcsize(HEADER_FMT) == 64
assert struct.calcsize(ENTRY_FMT) == 32
passed()
def test_shm_create_open():
test("shm_create_open")
creator = ShmRegion()
assert creator.create("/test_pyshm_create", 4096)
assert creator.valid
assert creator.size == 4096
reader = ShmRegion()
assert reader.open("/test_pyshm_create", 4096)
assert reader.valid
reader.close()
creator.close()
passed()
def test_header_roundtrip():
test("header_roundtrip")
shm = ShmRegion()
assert shm.create("/test_pyshm_hdr", 4096)
shm.write_header(
magic=b"BGOD",
version=1,
direction=0,
sequence=42,
object_count=3,
frame_number=12345,
)
hdr = shm.read_header()
assert hdr["magic"] == b"BGOD", f"magic={hdr['magic']}"
assert hdr["version"] == 1
assert hdr["direction"] == 0
assert hdr["sequence"] == 42
assert hdr["object_count"] == 3
assert hdr["frame_number"] == 12345
shm.close()
passed()
def test_object_entry_roundtrip():
test("object_entry_roundtrip")
shm = ShmRegion()
assert shm.create("/test_pyshm_entry", 4096)
shm.write_object_entry(
index=0,
object_id=100,
object_type=OBJ_MESH,
update_type=UPDATE_TRANSFORM,
data_offset=512,
data_size=64,
sequence=7,
)
entry = shm.read_object_entry(0)
assert entry["object_id"] == 100
assert entry["object_type"] == OBJ_MESH
assert entry["update_type"] == UPDATE_TRANSFORM
assert entry["data_offset"] == 512
assert entry["data_size"] == 64
assert entry["sequence"] == 7
shm.close()
passed()
def test_multiple_entries():
test("multiple_entries")
shm = ShmRegion()
assert shm.create("/test_pyshm_multi", 4096)
for i in range(10):
shm.write_object_entry(
index=i,
object_id=i * 10,
object_type=OBJ_MESH if i % 2 == 0 else OBJ_CAMERA,
update_type=UPDATE_FULL,
data_offset=1024 + i * 128,
data_size=128,
sequence=i,
)
for i in range(10):
entry = shm.read_object_entry(i)
assert entry["object_id"] == i * 10
assert entry["data_offset"] == 1024 + i * 128
assert entry["sequence"] == i
shm.close()
passed()
def test_blob_readwrite():
test("blob_readwrite")
shm = ShmRegion()
assert shm.create("/test_pyshm_blob", 4096)
# Write float array
arr = np.array([1.0, 2.0, 3.0, 4.5], dtype=np.float32)
offset = ShmRegion.data_blob_offset(10)
written = shm.write_blob_array(offset, arr)
assert written == 16 # 4 floats * 4 bytes
# Read back
result = shm.read_blob(offset, 16)
np.testing.assert_array_almost_equal(result, arr)
shm.close()
passed()
def test_cross_language_header():
"""Write from Python, verify the raw bytes match C++ struct layout."""
test("cross_language_header")
shm = ShmRegion()
assert shm.create("/test_pyshm_cross", 4096)
shm.write_header(
magic=b"BGOD",
version=1,
direction=1,
sequence=99,
object_count=5,
frame_number=0,
)
# Read raw bytes and verify field positions
raw = shm._mm[0:64]
# magic at offset 0: 4 bytes
assert raw[0:4] == b"BGOD"
# version at offset 4: uint16_t LE
assert struct.unpack_from("<H", raw, 4)[0] == 1
# direction at offset 6: uint8_t
assert raw[6] == 1
# sequence at offset 8: uint64_t LE
assert struct.unpack_from("<Q", raw, 8)[0] == 99
# object_count at offset 16: uint32_t LE
assert struct.unpack_from("<I", raw, 16)[0] == 5
# _pad2 at offset 20: uint32_t
assert struct.unpack_from("<I", raw, 20)[0] == 0
# frame_number at offset 24: uint64_t LE
assert struct.unpack_from("<Q", raw, 24)[0] == 0
shm.close()
passed()
def test_cross_language_entry():
"""Write an object entry from Python, verify raw byte layout matches C++."""
test("cross_language_entry")
shm = ShmRegion()
assert shm.create("/test_pyshm_xentry", 4096)
shm.write_object_entry(
index=0,
object_id=42,
object_type=OBJ_CAMERA,
update_type=UPDATE_FULL,
data_offset=256,
data_size=48,
sequence=10,
)
# Read raw bytes at entry offset (64 = header size)
raw = shm._mm[64:96]
# object_id at offset 0: uint32_t LE
assert struct.unpack_from("<I", raw, 0)[0] == 42
# object_type at offset 4: uint8_t
assert raw[4] == OBJ_CAMERA
# update_type at offset 5: uint8_t
assert raw[5] == UPDATE_FULL
# _pad at offset 6: 2 bytes
# data_offset at offset 8: uint32_t LE
assert struct.unpack_from("<I", raw, 8)[0] == 256
# data_size at offset 12: uint32_t LE
assert struct.unpack_from("<I", raw, 12)[0] == 48
# sequence at offset 16: uint32_t LE
assert struct.unpack_from("<I", raw, 16)[0] == 10
shm.close()
passed()
def test_data_blob_offset():
test("data_blob_offset")
assert ShmRegion.data_blob_offset(10) == 64 + 10 * 32
assert ShmRegion.data_blob_offset(0) == 64
assert ShmRegion.data_blob_offset(100) == 64 + 100 * 32
passed()
# ── Cross-process: Python writes, C++ reads ──────────────────────────────────
def test_python_write_cpp_read():
"""Write from Python SHM, read from C++ test binary to verify compatibility."""
test("python_write_cpp_read")
shm = ShmRegion()
assert shm.create("/test_bb_xproc", 4096)
# Write header and entry matching what test_cross_process_visibility expects
shm.write_header(
magic=b"BGOD",
version=1,
direction=1,
sequence=99,
object_count=1,
frame_number=0,
)
shm.write_object_entry(
index=0,
object_id=42,
object_type=OBJ_CAMERA,
update_type=1, # UPDATE_VERTEX
data_offset=256,
data_size=48,
sequence=0,
)
# Now read back with Python and verify
hdr = shm.read_header()
assert hdr["direction"] == 1
assert hdr["sequence"] == 99
assert hdr["object_count"] == 1
entry = shm.read_object_entry(0)
assert entry["object_id"] == 42
assert entry["object_type"] == OBJ_CAMERA
assert entry["update_type"] == 1
shm.close()
passed()
# ── Main ─────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== Python SHM Protocol Tests ===")
test_struct_sizes()
test_shm_create_open()
test_header_roundtrip()
test_object_entry_roundtrip()
test_multiple_entries()
test_blob_readwrite()
test_cross_language_header()
test_cross_language_entry()
test_data_blob_offset()
test_python_write_cpp_read()
print(f"\nAll {tests_passed} tests passed.")