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

addr¶

testinfra.modules.addr.Addr class

blockdevice¶

testinfra.modules.blockdevice.BlockDevice class

docker¶

testinfra.modules.docker.Docker class

environment¶

testinfra.modules.environment.Environment class

file¶

testinfra.modules.file.File class

group¶

testinfra.modules.group.Group class

interface¶

testinfra.modules.interface.Interface class

iptables¶

testinfra.modules.iptables.Iptables class

mount_point¶

testinfra.modules.mountpoint.MountPoint class

package¶

testinfra.modules.package.Package class

pip¶

testinfra.modules.pip.Pip class

pip_package¶

testinfra.modules.pip.PipPackage class

podman¶

testinfra.modules.podman.Podman 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

has_command_v¶: Return True if command -v is available

exists(command)¶: Return True if given command exist in $PATH

find_command(command, extrapaths=('/sbin', '/usr/sbin'))¶

Return path of given command

raise ValueError if command cannot be found

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
''
>>> cmd.succeeded
True
>>> cmd.failed
False

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

check_output(command, *args, **kwargs)¶

Get stdout of a command which has run successfully

Returns:	stdout without trailing newline
Raises:	AssertionError

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")

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.

Become is False by default. You can enable it with become=True.

Ansible arguments that are not related to the Ansible inventory or connection (both managed by testinfra) are also accepted through keyword arguments:

become_method str sudo, su, doas, etc.

become_user str become this user.

diff bool: when changing (small) files and templates, show the differences in those files.

extra_vars dict serialized to a JSON string, passed to Ansible.

one_line bool: condense output.

user str connect as this user.

verbose int level of verbosity

>>> host.ansible("apt", "name=nginx state=present")["changed"]
False
>>> host.ansible("apt", "name=nginx state=present", become=True)["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'
>>> host.ansible(
... "command",
... "id --user --name",
... check=False,
... become=True,
... become_user="http",
... )["stdout"]
'http'
>>> host.ansible(
... "apt",
... "name={{ packages }}",
... check=False,
... extra_vars={"packages": ["neovim", "vim"]},
... )
# Installs neovim and vim.

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'] == 'Skipped. You might want to try check=False'  # noqa

get_variables()¶

Returns a dict of ansible variables

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

Addr¶

class Addr(name)¶

Test remote address

Example:

>>> google = host.addr("google.com")
>>> google.is_resolvable
True
>>> '173.194.32.225' in google.ipv4_addresses
True
>>> google.is_reachable
True
>>> google.port(443).is_reachable
True
>>> google.port(666).is_reachable
False

Can also be use within a network namespace.

>>> localhost = host.addr("localhost", "ns1")
>>> localhost.is_resolvable
True

Network namespaces can only be used if ip command is available because in this case, the module use ip-netns as command prefix. In the other case, it will raise NotImplementedError.

name¶: Return host name

namespace¶: Return network namespace

namespace_exists¶: Test if the network namespace exists

is_resolvable¶: Return if address is resolvable

is_reachable¶: Return if address is reachable

ip_addresses¶: Return IP addresses of host

ipv4_addresses¶: Return IPv4 addresses of host

ipv6_addresses¶: Return IPv6 addresses of host

port(port)¶: Return address-port pair

BlockDevice¶

class BlockDevice(name)¶

Information for block device.

Should be used with sudo or under root.

If device is not a block device, RuntimeError is raised.

is_partition¶

Return True if the device is a partition.

>>> host.block_device("/dev/sda1").is_partition
True

>>> host.block_device("/dev/sda").is_partition
False

size¶

Return size if the device in bytes.

>>> host.block_device("/dev/sda1").size
512110190592

sector_size¶

Return sector size for the device in bytes.

>>> host.block_device("/dev/sda1").sector_size
512

block_size¶

Return block size for the device in bytes.

>>> host.block_device("/dev/sda").block_size
4096

start_sector¶

Return start sector of the device on the underlying device.

Usually the value is zero for full devices and is non-zero for partitions.

>>> host.block_device("/dev/sda1").start_sector
2048

>>> host.block_device("/dev/md0").start_sector
0

is_writable¶

Return True if device is writable (have no RO status)

>>> host.block_device("/dev/sda").is_writable
True

>>> host.block_device("/dev/loop1").is_writable
False

ra¶

Return Read Ahead for the device in 512-bytes sectors.

>>> host.block_device("/dev/sda").ra
256

Docker¶

class Docker(name)¶

Test docker containers running on system.

Example:

>>> nginx = host.docker("app_nginx")
>>> nginx.is_running
True
>>> nginx.id
'7e67dc7495ca8f451d346b775890bdc0fb561ecdc97b68fb59ff2f77b509a8fe'
>>> nginx.name
'app_nginx'

classmethod client_version()¶: Docker client version

classmethod server_version()¶: Docker server version

classmethod version(format=None)¶

Docker version with an optional format (Go template).

>>> host.docker.version()
Client: Docker Engine - Community
...
>>> host.docker.version("{{.Client.Context}}"))
default

