Devices

Devices are state machines that act as factories for commands and events. They manage connection lifecycle, automatic state transitions, and per-device logging.

A Device in Moler serves two roles simultaneously:

1. State machine — tracks which shell/environment the connection is currently in (e.g. local shell, remote SSH, root prompt) and can automatically navigate between states.
2. Factory — produces Command and Event objects pre-configured for the connection and current state.

This combination means client code can say goto_state("UNIX_REMOTE") and get_cmd("ls") without knowing anything about how SSH is invoked or which prompt regex to match.

Getting a device

Devices are retrieved via DeviceFactory.get_device(). The first call creates the device and opens its connection; subsequent calls with the same name return the cached instance.

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

load_config(config='my_devices.yml')
my_unix = DeviceFactory.get_device(name='MyMachine')

You can also create a device by class without a config file:

from moler.config import load_config

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

YAML configuration

Device definitions live in a YAML config file. The DEVICE_CLASS key names the Python class; CONNECTION_HOPS describes how to move between states.

DEVICES:

  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   # command used for transition
          command_params:
            expected_prompt: demo@
            host: test.rebex.net
            login: demo
            password: password
            set_timeout: None

Only the config file knows how to reach a remote state. Client code just calls goto_state("UNIX_REMOTE"). You can change credentials or the transition command without touching test logic.

States and transitions

Every device starts life in the NOT_CONNECTED state. When a connection is opened the device moves to its initial state (e.g. UNIX_LOCAL). Call goto_state() to move to any reachable state — the device works out the intermediate hops automatically.

remote_unix = DeviceFactory.get_device(name='RebexTestMachine')
# Device starts in UNIX_LOCAL after connection is opened
remote_unix.goto_state(state="UNIX_REMOTE")  # runs ssh automatically

Standard states

State	Description
NOT_CONNECTED	No connection open. All devices start here.
UNIX_LOCAL	Local shell on the machine running the test.
UNIX_LOCAL_ROOT	Root shell on the local machine (via su).
UNIX_REMOTE	Remote Unix shell (reached via SSH or telnet).
PROXY_PC	Intermediate jump host.

Device-specific states (e.g. ADB_SHELL, AT_REMOTE, SCPI) are defined by their respective device classes.

goto_state() parameters

device.goto_state(
    state="UNIX_REMOTE",
    timeout=60,                            # seconds before failure
    rerun=0,                               # retry attempts
    send_enter_after_changed_state=False,
    keep_state=False,                      # re-enter state if unexpectedly lost
    sleep_after_changed_state=0.5,         # pause after transition
)

Getting commands and events

Use get_cmd() and get_event() to obtain objects bound to the device’s connection and current state.

# get_cmd returns a ready-to-run Command object
ps_cmd = my_unix.get_cmd(cmd_name="ps", cmd_params={"options": "-ef"})
result = ps_cmd()  # run it

# get_event returns a ready-to-start Event object
proc_event = my_unix.get_event(
    event_name="wait4prompt",
    event_params={"prompt": r"^moler_bash#"},
)
proc_event.start()

cmd_name is the class name (without package), e.g. "ps", "ls", "ssh", "ping". Commands are resolved from the packages associated with the current state.

You can also use the shorthand run() which creates, runs, and returns the result in one call:

result = my_unix.run("ls", cmd_params={"options": "-l"})

Available device types

Class	Module	Description
UnixLocal	moler.device.unixlocal	Local Unix shell
UnixRemote	moler.device.unixremote	Remote Unix shell (via SSH)
AdbRemote	moler.device.adbremote	Android Debug Bridge shell
JuniperGeneric	moler.device.junipergeneric	Juniper network device
AtRemote	moler.device.atremote	AT command interface
Scpi	moler.device.scpi	SCPI instrument interface
PduAten	moler.device.pdu_aten	ATEN PDU power device
ProxyPc	moler.device.proxy_pc	Jump-host / proxy PC

Logging

Each device automatically creates two log files when the LOGGER section is present in the config:

• moler.log — high-level activity across all devices
• moler.<DeviceName>.log — all raw I/O and command activity for that device

LOGGER:
  PATH: ./logs
  DATE_FORMAT: "%H:%M:%S"

You can control logging per device at runtime:

my_unix.disable_logging()   # stop recording raw I/O
my_unix.enable_logging()    # resume recording

my_unix.set_logging_suffix("_run1")   # append suffix to log filename
my_unix.set_logging_suffix(None)      # remove suffix

Log rotation is supported by size or time:

LOGGER:
  PATH: ./logs
  LOG_ROTATION:
    KIND: size        # or 'time'
    INTERVAL: 5242880 # bytes (or seconds for time-based)
    BACKUP_COUNT: 999
    COMPRESS_AFTER_ROTATION: True

Use disable_logging() with care. Missing device logs can make it impossible to diagnose test failures after the fact.

Removing devices

When a device is no longer needed, remove it to close the connection and finish any running commands or events:

DeviceFactory.remove_device(name='MyMachine')
# or remove all at once:
DeviceFactory.remove_all_devices()

Commands

What get_cmd() returns and how to run it

Connections

The underlying IO layer devices use