pytest-arrakis 0.4.1

Pytest plugin providing Arrakis fixtures for testing

pytest-arrakis

Pytest plugin providing Arrakis fixtures for testing

---

Resources

• Source Code
• Issue Tracker

Installation

pip install pytest-arrakis

To enable publish fixtures (including a real Kafka-backed server), install the publish extra:

pip install pytest-arrakis[publish]

Fixtures

Core fixtures

These are available out of the box:

Fixture	Scope	Description
arrakis_channel_configs	session	List of paths to the bundled per-publisher config files
arrakis_backend	session	A MockBackend configured with all bundled channels (regular and replay)
arrakis_server	session	A running mock ArrakisFlightServer with a replay registry (sets ARRAKIS_SERVER env var)
arrakis_channels	session	Dict of channel metadata loaded from the bundled TOML (regular channels only)
arrakis_mock_publisher	function	A mocked Publisher with FakeProducer for sink testing (no server/broker needed)

Replay fixtures

These provide replay-specific resources for testing replay functionality:

Fixture	Scope	Description
arrakis_replay_channel_configs	session	List of paths to the bundled replay publisher config files
arrakis_replay_params	session	A ReplayParams for the bundled FAKE_REPLAY window (fixed 24-hour period)
arrakis_replay_channels	session	Dict of channel metadata from the replay configs only
arrakis_replay_mock_publisher	function	A mocked replay Publisher with FakeProducer for sink testing

Publish fixtures (requires publish extra)

These are conditionally registered when arrakis-backend-kafka and pytest-kafka-broker are installed:

Fixture	Scope	Description
arrakis_publish_backend	function	A real KafkaBackend pointed at an embedded Kafka broker (includes replay configs)
arrakis_publish_server	function	A publish-capable ArrakisFlightServer with a replay registry
arrakis_publisher	function	A real Publisher registered against the embedded broker
arrakis_replay_publisher	function	A real Publisher for the replay namespace against the embedded broker

Usage

Fetching data from a mock server

def test_fetch(arrakis_server):
    import arrakis

    block = arrakis.fetch(["H1:TEST-CHANNEL_SIN"], 0, 1)
    for channel, series in block.items():
        assert len(series) > 0

Testing a publisher without a real broker

def test_publish(arrakis_mock_publisher):
    publisher, fake_producer, published_blocks = arrakis_mock_publisher

    # publish a block through your code under test ...
    assert len(published_blocks) == 1

End-to-end publish testing (requires publish extra)

def test_roundtrip(arrakis_publisher):
    # arrakis_publisher is a real Publisher connected to an embedded Kafka broker
    ...

Testing replay channel discovery

def test_replay_find(arrakis_server, arrakis_replay_params):
    from arrakis import Client

    client = Client(arrakis_server.url)
    # replay channels are only visible when querying with the replay ID
    replay_channels = list(client.find(replay_id="FAKE_REPLAY"))
    assert len(replay_channels) > 0

Testing a replay publisher without a real broker

def test_replay_publish(arrakis_replay_mock_publisher):
    publisher, fake_producer, published_blocks = arrakis_replay_mock_publisher

    # publish a block through your code under test ...
    assert len(published_blocks) == 1

Bundled test channels

The plugin ships with a set of test channel definitions:

Regular channels

Channel	Publisher	Sample Rate	Data Type
H1:TEST-CHANNEL_SIN	FAKE_H1	512	float32
H1:TEST-CHANNEL_COS	FAKE_H1	16384	float64
H1:TEST-STATE_ONES	FAKE_H1	16	int32
L1:TEST-NOISE	FAKE_L1	4096	float32

Replay channels

These are tagged with replay_id = "FAKE_REPLAY" and are only visible when querying with the corresponding replay ID:

Channel	Publisher	Sample Rate	Data Type
H1:REPLAY-CHANNEL_SIN	FAKE_REPLAY_H1	512	float32
H1:REPLAY-STATE_ONES	FAKE_REPLAY_H1	16	int32

CI setup

GitLab CI

The core fixtures work without any extra system dependencies. However, the publish extra uses pytest-kafka-broker, which starts an embedded Kafka broker that requires a JVM at runtime. In Debian/Ubuntu-based CI images (including the standard python: Docker images), install it in a before_script:

test:
  image: python:3.13
  before_script:
    - apt-get update && apt-get install -y --no-install-recommends default-jre-headless

If you only use the core fixtures (no publish extra), the JVM is not needed.