Quickstart

Get from zero to a working Moler script in a few minutes.

This guide walks through two approaches: running commands via a device (recommended for test suites) and running commands via a direct connection (useful for quick scripts).

1. Install Moler

   Install the latest stable release from PyPI:

   pip install moler
   

   Moler requires Python 3.8 or later and runs on POSIX/Unix systems. See Installation for full details.

2. Create a devices YAML config

   Moler uses a YAML file to describe the devices your tests will connect to. Create my_devices.yml in your project directory:

   LOGGER:
     PATH: ./logs
     DATE_FORMAT: "%H:%M:%S"
   
   DEVICES:
     DEFAULT_CONNECTION:
       CONNECTION_DESC:
         io_type: terminal
         variant: threaded
   
     MyMachine:
       DEVICE_CLASS: moler.device.unixlocal.UnixLocal
   
     RebexTestMachine:
       DEVICE_CLASS: moler.device.unixremote.UnixRemote
       CONNECTION_HOPS:
         UNIX_LOCAL:                # from state
           UNIX_REMOTE:             # to state
             execute_command: ssh   # via command
             command_params:        # with params
               expected_prompt: demo@
               host: test.rebex.net
               login: demo
               password: password
               set_timeout: None
   

   The CONNECTION_HOPS section tells Moler how to transition between device states. Your test code calls goto_state("UNIX_REMOTE") and Moler figures out the path — you don’t hard-code SSH invocations in your tests.

   The LOGGER section is optional but recommended. When present, Moler writes a moler.log summarising activity across all devices, plus a per-device log for every connection.

3. Run a command via DeviceFactory

   Load the config, get a device, and run a command:

   from moler.config import load_config
   from moler.device import DeviceFactory
   
   load_config(config='my_devices.yml')                    # describe available devices
   my_unix = DeviceFactory.get_device(name='MyMachine')    # get device by name
   ps_cmd = my_unix.get_cmd(cmd_name="ps",                 # get command from device
                            cmd_params={"options": "-ef"})
   
   processes_info = ps_cmd()                               # run it; returns parsed result
   for proc_info in processes_info:
       if 'python' in proc_info['CMD']:
           print("PID: {info[PID]} CMD: {info[CMD]}".format(info=proc_info))
   

   PID: 1817 CMD: /usr/bin/python /usr/share/system-config-printer/applet.py
   PID: 21825 CMD: /usr/bin/python /home/gl/moler/examples/command/unix_ps.py
   

   Calling ps_cmd() runs the command synchronously and returns the parsed result as a list of dicts. There is no string parsing in your test code.

   To reach a remote device, first tell the device to transition to the right state:

   remote_unix = DeviceFactory.get_device(name='RebexTestMachine')  # starts in local shell
   remote_unix.goto_state(state="UNIX_REMOTE")                      # transitions via SSH
   
   ls_cmd = remote_unix.get_cmd(cmd_name="ls", cmd_params={"options": "-l"})
   remote_files = ls_cmd()
   
   if 'readme.txt' in remote_files['files']:
       readme_file_info = remote_files['files']['readme.txt']
       for attr in readme_file_info:
           print("  {:<18}: {}".format(attr, readme_file_info[attr]))
   

   readme.txt file:
     permissions       : -rw-------
     hard_links_count  : 1
     owner             : demo
     group             : users
     size_raw          : 403
     size_bytes        : 403
     date              : Apr 08  2014
     name              : readme.txt
   
4. Run a command via direct connection

   If you don’t need a full device configuration, you can create a command object directly and give it a connection to operate on:

   import time
   from moler.cmd.unix.ping import Ping
   from moler.connection_factory import get_connection
   
   host = 'www.google.com'
   terminal = get_connection(io_type='terminal', variant='threaded')  # open a terminal
   with terminal.open():
       ping_cmd = Ping(connection=terminal.moler_connection,
                       destination=host, options="-w 6")
       print(f"Start pinging {host} ...")
       ping_cmd.start()                            # start in background
       print(f"Doing other stuff while pinging {host} ...")
       time.sleep(3)
       ping_stats = ping_cmd.await_done(timeout=4) # wait for result
       print("ping {}: {}={}, {}={} [{}]".format(
           host,
           'packet_loss', ping_stats['packet_loss'],
           'time_avg',    ping_stats['time_avg'],
                          ping_stats['time_unit']
       ))
   

   Start pinging www.google.com ...
   Doing other stuff while pinging www.google.com ...
   ping www.google.com: packet_loss=0, time_avg=50.000 [ms]
   

   The terminal connection is a context manager — it opens and closes the terminal connection automatically. ping_cmd.start() runs the command in the background as a future. await_done(timeout=4) blocks until the result is available or the timeout expires.

Running commands in parallel

Because commands are futures, you can run multiple commands concurrently across different devices. Start one in the background with .start(), do other work in the foreground, then collect the background result with .await_done():

from moler.config import load_config
from moler.device.device import DeviceFactory

load_config(config='my_devices.yml')

my_unix = DeviceFactory.get_device(name='MyMachine')
host = 'www.google.com'
ping_cmd = my_unix.get_cmd(cmd_name="ping", cmd_params={"destination": host, "options": "-w 6"})

remote_unix = DeviceFactory.get_device(name='RebexTestMachine')
remote_unix.goto_state(state="UNIX_REMOTE")
ls_cmd = remote_unix.get_cmd(cmd_name="ls", cmd_params={"options": "-l"})

print(f"Start pinging {host} ...")
ping_cmd.start()                                # run ping in background
print(f"Let's check readme.txt at {remote_unix.name} while pinging {host} ...")

remote_files = ls_cmd()                         # run ls in foreground
file_info = remote_files['files']['readme.txt']
print("readme.txt file: owner={fi[owner]}, size={fi[size_bytes]}".format(fi=file_info))

ping_stats = ping_cmd.await_done(timeout=6)     # collect background result
print("ping {}: {}={}, {}={} [{}]".format(
    host,
    'packet_loss', ping_stats['packet_loss'],
    'time_avg',    ping_stats['time_avg'],
                   ping_stats['time_unit']
))

Start pinging www.google.com ...
Let's check readme.txt at RebexTestMachine while pinging www.google.com ...
readme.txt file: owner=demo, size=403
ping www.google.com: packet_loss=0, time_avg=49.686 [ms]

The ping and the ls run concurrently across two devices. The total wall-clock time is determined by the slower of the two operations, not their sum.

You can also configure Moler entirely from a Python dict instead of a YAML file, which is useful for programmatic or dynamic test setups:

from moler.config import load_config

load_config(config={
    'DEVICES': {
        'MyMachine': {
            'DEVICE_CLASS': 'moler.device.unixlocal.UnixLocal'
        }
    }
})