trigger.utils — CLI tools and utilities library

A collection of CLI tools and utilities used by Trigger.

class trigger.utils.JuniperElement(key, value)

Alias for field number 0


Alias for field number 1


alias of trigger.utils.HostPort


Returns an md5-crypt hash of a clear-text password.

To get md5-crypt from crypt(3) you must pass an 8-char string starting with ‘$1$’ and ending with ‘$’, resulting in a 12-char salt. This only works on systems where md5-crypt is default and is currently assumed to be Linux.

Parameters:passwd – Password string to be encrypted
trigger.utils.parse_node_port(nodeport, delimiter=':')

Parse a string in format ‘hostname’ or ‘hostname:port’ and return them as a 2-tuple.

trigger.utils.strip_juniper_namespace(path, key, value)

Given a Juniper XML element, strip the namespace and return a 2-tuple.

This is designed to be used as a postprocessor with parse().

  • key – The attribute name of the element.
  • value – The value of the element.


Command-line interface utilities for Trigger tools. Intended for re-usable pieces of code like user prompts, that don’t fit in other utils modules.

trigger.utils.cli.yesno(prompt, default=False, autoyes=False)

Present a yes-or-no prompt, get input, and return a boolean.

The default argument is ignored if autoyes is set.

  • prompt – Prompt text
  • default – Yes if True; No if False
  • autoyes – Automatically return True

Default behavior (hitting “enter” returns False):

>>> yesno('Blow up the moon?')
Blow up the moon? (y/N)

Reversed behavior (hitting “enter” returns True):

>>> yesno('Blow up the moon?', default=True)
Blow up the moon? (Y/n)

Automatically return True with autoyes; no prompt is displayed:

>>> yesno('Blow up the moon?', autoyes=True)

Find and return stdout’s terminal width, if applicable.


Find and return stdouts terminal size as (height, width)

class trigger.utils.cli.Whirlygig(start_msg='', done_msg='', max=100)

Prints a whirlygig for use in displaying pending operation in a command-line tool. Guaranteed to make the user feel warm and fuzzy and be 1000% bug-free.

  • start_msg – The status message displayed to the user (e.g. “Doing stuff:”)
  • done_msg – The completion message displayed upon completion (e.g. “Done.”)
  • max – Integer of the number of whirlygig repetitions to perform


>>> Whirlygig("Doing stuff:", "Done.", 12).run()

Executes the whirlygig!

class trigger.utils.cli.NullDevice

Used to supress output to sys.stdout (aka print).


>>> from trigger.utils.cli import NullDevice
>>> import sys
>>> print "1 - this will print to STDOUT"
1 - this will print to STDOUT
>>> original_stdout = sys.stdout  # keep a reference to STDOUT
>>> sys.stdout = NullDevice()     # redirect the real STDOUT
>>> print "2 - this won't print"
>>> sys.stdout = original_stdout  # turn STDOUT back on
>>> print "3 - this will print to SDTDOUT"
3 - this will print to SDTDOUT

Prints a demon holding a severed head. Best used when things go wrong, like production-impacting network outages caused by fat-fingered ACL changes.

Thanks to Jeff Sullivan for this best error message ever.


Takes an epoch timestamp and returns string of minutes:seconds.

Parameters:secs – Timestamp (in seconds)
>>> import time
>>> start = time.time()  # Wait a few seconds
>>> finish = time.time()
>>> min_sec(finish - start)

Print a pretty version of timestamp, including timezone info. Expects the incoming datetime object to have proper tzinfo.

Parameters:t – A datetime.datetime object
>>> import datetime
>>> from pytz import timezone
>>> localzone = timezone('US/Eastern')
<DstTzInfo 'US/Eastern' EST-1 day, 19:00:00 STD>
>>> t =
>>> print t
2011-07-19 12:40:30.820920-04:00
>>> print pretty_time(t)
09:40 PDT
>>> t = localzone.localize(datetime.datetime(2011,07,20,04,13))
>>> print t
2011-07-20 04:13:00-05:00
>>> print pretty_time(t)
tomorrow 02:13 PDT

Present a proceed prompt. Return True if Y, else False


Return the name of the current user.


Utils to import modules.

Taken verbatim from django.utils.importlib in Django 1.4.

trigger.utils.importlib.import_module(name, package=None)

Import a module and return the module object.

The package argument is required when performing a relative import. It specifies the package to use as the anchor point from which to resolve the relative import to an absolute import.

trigger.utils.importlib.import_module_from_path(full_path, global_name)

Import a module from a file path and return the module object.

Allows one to import from anywhere, something __import__() does not do. The module is added to sys.modules as global_name.

  • full_path – The absolute path to the module .py file
  • global_name – The name assigned to the module in sys.modules. To avoid confusion, the global_name should be the same as the variable to which you’re assigning the returned module.

Functions that perform network-based things like ping, port tests, etc., count=1, timeout=5)

Returns pass/fail for a ping. Supports POSIX only.

  • host – Hostname or address
  • count – Repeat count
  • timeout – Timeout in seconds
>>> from trigger.utils import network
False, port=23, timeout=5, check_result=False, expected_result='')

Attempts to connect to a TCP port. Returns a Boolean.

If check_result is set, the first line of output is retreived from the connection and the starting characters must match expected_result.

  • host – Hostname or address
  • port – Destination port
  • timeout – Timeout in seconds
  • check_result – Whether or not to do a string check (e.g. version banner)
  • expected_result – The expected result!
>>> test_tcp_port('', 80)
>>> test_tcp_port('', 12345)
False, port=22, timeout=5, version=('SSH-1.99', 'SSH-2.0', 'dcos_sshd run in non-FIPS mode'))

Connect to a TCP port and confirm the SSH version. Defaults to SSHv2.

