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)¶ -
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¶m2=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', }
-
exception
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
¶
-
is_symlink
¶
-
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
-
Interface¶
Iptables¶
-
class
Iptables
¶ Test iptables rule exists
-
rules
(table='filter', chain=None)¶ Returns list of iptables rules
Based on ouput of iptables -t TABLE -S CHAIN command
- optionally takes takes the following arguments:
- table: defaults to filter
- chain: defaults to all chains
>>> 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().get_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:
- 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='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='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()
orget()
, 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¶
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
foris_running
rcctl ls on
foris_enabled
(only OpenBSD >= 5.8) - NetBSD:
/etc/rc.d/$name onestatus
foris_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', ...]
- Unix sockets:
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.
-
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
-