classmethod get_containers(**filters)¶

Return a list of containers

By default return list of all containers, including non-running containers.

Filtering can be done using filters keys defined on https://docs.docker.com/engine/reference/commandline/ps/#filtering

Multiple filters for a given key is handled by giving a list of string as value.

>>> host.docker.get_containers()
[<docker nginx>, <docker redis>, <docker app>]
# Get all running containers
>>> host.docker.get_containers(status="running")
[<docker app>]
# Get containers named "nginx"
>>> host.docker.get_containers(name="nginx")
[<docker nginx>]
# Get containers named "nginx" or "redis"
>>> host.docker.get_containers(name=["nginx", "redis"])
[<docker nginx>, <docker redis>]

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¶: Test if the path is a regular file

is_directory¶: Test if the path exists and a directory

is_executable¶: Test if the path exists and permission to execute is granted

is_pipe¶: Test if the path exists and is a pipe

is_socket¶: Test if the path exists and is a socket

is_symlink¶: Test if the path exists and is a symbolic link

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¶: Return file group name as string

gid¶: Return file group id as integer

mode¶

Return file mode as octal integer

>>> host.file("/etc/shadow").mode
416  # Oo640 octal
>>> host.file("/etc/shadow").mode == 0o640
True
>>> oct(host.file("/etc/shadow").mode) == '0o640'
True

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

>>> import stat
>>> host.file("/etc/shadow").mode == stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP
True

contains(pattern)¶

Checks content of file for pattern

This uses grep and thus follows the grep regex syntax.

md5sum¶: Compute the MD5 message digest of the file content

sha256sum¶: Compute the SHA256 message digest of the file content

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

listdir()¶

Return list of items under the directory

>>> host.file("/tmp").listdir()
['foo_file', 'bar_dir']

Group¶

class Group(name=None)¶

Test unix group

exists¶

Test if group exists

>>> host.group("wheel").exists
True
>>> host.group("nosuchgroup").exists
False

gid¶

members¶: Return all users that are members of this group.

Interface¶

class Interface(name, family=None)¶

Test network interfaces

>>> host.interface("eth0").exists
True

Optionally, the protocol family to use can be enforced.

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

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']

routes(scope=None)¶

Return the routes associated with the interface, optionally filtered by scope (“host”, “link” or “global”).

>>> host.interface("eth0").routes()
[{'dst': 'default',
'flags': [],
'gateway': '192.0.2.1',
'metric': 3003,
'prefsrc': '192.0.2.5',
'protocol': 'dhcp'},
{'dst': '192.0.2.0/24',
'flags': [],
'metric': 3003,
'prefsrc': '192.0.2.5',
'protocol': 'dhcp',
'scope': 'link'}]

classmethod names()¶

Return the names of all the interfaces.

>>> host.interface.names()
['lo', 'tunl0', 'ip6tnl0', 'eth0']

classmethod default(family=None)¶

Return the interface used for the default route.

>>> host.interface.default()
<interface eth0>

Optionally, the protocol family to use can be enforced.

>>> host.interface.default("inet6")
None

Iptables¶

class Iptables¶

Test iptables rule exists

rules(table='filter', chain=None, version=4)¶

Returns list of iptables rules

Based on output of iptables -t TABLE -S CHAIN command

optionally takes takes the following arguments:

table: defaults to filter

chain: defaults to all chains

version: default 4 (iptables), optionally 6 (ip6tables)

