Skip to main content

​
Test Runner (runner.py)

The runner.py module is the main test orchestrator in the Roboticks testing framework. It manages the complete test lifecycle: starting the device manager, deploying compositions, executing pytest, and reporting results.

​
Overview

When the test runner detects test-framework/runner.py, it enters test-framework mode and delegates all orchestration to this script: 
Test Runner Binary
        |
        v
Detects test-framework/runner.py exists
        |
        v
python3 /opt/roboticks/test-framework/runner.py
        |
        v
+--------------------------------------------+
|              runner.py                     |
|                                            |
|  1. Setup - verify paths                   |
|  2. Start Device Manager (Python bindings) |
|  3. Verify Registration                    |
|  4. Wait for Session Creation              |
|  5. Discover & Start Compositions          |
|  6. Run Pytest                             |
|  7. Cleanup                                |
+--------------------------------------------+
Both cloud runners and self-hosted runners use the same test runner binary. The only difference is where it runs (EC2 vs your infrastructure). The execution logic is identical.

​
RoboticksTestRunner Class

The main class that orchestrates test execution: 
class RoboticksTestRunner:
    """Orchestrates roboticks test execution in Docker environment."""

    def __init__(self):
        # Paths
        self.roboticks_root = Path(os.environ.get("ROBOTICKS_ROOT", "/opt/roboticks"))
        self.deployment_path = self.roboticks_root / "deployment"
        self.test_output_path = self.roboticks_root / "test-output"
        self.device_manager_bin = self.roboticks_root / "install" / "bin" / "device-manager"

        # Configuration
        self.project_secret = os.environ.get("ROBOTICKS_PROJECT_SECRET", "")
        self.api_url = os.environ.get("ROBOTICKS_API_URL", "https://api.roboticks.io")
        self.target_composition = os.environ.get("ROBOTICKS_COMPOSITION")
        self.composition_mode = os.environ.get("ROBOTICKS_COMPOSITION_MODE", "prod")
        self.test_command = os.environ.get("ROBOTICKS_TEST_COMMAND", "pytest tests/ -v")
        self.test_timeout = int(os.environ.get("ROBOTICKS_TEST_TIMEOUT", "3600"))

​
Execution Flow

​
1. Setup Phase

Verifies all required paths and files exist: 
def setup(self) -> bool:
    """Set up test environment."""
    # Verify device manager binary exists
    if not self.device_manager_bin.exists():
        logger.error("Device manager binary not found")
        return False

    # Verify deployment path
    if not self.deployment_path.exists():
        logger.warning("Deployment path does not exist")

    return True

​
2. Start Device Manager

Uses Python bindings to start the device manager in-process: 
def start_device_manager(self) -> bool:
    """Start the device manager via Python bindings."""
    import roboticks_device

    # Create device manager instance
    self.device_manager = roboticks_device.DeviceManager()

    # Set up callbacks for session events
    self.device_manager.set_session_created_callback(self.on_session_created)
    self.device_manager.set_log_published_callback(self.on_log_published)

    # Initialize logging
    self.device_manager.initialize_logging("INFO")

    # Start device manager background thread
    if not self.device_manager.start(str(device_config_path)):
        return False

    return True

​
3. Verify Registration

Waits for device registration with the backend. During test execution, the device manager connects to Roboticks backend services just like any regular device - it authenticates, receives certificates, and establishes MQTT connections. 
def verify_registration(self, timeout: int = 30) -> bool:
    """Wait for device registration."""
    if self.device_manager.wait_for_registration(timeout):
        self.device_dsn = self.device_manager.get_dsn()
        logger.info(f"Registered with DSN: {self.device_dsn}")
        return True
    return False
Test Devices vs Production DevicesTest devices register with a special TEST_DEVICE=true flag. This means:
  • They follow the same authentication flow as production devices
  • They appear in your dashboard for monitoring during tests
  • They are automatically marked as test devices
  • They do not consume your purchased Device units
This allows you to run unlimited tests without affecting your billing.

​
4. Deploy Compositions

Discovers and starts compositions from the deployment directory: 
def run_deployment(self) -> bool:
    """Discover and start compositions."""
    # Discover compositions
    count = self.device_manager.discover_compositions(str(self.deployment_path))
    logger.info(f"Discovered {count} compositions")

    compositions = self.device_manager.get_composition_names()

    # Start specific composition or all
    if self.target_composition:
        if self.target_composition not in compositions:
            raise CompositionNotFoundError(self.target_composition, compositions)
        self.device_manager.start_composition(
            self.target_composition,
            mode=self.composition_mode
        )
    else:
        self.device_manager.start_all_compositions(mode=self.composition_mode)

    return True

​
5. Run Tests

Executes pytest with optional progress reporting: 
def run_tests(self) -> bool:
    """Run tests using pytest."""
    # Discover tests first
    test_nodeids = self.discover_tests()

    # Report discovered tests to backend
    if self.test_job_id:
        self.report_discovered_tests(test_nodeids)

    # Build pytest command
    cmd = shlex.split(self.test_command)

    # Add progress reporting plugin if job ID is set
    if self.test_job_id:
        cmd.extend([
            "--roboticks-job-id", self.test_job_id,
            "--roboticks-api-url", self.api_url,
            "-p", "pytest_roboticks_progress",
        ])

    # Run pytest
    result = subprocess.run(cmd, timeout=self.test_timeout)
    return result.returncode == 0

​
6. Cleanup

