How to Write pytest Tests for DimOS Inter-Process Communication (IPC) Channels

Create CPU-based shared memory channels using the make_frame_channel factory in pytest fixtures, then validate data integrity, sequence number tracking, and cross-process attachment through the descriptor and CPU_IPC_Factory.attach APIs.

The dimensionalOS/dimos repository implements module Inter-Process Communication (IPC) through a double-buffered shared-memory abstraction located in dimos/protocol/pubsub/shm/ipc_factory.py. Writing robust pytest tests for these channels requires understanding the FrameChannel abstract base class (lines 62‑99) and its concrete CpuShmChannel implementation (lines 133‑271). The following patterns demonstrate how to verify the single-slot buffering mechanism, descriptor-based serialization, and proper resource finalization.

Understanding the DimOS IPC Architecture

The FrameChannel Abstraction

At the heart of the system lies the FrameChannel class, which defines the contract for publishing and reading frames between DimOS modules. The concrete CpuShmChannel implements this contract using POSIX shared memory, providing a zero-copy transport for numpy arrays. A convenience factory function, make_frame_channel (lines 26‑31), instantiates the appropriate backend—currently CPU-only—handling the complexity of buffer allocation and metadata initialization.

Key Testing Considerations

When writing pytest tests for IPC, focus on these implementation details:

• Single-slot freshest frame: The channel maintains only the most recent frame; subsequent publish calls overwrite inactive buffers and flip the visible index.
• Descriptor-based attachment: The descriptor() method (lines 84‑92) returns a JSON-compatible dictionary containing shape, dtype, data_name, and ctrl_name, allowing separate processes to attach via CPU_IPC_Factory.attach.
• Ownership and cleanup: Channel creators register finalizers (_finalizer_data and _finalizer_ctrl) to unlink shared memory on garbage collection, while readers only close handles.

Setting Up pytest Fixtures for IPC Channels

Create a reusable fixture that instantiates a channel and ensures cleanup via the close() method. This pattern prevents shared memory leaks across test runs.

import numpy as np
import pytest
from dimos.protocol.pubsub.shm.ipc_factory import make_frame_channel

@pytest.fixture
def channel():
    # 64×48 grayscale frame (uint8)

    chan = make_frame_channel(shape=(64, 48), dtype=np.uint8)
    yield chan
    chan.close()  # Ensures SHM cleanup even if test fails

Testing Core Channel Operations

Validating Basic Publish-Read Round-Trips

Verify that data written via publish() is correctly retrieved by read(), including sequence number and timestamp metadata. The last_seq=-1 argument retrieves the freshest frame regardless of sequence history.

def test_publish_and_read(channel):
    # Create a synthetic frame

    frame = np.arange(64 * 48, dtype=np.uint8).reshape(64, 48)
    
    # Publish the frame

    channel.publish(frame)
    
    # Read back the most recent frame

    seq, ts, view = channel.read(last_seq=-1)
    assert view is not None
    np.testing.assert_array_equal(view, frame)
    
    # The sequence must have increased from its initial zero value

    assert seq > 0
    # Timestamp is a positive nanosecond value

    assert ts > 0

Detecting Fresh Frames with Sequence Numbers

The require_new parameter (implemented in the read method) allows modules to poll only for new data. Test this behavior to ensure consumers correctly identify when no new frame has arrived since the last sequence number.

def test_require_new_flag(channel):
    # First read returns a fresh frame

    seq0, _, view0 = channel.read(last_seq=-1)
    assert view0 is None  # Nothing published yet

    
    # Publish once

    channel.publish(np.zeros((64, 48), dtype=np.uint8))
    seq1, _, view1 = channel.read(last_seq=-1)
    assert view1 is not None
    
    # Subsequent read with require_new=True should return None

    seq2, _, view2 = channel.read(last_seq=seq1, require_new=True)
    assert view2 is None
    
    # With require=False it returns the same data without incrementing sequence

    seq3, _, view3 = channel.read(last_seq=seq1, require_new=False)
    assert view3 is not None
    assert seq3 == seq1

Testing Cross-Process Communication