Note that the default of (‘SSH-1.99’, ‘SSH-2.0’) both indicate SSHv2 per RFC 4253. (Ref:

  • host – Hostname or address
  • port – Destination port
  • timeout – Timeout in seconds
  • version – The SSH version prefix (e.g. “SSH-2.0”). This may also be a tuple of prefixes.
>>> test_ssh('localhost')
>>> test_ssh('localhost', version='SSH-1.5')

Determines if an IP address is internal to your network. Relies on networks specified in settings.INTERNAL_NETWORKS.

Parameters:ip – IP address to test.
>>> address_is_internal('')


Pluggable event notification system for Trigger.

trigger.utils.notifications.send_email(addresses, subject, body, sender, mailhost='localhost')

Sends an email to a list of recipients. Returns True when done.

  • addresses – List of email recipients
  • subject – The email subject
  • body – The email body
  • sender – The email sender
  • mailhost – (Optional) Mail server address
trigger.utils.notifications.send_notification(*args, **kwargs)

Simple entry point into notify that takes any arguments and tries to handle them to send a notification.

This relies on handlers to be definied within settings.NOTIFICATION_HANDLERS.

trigger.utils.notifications.email_handler(*args, **kwargs)

Default email notification handler.

trigger.utils.notifications.notify(*args, **kwargs)

Iterate thru registered handlers to handle events and send notifications.

Handlers should return True if they have performed the desired action or None if they have not.


Handlers for event notifications.

Handlers are specified by full module path within settings.NOTIFICATION_HANDLERS. These are then imported and registered internally in this module.

The primary public interface to this module is notify which is in turn called by send_notification to send notifications.

Handlers should return True if they have performed the desired action or None if they have not.

A handler can either define its own custom behavior, or leverage a custom Event object. The goal was to provide a simple public interface to customizing event notifications.

If not customized within NOTIFICATION_HANDLERS, the default notification type is an EmailEvent that is handled by email_handler.

trigger.utils.notifications.handlers.email_handler(*args, **kwargs)

Default email notification handler.

trigger.utils.notifications.handlers.notify(*args, **kwargs)

Iterate thru registered handlers to handle events and send notifications.

Handlers should return True if they have performed the desired action or None if they have not.

Event objects for the notification system.

These are intended to be used within event handlers such as email_handler().

If not customized within NOTIFICATION_HANDLERS, the default notification type is an EmailEvent that is handled by email_handler.


Base class for events.

It just populates the attribute dict with all keyword arguments thrown at the constructor.

All Event objects are expected to have a .handle() method that willl be called by a handler function. Any user-defined event objects must have a working .handle() method that returns True upon success or None upon a failure when handling the event passed to it.

If you specify required_args, these must have a value other than None when passed to the constructor.

class, message=None, sender=None, recipients=None, event_status='failure', **kwargs)

Base class for notification events.

The title and message arguments are the only two that are required. This is to simplify the interface when sending notifications and will cause notifications to send from the default sender to the default ``recipients that are specified withing the global settings.

If sender or recipients are specified, they will override the global defaults.

Note that this base class has no .handle() method defined.

  • title – The title/subject of the notification
  • message – The message/body of the notification
  • sender – A string representing the sender of the notification (such as an email address or a hostname)
  • recipients – An iterable containing strings representing the recipients of of the notification (such as a list of emails or hostnames)
  • event_status – Whether this event is a failure or a success
class, message=None, sender=None, recipients=None, event_status='failure', **kwargs)

An email notification event.


Provides a CVS like wrapper for local RCS (Revision Control System) with common commands.

class trigger.utils.rcs.RCS(filename, create=True)

Simple wrapper for CLI rcs command. An instance is bound to a file.

  • file – The filename (or path) to use
  • create – If set, create the file if it doesn’t exist
>>> from trigger.utils.rcs import RCS
>>> rcs = RCS('foo')
>>> rcs.lock()
>>> f = open('foo', 'w')
>>> f.write('bar\n')
>>> f.close()
>>> rcs.checkin('This is my commit message')
>>> print rcs.log()
RCS file: RCS/foo,v
Working file: foo
head: 1.2
locks: strict
access list:
symbolic names:
keyword substitution: kv
total revisions: 2;     selected revisions: 2
revision 1.2
date: 2011/07/08 21:01:28;  author: jathan;  state: Exp;  lines: +1 -0
This is my commit message
revision 1.1
date: 2011/07/08 20:56:53;  author: jathan;  state: Exp;
first commit
checkin(logmsg='none', initial=False, verbose=False)

Perform an RCS checkin. If successful this also unlocks the file, so there is no need to unlock it afterward.

  • logmsg – The RCS commit message
  • initial – Initialize a new RCS file, but do not deposit any revision
  • verbose – Print command output
>>> rcs.checkin('This is my commit message')

Perform an RCS checkout with lock. Returns boolean of whether lock was sucessful.

Parameters:verbose – Print command output
>>> rcs.lock()
lock_loop(callback=None, timeout=5, verbose=False)

Keep trying to lock the file until a lock is obtained.

  • callback – The function to call after lock is complete
  • timeout – How long to sleep between lock attempts
  • verbose – Print command output
>>> rcs.lock_loop(timeout=1) 
Sleeping to wait for the lock on the file: foo
Sleeping to wait for the lock on the file: foo
>>> rcs.lock_loop(timeout=1, verbose=True)
RCS/foo,v  -->  foo
co: RCS/foo,v: Revision 1.2 is already locked by joe.
Sleeping to wait for the lock on the file: foo
RCS/foo,v  -->  foo
co: RCS/foo,v: Revision 1.2 is already locked by joe.

Returns the RCS log as a string (see above).


Perform an RCS checkout with unlock (for cancelling changes).

Parameters:verbose – Print command output
>>> rcs.unlock()