Modules

Testinfra modules are provided as pytest fixtures, declare them as arguments of your test function to make them available:

def test_foo(Package, File, Command):
    [...]

Note that you can also Using unittest.

Command

class testinfra.modules.command.Command(command, *args)

Run given command and return rc (exit status), stdout and stderr

>>> cmd = Command("ls -l /etc/passwd")
>>> cmd.rc
0
>>> cmd.stdout
'-rw-r--r-- 1 root root 1790 Feb 11 00:28 /etc/passwd\n'
>>> cmd.stderr
''

Good practice: always use shell arguments quoting to avoid shell injection

>>> cmd = Command("ls -l -- %s", "/;echo inject")
CommandResult(
    rc=2, stdout='',
    stderr='ls: cannot access /;echo inject: No such file or directory\n',
    command="ls -l '/;echo inject'")
exists(command)

Return True if given command exist in $PATH

check_output(command, *args, **kwargs)

Get stdout of a command which has run successfully

Returns:stdout without trailing newline
Raises:AssertionError
run_expect(expected, command, *args, **kwargs)

Run command and check it return an expected exit status

Parameters:expected – A list of expected exit status
Raises:AssertionError
run_test(command, *args, **kwargs)

Run command and check it return an exit status of 0 or 1

Raises:AssertionError

LocalCommand

testinfra.plugin.LocalCommand(command, *args)

Run commands locally

Same as Command but run commands locally with subprocess even when the connection backend is not “local”.

Note: LocalCommand does NOT respect --sudo option

TestinfraBackend

class testinfra.backend.base.BaseBackend

Represent the connection to the remote or local system

classmethod get_connection_type()

Return the connection backend used as string.

Can be local, paramiko, ssh, docker, salt or ansible

get_hostname()

Return the hostname (for testinfra) of the remote or local system

Can be useful for multi-hosts tests:

Example:

import requests


def test(TestinfraBackend):
    host = TestinfraBackend.get_hostname()
    response = requests.get("http://" + host)
    assert response.status_code == 200
$ testinfra --hosts=server1,server2 test.py

