Documentation Index
Fetch the complete documentation index at: https://mintlify.com/nokia/moler/llms.txt
Use this file to discover all available pages before exploring further.
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).
Install Moler
Install the latest stable release from PyPI:Moler requires Python 3.8 or later and runs on POSIX/Unix systems. See Installation for full details. 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
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.
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
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
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]
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]
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'
}
}
})