Stops compositions and device manager: 
def cleanup(self) -> None:
    """Clean up test environment."""
    if self.device_manager:
        # Stop active session
        session_id = self.device_manager.get_active_session_id()
        if session_id:
            self.device_manager.stop_active_session()

        # Stop device manager
        if self.device_manager.is_running():
            self.device_manager.stop(timeout_seconds=5)

​
Customizing runner.py

You can customize the runner for your needs. Common customizations:

​
Custom Pre-Test Setup

class MyTestRunner(RoboticksTestRunner):
    def setup(self) -> bool:
        if not super().setup():
            return False

        # Custom setup: load test fixtures
        self.load_test_data()

        # Custom setup: configure external services
        self.setup_mock_server()

        return True

    def load_test_data(self):
        """Load custom test data."""
        data_path = self.roboticks_root / "test-data"
        # ... load data

    def setup_mock_server(self):
        """Start mock server for testing."""
        # ... start mock server

​
Custom Device Configuration

def create_custom_device_config(self):
    """Generate custom device configuration."""
    config = {
        "project_secret": self.project_secret,
        "device_type": "custom-test-runner",
        "api_url": self.api_url,
        "storage_path": "/var/roboticks",
        "session": {
            "auto_start": True
        },
        # Custom settings
        "telemetry": {
            "enabled": True,
            "interval_seconds": 1
        }
    }

    config_path = self.roboticks_root / "device" / "device-config.yaml"
    with open(config_path, "w") as f:
        yaml.dump(config, f)

​
Custom Test Command

# Via environment variable
os.environ["ROBOTICKS_TEST_COMMAND"] = "pytest tests/ -v --tb=long -x"

# Or in custom runner
class MyTestRunner(RoboticksTestRunner):
    def __init__(self):
        super().__init__()
        self.test_command = "pytest tests/ -v --cov=my_module --html=report.html"

​
Targeting Specific Compositions

# Via environment variable
os.environ["ROBOTICKS_COMPOSITION"] = "HelloWorldComposition"
os.environ["ROBOTICKS_COMPOSITION_MODE"] = "dev"

# This will only start HelloWorldComposition in dev mode

​
Python Bindings API

The runner uses the roboticks_device Python bindings to control the device manager:

​
DeviceManager Class

class DeviceManager:
    # Lifecycle
    def start(config_path: str) -> bool
    def stop(timeout_seconds: int = 5) -> bool
    def is_running() -> bool

    # Registration
    def wait_for_registration(timeout_seconds: int) -> bool
    def is_registered() -> bool
    def get_dsn() -> str

    # Composition Control
    def discover_compositions(deployment_path: str) -> int
    def get_composition_names() -> list[str]
    def start_all_compositions(mode: str = "prod") -> bool
    def stop_all_compositions() -> bool
    def start_composition(name: str, mode: str = "prod") -> bool
    def stop_composition(name: str) -> bool
    def is_composition_running(name: str) -> bool

    # Session Management
    def get_active_session_id() -> str | None
    def stop_active_session() -> bool

    # Callbacks
    def set_session_created_callback(callback: Callable[[str], None])
    def set_log_published_callback(callback: Callable[[str, str, str], None])

    # Logging
    def initialize_logging(level: str = "INFO")

​
Environment Variables Reference

VariableDescriptionDefault
ROBOTICKS_ROOTBase path for all roboticks files/opt/roboticks
ROBOTICKS_PROJECT_SECRETProject secret for device authRequired
ROBOTICKS_API_URLBackend API URLhttps://api.roboticks.io
ROBOTICKS_TEST_JOB_IDJob ID for progress reporting-
ROBOTICKS_JOB_TOKENAuth token for job API calls-
ROBOTICKS_ORG_SLUGOrganization slug-
ROBOTICKS_PROJECT_SLUGProject slug-
ROBOTICKS_COMPOSITIONTarget composition (optional)All
ROBOTICKS_COMPOSITION_MODEMode: prod or devprod
ROBOTICKS_TEST_COMMANDPytest command to runpytest tests/ -v
ROBOTICKS_TEST_TIMEOUTMax test duration (seconds)3600

​
Entry Point

The main() function is the entry point called by the self-hosted runner: 
def main() -> None:
    """Main entry point for the test runner."""
    # Set up logging
    roboticks_root = Path(os.environ.get("ROBOTICKS_ROOT", "/opt/roboticks"))
    log_file = roboticks_root / "test-output" / "test-run.log"
    setup_logging(log_file)

    # Create and run
    runner = RoboticksTestRunner()
    exit_code = runner.run()

    sys.exit(exit_code)


if __name__ == "__main__":
    main()
The runner uses os._exit() in the finally block to ensure clean termination, especially when Python bindings are involved. This prevents hanging on background threads.

​
Error Handling

The runner handles errors at each stage and provides detailed logging: 
def run(self) -> int:
    """Run the complete test suite."""
    try:
        if not self.setup():
            return 1

        if not self.start_device_manager():
            return 1

        if not self.verify_registration():
            return 1

        if not self.verify_session_created():
            return 1

        if not self.run_deployment():
            return 1

        if not self.run_tests():
            return 1

        return 0

    except Exception as e:
        logger.error(f"Test run failed: {e}", exc_info=True)
        return 1

    finally:
        self.cleanup(print_dm_logs=True)

​
Next Steps

Writing Tests

Learn how to write tests using ZMQ monitoring.

Packaging Tests

Package your tests for the platform.