Testing & Development
Test harness, pytest fixtures, development setup, building, and release process
Test Harness
The merobox.testing module provides context managers and pytest fixtures for spinning up clusters and running workflows in test suites.
cluster() Context Manager
Spins up an isolated cluster of merod nodes for the duration of the block, tearing down on exit.
async with cluster(
count=3, # number of nodes to spawn
prefix="test", # name prefix for containers
image="merod:latest", # Docker image to use
) as env:
# env is a dict with:
# nodes — list of node names
# endpoints — dict mapping node name to URL
assert len(env["nodes"]) == 3
pytest_cluster Fixture
A pytest fixture that wraps cluster() for seamless test integration.
from merobox.testing import pytest_cluster
@pytest.fixture(scope="module") # scope: function | class | module | session
async def my_cluster():
async with pytest_cluster(scope="module") as env:
yield env
async def test_node_health(my_cluster):
for name, url in my_cluster["endpoints"].items():
# assert node is reachable
...
Workflow Testing
workflow() Context Manager
Executes a workflow YAML file and exposes the resulting environment for assertions.
async with workflow(
workflow_path="tests/e2e.yaml",
prefix="e2e",
) as env:
# env is a dict with:
# workflow_result — final StepResult
# nodes — list of node names
# endpoints — dict mapping node name to URL
# manager — NodeManager instance for direct control
assert env["workflow_result"].success
pytest_workflow Fixture
A full-featured pytest fixture for workflow-based integration tests.
@pytest.fixture(scope="module")
async def e2e_env():
async with pytest_workflow(
workflow_path="tests/e2e.yaml",
prefix="e2e",
image="merod:latest",
chain_id="localnet",
wait_for_ready=True,
scope="module",
) as env:
yield env
async def test_workflow_passes(e2e_env):
assert e2e_env["workflow_result"].success
Fixture Parameters
- workflow_path — path to the workflow YAML file
- prefix — name prefix for containers and resources
- image — Docker image override for merod nodes
- chain_id — NEAR chain ID for sandbox configuration
- wait_for_ready — block until all nodes are healthy
- scope — pytest fixture scope (function, class, module, session)
Environment Dict Keys
- workflow_result — final StepResult from the workflow execution
- nodes — list of node name strings
- endpoints — dict mapping node names to HTTP endpoint URLs
- manager — the active NodeManager instance for direct node control
Development Setup
Prerequisites
Local Development Setup
Clone and enter the repository
cd merobox
Create a virtual environment
source .venv/bin/activate
Install dependencies
# or: make install
Building
All build tasks are managed via make. Run make help for the full list.
Running Tests
Tests use pytest with async support via pytest-asyncio.
All tests
# equivalent to: pytest tests/ -v
Specific tests
pytest tests/ -k "test_cluster" -v
Adding New Commands
merobox uses Click for CLI command definitions. Each command lives in its own file under merobox/commands/.
Create the command file
Create a new file in merobox/commands/, e.g. merobox/commands/my_command.py.
Implement the Click command
@click.command()
@click.argument("name")
@click.option("--verbose", is_flag=True, help="Enable verbose output")
def my_command(name, verbose):
"""Short description of the command."""
...
Register the command
Add the import to merobox/commands/__init__.py and register with the CLI group.
# In the CLI group registration:
cli.add_command(my_command)
Update __all__
Add the command name to __all__ in merobox/commands/__init__.py for clean imports.
Adding New Step Types
Step types extend BaseStep and register with the step factory for use in workflows.
Create the step file
Create a new file in merobox/steps/, e.g. merobox/steps/my_step.py.
Inherit from BaseStep
class MyStep(BaseStep):
step_type = "my_step"
Implement required methods
return ["target", "value"]
def _validate_field_types(self) -> None:
# Validate config field types
if not isinstance(self.config["target"], str):
raise StepValidationError("target must be a string")
async def execute(self, context) -> StepResult:
# Step logic here
return ok({"result": "value"})
Register in the step factory
Add the step type mapping in the executor’s step factory registry.
STEP_REGISTRY["my_step"] = MyStep
Release Process
Releases are version-gated and fully automated once a tag is pushed.
Version Source of Truth
The version string lives in merobox/__init__.py and is the single source of truth for the package version.
__version__ = "0.4.2"
Automated Pipeline
Auto-tag
CI detects version bump in __init__.py and creates a git tag v0.4.2.
Build binaries
Wheel and sdist are built for all supported platforms (Linux, macOS, Windows).
GitHub Release
A GitHub Release is created with the tag, changelog, and build artifacts attached.
PyPI Publish
The wheel and sdist are uploaded to PyPI for pip install merobox access.
Project Structure
__init__.py # Version (__version__), package root
cli.py # Click CLI entry point
executor.py # WorkflowExecutor orchestration
config.py # Configuration loading and merging
errors.py # MeroboxError hierarchy
commands/ # CLI command implementations
__init__.py
run.py # merobox run
stop.py # merobox stop
health.py # merobox health
logs.py # merobox logs
nuke.py # merobox nuke
remote.py # merobox remote
bootstrap.py # merobox bootstrap
steps/ # Workflow step implementations
base.py # BaseStep, StepResult, ok(), fail()
call.py # call step
assert_step.py # assert step
parallel.py # parallel step
repeat.py # repeat step
script.py # script step
...
managers/ # Node management backends
docker.py # DockerManager
binary.py # BinaryManager
sandbox.py # SandboxManager
remote.py # RemoteNodeManager
testing/ # Test harness module
__init__.py # cluster(), workflow(), fixtures
auth/ # Authentication
manager.py # AuthManager, JWT handling
tests/ # Test suite
Makefile # Build and task automation
pyproject.toml # Package configuration