Modules

Testinfra modules are provided through the host fixture, declare it as arguments of your test function to make it available within it.

def test_foo(host):
    # [...]

host

class Host(backend)
ansible

testinfra.modules.ansible.Ansible class

file

testinfra.modules.file.File class

group

testinfra.modules.group.Group class

interface

testinfra.modules.interface.Interface class

mount_point

testinfra.modules.mountpoint.MountPoint class

package

testinfra.modules.package.Package class

pip_package

testinfra.modules.pip.PipPackage class

process

testinfra.modules.process.Process class

puppet_resource

testinfra.modules.puppet.PuppetResource class

facter

testinfra.modules.puppet.Facter class

salt

testinfra.modules.salt.Salt class

service

testinfra.modules.service.Service class

socket

testinfra.modules.socket.Socket class

sudo

testinfra.modules.sudo.Sudo class

supervisor

testinfra.modules.supervisor.Supervisor class

sysctl

testinfra.modules.sysctl.Sysctl class

system_info

testinfra.modules.systeminfo.SystemInfo class

user

testinfra.modules.user.User class

check_output(command, *args, **kwargs)

Get stdout of a command which has run successfully

Returns:stdout without trailing newline
Raises:AssertionError
exists(command)

Return True if given command exist in $PATH

classmethod get_host(hostspec, **kwargs)

Return a Host instance from hostspec

hostspec should be like <backend_type>://<name>?param1=value1&param2=value2

Params can also be passed in **kwargs (eg. get_host(“local://”, sudo=True) is equivalent to get_host(“local://?sudo=true”))

Examples:

>>> get_host("local://", sudo=True)
>>> get_host("paramiko://user@host", ssh_config="/path/my_ssh_config")
>>> get_host("ansible://all?ansible_inventory=/etc/ansible/inventory")
run(command, *args, **kwargs)

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

>>> cmd = host.run("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 = host.run("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'")
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

Ansible

class 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.

>>> host.ansible("apt", "name=nginx state=present")["changed"]
False
>>> host.ansible("command", "echo foo", check=False)["stdout"]
'foo'
>>> host.ansible("setup")["ansible_facts"]["ansible_lsb"]["codename"]
'jessie'
>>> host.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:
...     host.ansible("command", "echo foo")
... except host.ansible.AnsibleException as exc:
...     assert exc.result['failed'] is True
...     assert exc.result['msg'] == 'check mode not supported for command'
get_variables()

Returns a dict of ansible variables

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

File

class File(path)

Test various files attributes

exists

Test if file exists

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

Resolve symlink

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

Return file owner as string

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

Return file user id as integer

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

Return file mode as octal integer

>>> host.file("/etc/passwd").mode
384  # 0o600 (octal)
>>> host.file("/etc/password").mode == 0o600
True
>>> oct(host.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
>>> host.file("/etc/password").mode == stat.S_IRUSR | stat.S_IWUSR
True
contains(pattern)
md5sum
sha256sum
content

Return file content as bytes

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

Return file content as string

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

Return time of last modification as datetime.datetime object

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

Return size of file in bytes

Group

class Group(name=None)

Test unix group

exists
gid

Interface

class Interface(name)

Test network interfaces

exists
speed
addresses

Return ipv4 and ipv6 addresses on the interface

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

MountPoint

class MountPoint(path)

Test Mount Points

exists

Return True if the mountpoint exists

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

Returns the filesystem type associated

>>> host.mount_point("/").filesystem
'ext4'
device

Return the device associated

>>> host.mount_point("/").device
'/dev/sda1'
options

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

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

Returns a list of MountPoint instances

>>> host.mount_point.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)>]

Package

class Package(name)

Test packages status and version

is_installed

Test if the package is installed

>>> host.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

>>> host.package("nginx").release
'1.el6'
version

Return package version as returned by the package system

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

PipPackage

class PipPackage

Test pip packages status and version

get_packages(pip_path=u'pip')

Get all installed packages and versions returned by pip list:

>>> host.pip_package.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

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

Process

class Process

Test Processes attributes

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

>>> master = host.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 = host.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 host.process.filter(comm="php5-fpm")])
19.2
filter(**filters)

Get a list of matching process

>>> host.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.

PuppetResource

class PuppetResource(type, name=None)

Get puppet resources

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

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

Facter

class Facter(*facts)

Get facts with facter

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

Salt

class Salt(function, args=None, local=False, config=None)

Run salt module functions

>>> host.salt("pkg.version", "nginx")
'1.6.2-5'
>>> host.salt("pkg.version", ["nginx", "php5-fpm"])
{'nginx': '1.6.2-5', 'php5-fpm': '5.6.7+dfsg-1'}
>>> host.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

Service

class 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 rcctl ls on for is_enabled (only OpenBSD >= 5.8)
  • 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

Socket

class 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

>>> host.socket("unix:///var/run/docker.sock").is_listening
False
>>> # This HTTP server listen on all ipv4 adresses but not on ipv6
>>> host.socket("tcp://0.0.0.0:80").is_listening
True
>>> host.socket("tcp://:::80").is_listening
False
>>> host.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).

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

Return a list of all listening sockets

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

Sudo

class 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 host.sudo():
...     host.check_output("whoami")
...     with host.sudo("www-data"):
...         host.check_output("whoami")
...
'root'
'www-data'

Supervisor

class Supervisor(name, _attrs_cache=None)

Test supervisor managed services

>>> gunicorn = host.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

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

Sysctl

class Sysctl(name)

Test kernel parameters

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

SystemInfo

class SystemInfo

Return system informations

type

OS type

>>> host.system_info.type
'linux'
distribution

Distribution name

>>> host.system_info.distribution
'debian'
release

Distribution release number

>>> host.system_info.release
'7.8'
codename

Release code name

>>> host.system_info.codename
'wheezy'

User

class User(name=None)

Test unix users

If name is not supplied, test the current user

name

Return user name

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

>>> host.user("phil").expiration_date
datetime.datetime(2020, 1, 1, 0, 0)
>>> host.user("root").expiration_date
None