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

> Learn to write pytest tests for dimOS IPC channels. Use make_frame_channel in fixtures to validate data integrity and cross-process attachment with descriptor API.

- Repository: [Dimensional/dimos](https://github.com/dimensionalOS/dimos)
- Tags: testing
- Published: 2026-03-15

---

**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`](https://github.com/dimensionalOS/dimos/blob/main/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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
@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`](https://github.com/dimensionalOS/dimos/blob/main/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.