Skip to main content
Version: Next

Python Customization

This guide covers basic Python integration and testing capabilities for Merobox CLI.

Overview​

Merobox provides basic Python integration for testing and automation:

  • Testing Integration: Use Merobox in your Python test suites
  • Basic Automation: Automate Merobox workflows with Python scripts
  • Environment Management: Manage test environments programmatically

Basic Python Integration​

Using Merobox in Python Scripts​

# basic_integration.py
import subprocess
import json

def run_merobox_command(command):
    """Run a Merobox command and return the result."""
    try:
        result = subprocess.run(
            ['merobox'] + command,
            capture_output=True,
            text=True,
            check=True
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"Error running merobox command: {e}")
        return None

# Example usage
def start_nodes(count=2):
    """Start Merobox nodes."""
    output = run_merobox_command(['run', '--count', str(count)])
    print(f"Started {count} nodes: {output}")

def check_health():
    """Check node health."""
    output = run_merobox_command(['health'])
    print(f"Health status: {output}")

def stop_nodes():
    """Stop all nodes."""
    output = run_merobox_command(['stop', '--all'])
    print(f"Stopped nodes: {output}")

Testing Integration​

Basic Testing Setup​

# test_merobox.py
import pytest
import subprocess
import time

class MeroboxTestHelper:
    def __init__(self):
        self.nodes = []

    def start_nodes(self, count=2):
        """Start Merobox nodes for testing."""
        result = subprocess.run(
            ['merobox', 'run', '--count', str(count)],
            capture_output=True,
            text=True
        )
        if result.returncode == 0:
            self.nodes = [f"calimero-node-{i+1}" for i in range(count)]
            return True
        return False

    def stop_nodes(self):
        """Stop all test nodes."""
        subprocess.run(['merobox', 'stop', '--all'])
        self.nodes = []

    def check_health(self):
        """Check health of all nodes."""
        result = subprocess.run(
            ['merobox', 'health'],
            capture_output=True,
            text=True
        )
        return result.returncode == 0

# Test fixture
@pytest.fixture
def merobox_helper():
    helper = MeroboxTestHelper()
    yield helper
    helper.stop_nodes()

def test_basic_functionality(merobox_helper):
    """Test basic Merobox functionality."""
    assert merobox_helper.start_nodes(2)
    assert merobox_helper.check_health()
    merobox_helper.stop_nodes()

Workflow Automation​

Running Workflows from Python​

# workflow_automation.py
import subprocess
import yaml
import tempfile
import os

def create_workflow_file(workflow_config):
    """Create a temporary workflow file."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
        yaml.dump(workflow_config, f)
        return f.name

def run_workflow(workflow_config):
    """Run a Merobox workflow from Python."""
    workflow_file = create_workflow_file(workflow_config)

    try:
        result = subprocess.run(
            ['merobox', 'bootstrap', 'run', workflow_file],
            capture_output=True,
            text=True,
            check=True
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"Workflow failed: {e}")
        return None
    finally:
        os.unlink(workflow_file)

# Example workflow configuration
def create_test_workflow():
    """Create a simple test workflow."""
    return {
        'description': 'Test workflow from Python',
        'name': 'Python Test Workflow',
        'nodes': {
            'chain_id': 'testnet-1',
            'count': 1,
            'image': 'ghcr.io/calimero-network/merod:edge'
        },
        'steps': [
            {
                'name': 'Test Step',
                'type': 'script',
                'script': 'echo "Hello from Python workflow!"'
            }
        ],
        'stop_all_nodes': True
    }

# Run the workflow
workflow_config = create_test_workflow()
result = run_workflow(workflow_config)
print(f"Workflow result: {result}")

Best Practices​

Error Handling​

# Always handle errors properly
def safe_merobox_command(command):
    """Run Merobox command with proper error handling."""
    try:
        result = subprocess.run(
            ['merobox'] + command,
            capture_output=True,
            text=True,
            check=True
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"Command failed: {e}")
        print(f"Error output: {e.stderr}")
        return None
    except FileNotFoundError:
        print("Merobox not found. Please install it first.")
        return None

Resource Cleanup​

# Always clean up resources
def test_with_cleanup():
    """Test with proper cleanup."""
    try:
        # Start nodes
        subprocess.run(['merobox', 'run', '--count', '2'])

        # Run tests
        result = subprocess.run(['merobox', 'health'])
        assert result.returncode == 0

    finally:
        # Always cleanup
        subprocess.run(['merobox', 'stop', '--all'])

Next Steps​

Now that you understand basic Python integration with Merobox:

Was this page helpful?
Need some help? Check Support page