>>> host.iptables.rules()
[
    '-P INPUT ACCEPT',
    '-P FORWARD ACCEPT',
    '-P OUTPUT ACCEPT',
    '-A INPUT -i lo -j ACCEPT',
    '-A INPUT -j REJECT'
    '-A FORWARD -j REJECT'
]
>>> host.iptables.rules("nat", "INPUT")
['-P PREROUTING ACCEPT']

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:

apk (Alpine)
apt (Debian, Ubuntu, …)
pacman (Arch, Manjaro )
pkg (FreeBSD)
pkg_info (NetBSD)
pkg_info (OpenBSD)
rpm (RHEL, RockyLinux, Fedora, …)

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'

Pip¶

class Pip(name, pip_path='pip')¶

Test pip package manager and packages

is_installed¶

Test if the package is installed

>>> host.package("pip").is_installed
True

version¶

Return package version as returned by pip

>>> host.package("pip").version
'18.1'

classmethod check(pip_path='pip')¶

Verify installed packages have compatible dependencies.

>>> cmd = host.pip_package.check()
>>> cmd.rc
0
>>> cmd.stdout
No broken requirements found.

Can only be used if pip check command is available, for pip versions >= 9.0.0.

classmethod get_packages(pip_path='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'}}

classmethod get_outdated_packages(pip_path='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'}}

PipPackage¶

class PipPackage(name, pip_path='pip')¶: Deprecated since version 6.2.

Use Pip instead.

Podman¶

class Podman(name)¶

Test podman containers running on system.

Example:

>>> nginx = host.podman("app_nginx")
>>> nginx.is_running
True
>>> nginx.id
'7e67dc7495ca8f451d346b775890bdc0fb561ecdc97b68fb59ff2f77b509a8fe'
>>> nginx.name
'app_nginx'

classmethod get_containers(**filters)¶

Return a list of containers

By default return list of all containers, including non-running containers.

Filtering can be done using filters keys defined in podman-ps(1).

Multiple filters for a given key is handled by giving a list of string as value.

>>> host.podman.get_containers()
[<podman nginx>, <podman redis>, <podman app>]
# Get all running containers
>>> host.podman.get_containers(status="running")
[<podman app>]
# Get containers named "nginx"
>>> host.podman.get_containers(name="nginx")
[<podman nginx>]
# Get containers named "nginx" or "redis"
>>> host.podman.get_containers(name=["nginx", "redis"])
[<podman nginx>, <podman redis>]

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, Upstart or OpenRC, 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

is_valid¶

Test if service is valid

This method is only available in the systemd implementation, it will raise NotImplementedError in others implementation

is_masked¶

Test if service is masked

This method is only available in the systemd implementation, it will raise NotImplementedError in others implementations

systemd_properties¶

Properties of the service (unit).

Return service properties as a dict, empty properties are not returned.

>>> ntp = host.service("ntp")
>>> ntp.systemd_properties["FragmentPath"]
'/lib/systemd/system/ntp.service'

This method is only available in the systemd implementation, it will raise NotImplementedError in others implementations

Note: based on systemctl show

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 addresses 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 (address, 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, supervisorctl_path='supervisorctl', supervisorctl_conf=None, _attrs_cache=None)¶

Test supervisor managed services

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

The path where supervisorctl and its configuration file reside can be specified.

>>> gunicorn = host.supervisor("gunicorn", "/usr/bin/supervisorctl", "/etc/supervisor/supervisord.conf")
>>> gunicorn.status
'RUNNING'

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(supervisorctl_path='supervisorctl', supervisorctl_conf=None)¶

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)>]

The path where supervisorctl and its configuration file reside can be specified.

>>> host.supervisor.get_services("/usr/bin/supervisorctl", "/etc/supervisor/supervisord.conf")
[<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 information

type¶

OS type

>>> host.system_info.type
'linux'

distribution¶

Distribution name

>>> host.system_info.distribution
'debian'

release¶

Distribution release number

>>> host.system_info.release
'10.2'

codename¶

Release code name

>>> host.system_info.codename
'bullseye'

arch¶

Host architecture

>>> host.system_info.arch
'x86_64'

User¶

class User(name=None)¶

Test unix users

If name is not supplied, test the current user

name¶: Return user name

exists¶

Test if user exists

>>> host.user("root").exists
True
>>> host.user("nosuchuser").exists
False

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 encrypted user password

password_max_days¶: Return the maximum number of days between password changes

password_min_days¶: Return the minimum number of days between password changes

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