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:
- Node Management - Complete node management guide
- Workflows - Workflow system and automation
- Environment Variables - Configuration options
- Troubleshooting - Common issues and solutions
Was this page helpful?