test.py::test[paramiko://server1] PASSED
test.py::test[paramiko://server2] PASSED
get_module(name)

Return the testinfra module adapted to the current backend

def test(Package):
    [...]

# Is equivalent to
def test(TestinfraBackend):
    Package = TestinfraBackend.get_module("Package")

Sudo

class testinfra.modules.sudo.Sudo(user=None)

Sudo module allow to run certain portion of code under another user.

It is used as a context manager and can be nested.

>>> Command.check_output("whoami")
'phil'
>>> with Sudo():
...     Command.check_output("whoami")
...     with Sudo("www-data"):
...         Command.check_output("whoami")
...
'root'
'www-data'

File

class testinfra.modules.file.File(path)

Test various files attributes

exists

Test if file exists

>>> File("/etc/passwd").exists
True
>>> File("/nonexistent").exists
False
is_file
is_directory
is_pipe
is_socket
linked_to

Resolve symlink

>>> File("/var/lock").linked_to
'/run/lock'
user

Return file owner as string

>>> File("/etc/passwd").user
'root'
uid

Return file user id as integer

>>> File("/etc/passwd").uid
0
group
gid
mode

Return file mode as octal integer

>>> File("/etc/passwd").mode
384  # 0o600 (octal)
>>> File("/etc/password").mode == 0o600
True
>>> oct(File("/etc/password").mode) == '0600'
True

Note: Python 3 oct(x)_ function will produce '0o600'

You can also utilize the file mode constants from the stat library for testing file mode.

>>> import stat
>>> File("/etc/password").mode == stat.S_IRUSR | stat.S_IWUSR
True
contains(pattern)
md5sum
sha256sum
content

Return file content as bytes

>>> File("/tmp/foo").content
b'caf\xc3\xa9'
content_string

Return file content as string

>>> File("/tmp/foo").content_string
'café'
mtime

Return time of last modification as datetime.datetime object

>>> File("/etc/passwd").mtime
datetime.datetime(2015, 3, 15, 20, 25, 40)
size

Return size of file in bytes

User

class testinfra.modules.user.User(name=None)

Test unix users

If name is not supplied, test the current user

name

Return user name

exists
uid

Return user ID

gid

Return effective group ID

group

Return effective group name

gids

Return the list of user group IDs

groups

Return the list of user group names

home

Return the user home directory

shell

Return the user login shell

password

Return the crypted user password

gecos

Return the user comment/gecos field

expiration_date

Return the account expiration date

>>> User("phil").expiration_date
datetime.datetime(2020, 1, 1, 0, 0)
>>> User("root").expiration_date
None
classmethod get_module_class(_backend)

Group

class testinfra.modules.group.Group(name=None)

Test unix group

exists
gid

Interface

class testinfra.modules.interface.Interface(name)

Test network interfaces

exists
speed
addresses

Return ipv4 and ipv6 addresses on the interface

>>> Interface("eth0").addresses
['192.168.31.254', '192.168.31.252', 'fe80::e291:f5ff:fe98:6b8c']

Package

class testinfra.modules.package.Package(name)

Test packages status and version

is_installed

Test if the package is installed

>>> Package("nginx").is_installed
True

Supported package systems:

  • apt (Debian, Ubuntu, ...)
  • rpm (RHEL, Centos, Fedora, ...)
  • pkg_info (OpenBSD)
  • pkg_info (NetBSD)
  • pkg (FreeBSD)
release

Return the release specific info from the package version

>>> Package("nginx").release
'1.el6'
version

Return package version as returned by the package system

>>> Package("nginx").version
'1.2.1-2.2+wheezy3'

PipPackage

class testinfra.modules.pip.PipPackage

Test pip packages status and version

get_packages(pip_path=u'pip')

Get all installed packages and versions returned by pip list:

>>> PipPackage.get_packages(pip_path='~/venv/website/bin/pip')
{'Django': {'version': '1.10.2'},
 'mywebsite': {'version': '1.0a3', 'path': '/srv/website'},
 'psycopg2': {'version': '2.6.2'}}
get_outdated_packages(pip_path=u'pip')

Get all outdated packages with current and latest version

>>> PipPackage.get_outdated_packages(pip_path='~/venv/website/bin/pip')
{'Django': {'current': '1.10.2', 'latest': '1.10.3'}}

Process

class testinfra.modules.process.Process

Test Processes attributes

Processes are selected using filter() or get(), attributes names are described in the ps(1) man page.

>>> master = Process.get(user="root", comm="nginx")
# Here is the master nginx process (running as root)
>>> master.args
'nginx: master process /usr/sbin/nginx -g daemon on; master_process on;'
# Here are the worker processes (Parent PID = master PID)
>>> workers = Process.filter(ppid=master.pid)
>>> len(workers)
4
# Nginx don't eat memory
>>> sum([w.pmem for w in workers])
0.8
# But php does !
>>> sum([p.pmem for p in Process.filter(comm="php5-fpm")])
19.2
filter(**filters)

Get a list of matching process

>>> Process.filter(user="root", comm="zsh")
[<process zsh (pid=2715)>, <process zsh (pid=10502)>, ...]
get(**filters)

Get one matching process

Raise RuntimeError if no process found or multiple process matching filters.

Service

class testinfra.modules.service.Service(name)

Test services

Implementations:

  • Linux: detect Systemd or Upstart, fallback to SysV
  • FreeBSD: service(1)
  • OpenBSD: /etc/rc.d/$name check for is_running (is_enabled is not yet implemented)
  • NetBSD: /etc/rc.d/$name onestatus for is_running (is_enabled is not yet implemented)
is_running

Test if service is running

is_enabled

Test if service is enabled

Supervisor

class testinfra.modules.supervisor.Supervisor(name, _attrs_cache=None)

Test supervisor managed services

>>> gunicorn = Supervisor("gunicorn")
>>> gunicorn.status
'RUNNING'
>>> gunicorn.is_running
True
>>> gunicorn.pid
4242
is_running

Return True if managed service is in status RUNNING

status

Return the status of the managed service

Status can be STOPPED, STARTING, RUNNING, BACKOFF, STOPPING, EXITED, FATAL, UNKNOWN.

See http://supervisord.org/subprocess.html#process-states

pid

Return the pid (as int) of the managed service

classmethod get_services()

Get a list of services running under supervisor

>>> Supervisor.get_services()
[<Supervisor(name="gunicorn", status="RUNNING", pid=4232)>
 <Supervisor(name="celery", status="FATAL", pid=None)>]

Socket

class testinfra.modules.socket.Socket(socketspec)

Test listening tcp/udp and unix sockets

socketspec must be specified as <protocol>://<host>:<port>

This module requires the netstat command to on the target host.

Example:

  • Unix sockets: unix:///var/run/docker.sock
  • All ipv4 and ipv6 tcp sockets on port 22: tcp://22
  • All ipv4 sockets on port 22: tcp://0.0.0.0:22
  • All ipv6 sockets on port 22: tcp://:::22
  • udp socket on 127.0.0.1 port 69: udp://127.0.0.1:69
is_listening

Test if socket is listening

>>> Socket("unix:///var/run/docker.sock").is_listening
False
>>> # This HTTP server listen on all ipv4 adresses but not on ipv6
>>> Socket("tcp://0.0.0.0:80").is_listening
True
>>> Socket("tcp://:::80").is_listening
False
>>> Socket("tcp://80").is_listening
False

Note

If you don’t specify a host for udp and tcp sockets, then the socket is listening if and only if the socket listen on both all ipv4 and ipv6 addresses (ie 0.0.0.0 and ::)

clients

Return a list of clients connected to a listening socket

For tcp and udp sockets a list of pair (adress, port) is returned. For unix sockets a list of None is returned (thus you can make a len() for counting clients).

>>> Socket("tcp://22").clients
[('2001:db8:0:1', 44298), ('192.168.31.254', 34866)]
>>> Socket("unix:///var/run/docker.sock")
[None, None, None]
classmethod get_listening_sockets()

Return a list of all listening sockets

>>> Socket.get_listening_sockets()
['tcp://0.0.0.0:22', 'tcp://:::22', 'unix:///run/systemd/private', ...]

SystemInfo

class testinfra.modules.systeminfo.SystemInfo

Return system informations

type

OS type

>>> SystemInfo.type
'linux'
distribution

Distribution name

>>> SystemInfo.distribution
'debian'
release

Distribution release number

>>> SystemInfo.release
'7.8'
codename

Release code name

>>> SystemInfo.codename
'wheezy'

Salt

class testinfra.modules.salt.Salt(function, args=None)

Run salt module functions

>>> Salt("pkg.version", "nginx")
'1.6.2-5'
>>> Salt("pkg.version", ["nginx", "php5-fpm"])
{'nginx': '1.6.2-5', 'php5-fpm': '5.6.7+dfsg-1'}
>>> Salt("grains.item", ["osarch", "mem_total", "num_cpus"])
{'osarch': 'amd64', 'num_cpus': 4, 'mem_total': 15520}

Run salt-call sys.doc to get a complete list of functions

Ansible

class testinfra.modules.ansible.Ansible(module_name, module_args=None, check=True)

Run Ansible module functions

This module is only available with the ansible connection backend.

Check mode is enabled by default, you can disable it with check=False.

>>> Ansible("apt", "name=nginx state=present")["changed"]
False
>>> Ansible("command", "echo foo", check=False)["stdout"]
'foo'
>>> Ansible("setup")["ansible_facts"]["ansible_lsb"]["codename"]
'jessie'
>>> Ansible("file", "path=/etc/passwd")["mode"]
'0640'
exception AnsibleException(result)

Exception raised when an error occur in an ansible call

result from ansible can be accessed through the result attribute

>>> try:
...     Ansible("command", "echo foo")
... except Ansible.AnsibleException as exc:
...     assert exc.result['failed'] is True
...     assert exc.result['msg'] == 'check mode not supported for command'
Ansible.get_variables()

Returns a dict of ansible variables

>>> Ansible.get_variables()
{
    'inventory_hostname': 'localhost',
    'group_names': ['ungrouped'],
    'foo': 'bar',
}

PuppetResource

class testinfra.modules.puppet.PuppetResource(type, name=None)

Get puppet resources

Run puppet resource --types to get a list of available types.

>>> PuppetResource("user", "www-data")
{
    'www-data': {
        'ensure': 'present',
        'comment': 'www-data',
        'gid': '33',
        'home': '/var/www',
        'shell': '/usr/sbin/nologin',
        'uid': '33',
    },
}

Facter

class testinfra.modules.puppet.Facter(*facts)

Get facts with facter

>>> Facter()
{
    "operatingsystem": "Debian",
    "kernel": "linux",
    [...]
}
>>> Facter("kernelversion", "is_virtual")
{
  "kernelversion": "3.16.0",
  "is_virtual": "false"
}

Sysctl

class testinfra.modules.sysctl.Sysctl(name)

Test kernel parameters

>>> Sysctl("kernel.osrelease")
"3.16.0-4-amd64"
>>> Sysctl("vm.dirty_ratio")
20

MountPoint

class testinfra.modules.mountpoint.MountPoint(path)

Test Mount Points

exists

Return True if the mountpoint exists

>>> MountPoint("/").exists
True
>>> MountPoint("/not/a/mountpoint").exists
False
filesystem

Returns the filesystem type associated

>>> MountPoint("/").filesystem
'ext4'
device

Return the device associated

>>> MountPoint("/").device
'/dev/sda1'
options

Return a list of options that a mount point has been created with

>>> MountPoint("/").options
['rw', 'relatime', 'data=ordered']
classmethod get_mountpoints()

Returns a list of MountPoint instances

>>> MountPoint.get_mountpoints()
[<MountPoint(path=/proc, device=proc, filesystem=proc, options=rw,nosuid,nodev,noexec,relatime)>
 <MountPoint(path=/, device=/dev/sda1, filesystem=ext4, options=rw,relatime,errors=remount-ro,data=ordered)>]