Using Descriptors for Writer-Reader Attachment

Simulate multi-process scenarios by passing the descriptor dictionary between writer and reader instances. This mirrors real DimOS deployments where perception modules publish data and control modules consume it in separate processes.

def test_attach_reader_writer():
    # Writer side creates the channel and exports its descriptor

    writer = make_frame_channel(shape=(32, 32), dtype=np.float32)
    descriptor = writer.descriptor()
    
    # Reader attaches to the same SHM segment using the factory

    from dimos.protocol.pubsub.shm.ipc_factory import CPU_IPC_Factory
    reader = CPU_IPC_Factory.attach(descriptor)
    
    # Writer publishes a frame

    data = np.full((32, 32), 3.14, dtype=np.float32)
    writer.publish(data)
    
    # Reader sees the same data through the attached view

    _, _, view = reader.read(last_seq=-1)
    assert view is not None
    np.testing.assert_allclose(view, data)
    
    # Cleanup both handles

    writer.close()
    reader.close()

Parametrized Testing for Multiple Data Types

Validate that the IPC implementation handles various image and tensor formats by parametrizing fixtures over different shapes and numpy dtypes.

@pytest.fixture(params=[
    ((64, 48), np.uint8),
    ((32, 32, 3), np.float32),
    ((128, 128, 4), np.uint16),
])
def varied_channel(request):
    shape, dtype = request.param
    chan = make_frame_channel(shape=shape, dtype=dtype)
    yield chan, shape, dtype
    chan.close()

def test_varied_channels(varied_channel):
    chan, shape, dtype = varied_channel
    data = np.random.rand(*shape).astype(dtype)
    chan.publish(data)
    _, _, view = chan.read(last_seq=-1)
    assert view.shape == shape
    np.testing.assert_array_equal(view, data)

Managing Shared Memory Cleanup

The CpuShmChannel distinguishes between owners and readers during initialization. Only creators register finalizers that unlink shared memory segments when the object is garbage collected. Always explicitly call close() in fixture teardown to ensure immediate resource release and prevent EBUSY errors in subsequent tests. Readers calling close() only detach handles without destroying the underlying shared memory, allowing the writer to continue operating if the test scope permits.

Summary

• Use make_frame_channel to instantiate test channels with specific shapes and dtypes, wrapping them in pytest fixtures with explicit close() cleanup.
• Verify the single-slot contract by checking that sequence numbers increment on new publishes and that require_new=True correctly returns None for stale reads.
• Test multi-process workflows by serializing descriptor() output and reconstructing channels via CPU_IPC_Factory.attach.
• Parametrize over data types to ensure the shared memory buffer handles various numpy arrays used in perception and control stacks.
• Reference implementation details in dimos/protocol/pubsub/shm/ipc_factory.py when debugging attachment or buffer overwrite behaviors.

Frequently Asked Questions

How do I clean up shared memory resources after pytest tests?

Always yield your channel from a fixture and call chan.close() in the teardown phase. The CpuShmChannel sets finalizers only for the creating process (owner), which automatically unlink shared memory segments on garbage collection, but explicit cleanup prevents resource exhaustion during large test suites.

Can I test IPC between multiple processes in a single test function?

Yes. Instantiate a writer channel using make_frame_channel, export its descriptor(), then reconstruct a reader instance via CPU_IPC_Factory.attach(descriptor). Both instances operate on the same underlying POSIX shared memory segments, allowing you to verify cross-process data integrity within a single pytest function.

What data types and array shapes are supported by CpuShmChannel?

The channel supports any numpy dtype (including uint8, float32, and uint16) and arbitrary shapes compatible with shared memory allocation. Use pytest parametrization to validate your specific use cases, such as grayscale images (64, 48), RGB tensors (32, 32, 3), or higher-dimensional feature maps.

How does the sequence number help in testing IPC channels?

The sequence number returned by read() acts as a monotonic counter that increments with each publish() call. In tests, compare sequence numbers across reads to verify that require_new=True correctly filters unchanged data and that the double-buffered implementation properly advances indices when new frames arrive.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}