• [جدید] پرس‌وجوی تمام نام‌های کاربری و نام‌های گروه
• [اصلاح] جلوگیری از بن‌بست Paramiko هنگامی که آزمون بیش از 2MB داده به stdout ارسال می‌کند
• [اصلاح] هماهنگی با تغییرات ماژول shell در ansible
• [اصلاح] افزودن 4 به کد خروج مورد انتظار هنگام اجرای "systemctl is-active"
• [اصلاح] رفع خطای KeyError در MountPoint.__repr__() در صورت عدم وجود mount
• [مستندات] استفاده از دستور pytest به‌جای py.test
• [مستندات] گسترش مستندات backend با افزودن یک بخش عمومی برای مشخصات میزبان (host spec)
• [متفرقه] اجرای lint برای py 3.12 و 3.13 نیز
• [متفرقه] تغییر سیستم بسته‌بندی برای استفاده از hatchling
• [متفرقه] حذف آرگومان اضافی و استفاده‌نشده‌ی "args" در run_winrm()
• [متفرقه] استفاده از ruff format به‌جای black/isort
• [متفرقه] استفاده از ruff به‌جای flake8
• [متفرقه] استفاده از f-string به‌جای str.format()
• [متفرقه] استفاده از dict، list و tuple توکار برای تعیین نوع (typing)
• [متفرقه] استفاده از python 3.11 در طول آزمون‌ها
• [متفرقه] اصلاح آزمون‌های salt
• [متفرقه] رفع مشکل آزمون‌هایی که به دلیل سپری شدن تاریخ انقضا ناموفق می‌شدند
• [متفرقه] حذف کتابخانه crypt از فرایند آزمون

•

[اصلاح] رفع مشکل سازگاری command -v با پوسته dash

•

رفع مشکل ساخت مستندات

• رفع مشکل تطبیق ansible بدون inventory با "localhost"
• رفع مشکل پشتیبانی از ansible 2.8 بدون inventory
• رفع مشکلی در ansible/paramiko که پیکربندی میزبان‌ها را از ~/.ssh/config نمی‌خواند
• امکان ارسال --ssh-config و --ssh-identity-file به اتصال ansible

• socket: رفع مشکل در شناسایی سوکت‌های شنونده udp (#311)
• ssh backend: افزودن پشتیبانی از GSSAPI

• backends: بازگشت به کدگذاری UTF-8 در صورت ASCII بودن کدگذاری سیستم.
• Service: رفع باگ is_running() در سیستم‌های مبتنی بر Upstart.

$ pip install paramiko
$ pytest -v --hosts=localhost,root@webserver:2222 test_myinfra.py
====================== test session starts ======================
platform linux -- Python 2.7.3 -- py-1.4.26 -- pytest-2.6.4
plugins: testinfra
collected 3 items
test_myinfra.py::test_passwd_file[localhost] PASSED
test_myinfra.py::test_nginx_is_installed[localhost] PASSED
test_myinfra.py::test_nginx_running_and_enabled[localhost] PASSED
test_myinfra.py::test_passwd_file[root@webserver:2222] PASSED
test_myinfra.py::test_nginx_is_installed[root@webserver:2222] PASSED
test_myinfra.py::test_nginx_running_and_enabled[root@webserver:2222] PASSED
=================== 6 passed in 8.49 seconds ====================

testinfra_hosts = ["localhost", "root@webserver:2222"]
def test_foo(host):
    [....]

$ pip install pytest-xdist
# Launch tests using 3 processes
$ pytest -n 3 -v --host=web1,web2,web3,web4,web5,web6 test_myinfra.py

# Test recursively all test files (starting with `test_`) in current directory
$ pytest
# Filter function/hosts with pytest -k option
$ pytest --hosts=webserver,dnsserver -k webserver -k nginx

$ pip install pytest-testinfra[ansible,salt]

--hosts=<backend>://<user>:<password>@<host>:<port>

import urllib.parse
user = urllib.parse.quote('user:name')
password = urllib.parse.quote('p@ssw:rd')
host = 'hostname'
port = 22
host_spec = f"ssh://{user}:{password}@{host}:{port}"
print(host_spec)

testinfra --hosts=ssh://user:password@hostname:22

testinfra --hosts=docker://container_id

testinfra --hosts=ssh://user:password@hostname:22,docker://container_id,local://

$ pytest --sudo test_myinfra.py

$ pytest --ssh-config=/path/to/ssh_config --hosts=server

$ pytest --hosts='docker://[user@]container_id_or_name'

$ pytest --hosts='podman://[user@]container_id_or_name'

$ pytest --hosts='ssh://server'
$ pytest --ssh-config=/path/to/ssh_config --hosts='ssh://server'
$ pytest --ssh-identity-file=/path/to/key --hosts='ssh://server'
$ pytest --hosts='ssh://server?timeout=60&controlpersist=120'
$ pytest --hosts='ssh://server' --ssh-extra-args='-o StrictHostKeyChecking=no'

$ pytest --hosts='salt://*'
$ pytest --hosts='salt://minion1,salt://minion2'
$ pytest --hosts='salt://web*'
$ pytest --hosts='salt://G@os:Debian'

$ pytest --hosts='ansible://all' # tests all inventory hosts
$ pytest --hosts='ansible://host1,ansible://host2'
$ pytest --hosts='ansible://web*'

$ pytest --force-ansible --hosts='ansible://all'
$ pytest --hosts='ansible://host?force_ansible=True'

# will use the default namespace and default container
$ pytest --hosts='kubectl://mypod-a1b2c3'
# specify container name and namespace
$ pytest --hosts='kubectl://somepod-2536ab?container=nginx&namespace=web'
# specify the kubeconfig context to use
$ pytest --hosts='kubectl://somepod-2536ab?context=k8s-cluster-a&container=nginx'
# you can specify kubeconfig either from KUBECONFIG environment variable
# or when working with multiple configuration with the "kubeconfig" option
$ pytest --hosts='kubectl://somepod-123?kubeconfig=/path/kubeconfig,kubectl://otherpod-123?kubeconfig=/other/kubeconfig'

$ pytest --hosts='winrm://Administrator:Password@127.0.0.1'
$ pytest --hosts='winrm://vagrant@127.0.0.1:2200?no_ssl=true&no_verify_ssl=true'

$ pytest --hosts='winrm://vagrant@127.0.0.1:2200?read_timeout_sec=120&operation_timeout_sec=100'

$ pytest --hosts='lxc://container_name'

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

property has_command_v: bool

در صورتی که command -v در دسترس باشد، مقدار True را برمی‌گرداند

find_command(command: str, extrapaths: Iterable[str] = ('/sbin', '/usr/sbin')) -> str

مسیر دستور داده‌شده را برمی‌گرداند

در صورت پیدا نشدن دستور، خطای ValueError ایجاد می‌کند

run(command: str, *args: str, **kwargs: Any) -> CommandResult <#testinfra.backend.base.CommandResult>

دستور داده‌شده را اجرا کرده و rc (وضعیت خروج)، stdout و 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

روش مناسب: همواره از نقل‌قول آرگومان‌های شل (shell quoting) استفاده کنید تا از تزریق شل (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: list[int], command: str, *args: str, **kwargs: Any) -> CommandResult <#testinfra.backend.base.CommandResult>

دستور را اجرا کرده و بررسی می‌کند که آیا وضعیت خروج موردانتظار را برمی‌گرداند یا خیر

run_test(command: str, *args: str, **kwargs: Any) -> CommandResult <#testinfra.backend.base.CommandResult>

دستور را اجرا کرده و بررسی می‌کند که آیا وضعیت خروج 0 یا 1 را برمی‌گرداند یا خیر

check_output(command: str, *args: str, **kwargs: Any) -> str

دریافت stdout دستوری که با موفقیت اجرا شده است

classmethod get_host(hostspec: str, **kwargs: Any) -> Host <#testinfra.host.Host>

یک نمونه از Host را از hostspec برمی‌گرداند

مقدار hostspec باید به این صورت باشد: <backend_type>://<name>?param1=value1&param2=value2

پارامترها را می‌توان در **kwargs نیز ارسال کرد (به عنوان مثال، get_host("local://", sudo=True) معادل get_host("local://?sudo=true") است)

مثال‌ها:

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

exception AnsibleException(result)

استثنایی که هنگام وقوع خطا در یک فراخوانی ansible ایجاد می‌شود

نتیجهٔ حاصل از ansible از طریق مشخصهٔ result قابل دسترسی است

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

یک dict از متغیرهای ansible را برمی‌گرداند

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

class Addr(name)

بررسی نشانی دوردست

مثال:

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

همچنین می‌تواند درون یک فضای نام شبکه https://man7.org/linux/man-pages/man7/namespaces.7.html استفاده شود.

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

فضاهای نام شبکه تنها زمانی قابل استفاده هستند که دستور ip https://man7.org/linux/man-pages/man8/ip.8.html در دسترس باشد؛ زیرا در این صورت، ماژول از ip-netns https://man7.org/linux/man-pages/man8/ip-netns.8.html به عنوان پیشوند دستور استفاده می‌کند. در غیر این صورت، خطای NotImplementedError برانگیخته خواهد شد.

property is_partition

در صورتی که دستگاه یک پارتیشن باشد True بازمی‌گرداند.

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

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

property size

اندازه دستگاه را بر حسب بایت بازمی‌گرداند.

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

property sector_size

اندازه سکتور دستگاه را بر حسب بایت بازمی‌گرداند.

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

property block_size

اندازه بلوک دستگاه را بر حسب بایت بازمی‌گرداند.

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

property start_sector

سکتور آغازین دستگاه را بر روی دستگاه زیرین بازمی‌گرداند.

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

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

property is_writable

اگر دستگاه قابل نوشتن باشد (وضعیت RO نداشته باشد) True بازمی‌گرداند.

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

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

property ra

مقدار Read Ahead دستگاه را بر حسب سکتورهای ۵۱۲ بایتی بازمی‌گرداند.

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

class Docker(name)

بررسی و آزمودن کانتینرهای docker در حال اجرا بر روی سیستم.

مثال:

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

property exists

بررسی وجود داشتن فایل

>>> host.file("/etc/passwd").exists
True
>>> host.file("/nonexistent").exists
False

property is_file

بررسی اینکه آیا مسیر داده‌شده یک فایل معمولی (regular file) است یا خیر

property is_directory

بررسی اینکه آیا مسیر داده‌شده وجود دارد و یک دایرکتوری است یا خیر

property is_executable

بررسی اینکه آیا مسیر داده‌شده وجود دارد و مجوز اجرا به آن اعطا شده است یا خیر

property is_pipe

بررسی اینکه آیا مسیر داده‌شده وجود دارد و یک لوله (pipe) است یا خیر

property is_socket

بررسی اینکه آیا مسیر داده‌شده وجود دارد و یک سوکت (socket) است یا خیر

property is_symlink

بررسی اینکه آیا مسیر داده‌شده وجود دارد و یک پیوند نمادین (symlink) است یا خیر

property linked_to

حل پیوند نمادین و مشخص کردن مقصد آن

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

property user

بازگرداندن مالک فایل به‌صورت رشته

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

property uid

بازگرداندن شناسه کاربری (UID) فایل به‌صورت یک عدد صحیح

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

property group

بازگرداندن نام گروه فایل به‌صورت رشته

property gid

بازگرداندن شناسه گروه (GID) فایل به‌صورت یک عدد صحیح

property mode

بازگرداندن حالت دسترسی فایل (file mode) به‌صورت عدد صحیح در مبنای هشت (octal)

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

همچنین می‌توانید از مقادیر ثابت حالت دسترسی فایل در کتابخانهٔ stat https://docs.python.org/3/library/stat.html برای بررسی حالت فایل استفاده کنید.

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

contains(pattern)

بررسی محتوای فایل برای یافتن الگو (pattern)

این متد از grep استفاده کرده و در نتیجه از قواعد نحو عبارات منظم (regex) در grep پیروی می‌کند.

property sha256sum

محاسبه چکیده پیام SHA256 محتوای فایل

property content

بازگرداندن محتوای فایل به صورت بایت

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

property content_string

بازگرداندن محتوای فایل به صورت رشته

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

property mtime

بازگرداندن زمان آخرین تغییر به عنوان یک شیء datetime.datetime

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

property size

بازگرداندن اندازه فایل بر حسب بایت

listdir()

بازگرداندن فهرست موارد موجود در دایرکتوری

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

property exists

بررسی وجود داشتن گروه

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

property get_all_groups

بازگرداندن فهرستی از نام‌های گروه‌های محلی و دوردست

>>> host.group("anyname").get_all_groups
["root", "wheel", "man", "tty", <...>]

property get_local_groups

بازگرداندن فهرستی از نام‌های گروه‌های محلی

>>> host.group("anyname").get_local_groups
["root", "wheel", "man", "tty", <...>]

property gid

property members

بازگرداندن تمام کاربرانی که عضو این گروه هستند.

property exists

property speed

property addresses

بازگرداندن آدرس‌های ipv4 و ipv6 روی رابط

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

property link

بازگرداندن ویژگی‌های پیوند (link) مرتبط با رابط.

>>> host.interface("lo").link
{'address': '00:00:00:00:00:00',
'broadcast': '00:00:00:00:00:00',
'flags': ['LOOPBACK', 'UP', 'LOWER_UP'],
'group': 'default',
'ifindex': 1,
'ifname': 'lo',
'link_type': 'loopback',
'linkmode': 'DEFAULT',
'mtu': 65536,
'operstate': 'UNKNOWN',
'qdisc': 'noqueue',
'txqlen': 1000}

routes(scope=None)

بازگرداندن مسیرهای مرتبط با رابط، با امکان پالایش اختیاری بر اساس دامنه scope ("host"، "link" یا "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()

بازگرداندن نام تمامی رابط‌ها.

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

classmethod default(family=None)

بازگرداندن رابط مورد استفاده برای مسیر پیش‌فرض (default route).

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

به صورت اختیاری، می‌توان خانواده پروتکل مورد استفاده را مشخص کرد.

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

class Iptables

بررسی وجود قانون در iptables

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

property exists

در صورت وجود نقطه اتصال مقدار True برمی‌گرداند

>>> host.mount_point("/").exists
True

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

property filesystem

نوع سیستم‌فایل مرتبط را برمی‌گرداند

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

property device

دستگاه مرتبط را برمی‌گرداند

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

property options

فهرستی از گزینه‌هایی که نقطه اتصال با آن‌ها ایجاد شده است را برمی‌گرداند

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

classmethod get_mountpoints()

فهرستی از نمونه‌های MountPoint را برمی‌گرداند

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

property is_installed

بررسی اینکه آیا بسته نصب شده است یا خیر

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

سیستم‌های بسته پشتیبانی‌شده:

property release

اطلاعات مشخصه انتشار (release) را از نسخه بسته برمی‌گرداند

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

property version

نسخه بسته را آن‌طور که توسط سیستم بسته بازگردانده شده برمی‌گرداند

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

class Pip(name, pip_path='pip')

آزمودن مدیر بسته و بسته‌های pip

classmethod get_containers(**filters)

بازگرداندن فهرستی از کانتینرها

به‌طور پیش‌فرض فهرستی از همه‌ی کانتینرها، شامل کانتینرهای غیر در حال اجرا را بازمی‌گرداند.

فیلتر کردن می‌تواند با استفاده از کلیدهای فیلتر تعریف‌شده در podman-ps(1) انجام شود.

اعمال فیلترهای چندگانه برای یک کلید مشخص، با ارائه‌ی فهرستی از رشته‌ها به عنوان مقدار انجام می‌شود.

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

filter(**filters)

دریافت فهرستی از فرآیندهای منطبق

>>> host.process.filter(user="root", comm="zsh")
[<process zsh (pid=2715)>, <process zsh (pid=10502)>, ...]

get(**filters)

دریافت یک فرآیند منطبق

اگر فرآیندی یافت نشود یا چندین فرآیند با فیلترها مطابقت داشته باشند، RuntimeError ایجاد می‌شود.

class PuppetResource(type, name=None)

دریافت منابع puppet

برای دریافت فهرستی از انواع در دسترس، دستور puppet resource --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',
    },
}

• لینوکس: شناسایی Systemd، Upstart یا OpenRC، و بازگشت به SysV در صورت عدم شناسایی
• فری‌بی‌اس‌دی: service(1)
• اوپن‌بی‌اس‌دی: /etc/rc.d/$name check برای is_running rcctl ls on برای is_enabled (تنها در OpenBSD >= 5.8)
• نت‌بی‌اس‌دی: /etc/rc.d/$name onestatus برای is_running (is_enabled هنوز پیاده‌سازی نشده است)

property exists

بررسی اینکه آیا سرویس وجود دارد یا خیر

property is_running

بررسی اینکه آیا سرویس در حال اجرا است یا خیر

property is_enabled

بررسی اینکه آیا سرویس فعال (enabled) است یا خیر

property is_valid

بررسی اینکه آیا سرویس معتبر است یا خیر

این متد تنها در پیاده‌سازی systemd موجود است، و در سایر پیاده‌سازی‌ها خطای NotImplementedError ایجاد خواهد کرد

property is_masked

بررسی اینکه آیا سرویس ماسک (masked) شده است یا خیر

این متد تنها در پیاده‌سازی systemd موجود است، و در سایر پیاده‌سازی‌ها خطای NotImplementedError ایجاد خواهد کرد

property systemd_properties

ویژگی‌های سرویس (unit).

ویژگی‌های سرویس را به عنوان یک dict برمی‌گرداند، ویژگی‌های خالی بازگردانده نمی‌شوند.

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

این متد تنها در پیاده‌سازی systemd موجود است، و در سایر پیاده‌سازی‌ها خطای NotImplementedError ایجاد خواهد کرد

نکته: بر اساس systemctl show https://man7.org/linux/man-pages/man1/systemctl.1.html

class Socket(socketspec)

بررسی سوکت‌های در حال شنود tcp/udp و unix

socketspec باید به صورت <protocol>://<host>:<port> مشخص شود

این ماژول نیازمند وجود دستور netstat روی میزبان هدف است.

مثال:

property is_running

در صورتی که سرویس مدیریت‌شده در وضعیت RUNNING باشد، True را برمی‌گرداند

property status

وضعیت سرویس مدیریت‌شده را برمی‌گرداند

وضعیت می‌تواند STOPPED، STARTING، RUNNING، BACKOFF، STOPPING، EXITED، FATAL یا UNKNOWN باشد.

ببینید: http://supervisord.org/subprocess.html#process-states

property pid

شناسه فرایند (pid به صورت int) سرویس مدیریت‌شده را برمی‌گرداند

classmethod get_services(supervisorctl_path='supervisorctl', supervisorctl_conf=None)

فهرستی از سرویس‌های در حال اجرا تحت supervisor را برمی‌گرداند

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

مسیری که supervisorctl و فایل پیکربندی آن در آن قرار دارند می‌تواند مشخص شود.

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

class Sysctl(name)

بررسی پارامترهای هسته

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

property type

نوع سیستم‌عامل

>>> host.system_info.type
'linux'

property distribution

نام توزیع

>>> host.system_info.distribution
'debian'

property release

شماره انتشار توزیع

>>> host.system_info.release
'10.2'

property codename

اسم رمز انتشار

>>> host.system_info.codename
'bullseye'

property arch

معماری میزبان

>>> host.system_info.arch
'x86_64'

property name

بازگرداندن نام کاربر

property exists

بررسی وجود کاربر

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

property uid

بازگرداندن شناسه کاربر

property gid

بازگرداندن شناسه گروه مؤثر

property group

بازگرداندن نام گروه مؤثر

property gids

بازگرداندن فهرست شناسه‌های گروه‌های کاربر

property groups

بازگرداندن فهرست نام‌های گروه‌های کاربر

property home

بازگرداندن دایرکتوری خانگی کاربر

property shell

بازگرداندن پوسته ورود کاربر

property password

بازگرداندن گذرواژه رمزگذاری‌شده کاربر

property password_max_days

بازگرداندن حداکثر تعداد روزهای بین تغییر گذرواژه

property password_min_days

بازگرداندن حداقل تعداد روزهای بین تغییر گذرواژه

property gecos

بازگرداندن فیلد توضیحات/gecos کاربر

property expiration_date

بازگرداندن تاریخ انقضای حساب کاربری

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

property get_all_users

بازگرداندن فهرستی از نام‌های کاربران محلی و راه‌دور

>>> host.user().get_all_users
["root", "bin", "daemon", "lp", <...>]

property get_local_users

بازگرداندن فهرستی از نام‌های کاربران محلی

>>> host.user().get_local_users
["root", "bin", "daemon", "lp", <...>]

class CommandResult(backend: BaseBackend, exit_status: int, command: bytes, _stdout: str | bytes, _stderr: str | bytes)

شیئی که تمام جزئیات بازگردانده‌شده از اجرای دستور را کپسوله‌سازی می‌کند.

مثال:

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

>>> import testinfra
>>> host = testinfra.get_host("paramiko://root@server:2222", sudo=True)
>>> host.file("/etc/shadow").mode == 0o640
True

import testinfra
def test_same_passwd():
    a = testinfra.get_host("ssh://a")
    b = testinfra.get_host("ssh://b")
    assert a.file("/etc/passwd").content == b.file("/etc/passwd").content

# BAD: If the test fails on nginx, python is not tested
def test_packages(host):
    for name, version in (
        ("nginx", "1.6"),
        ("python", "2.7"),
    ):
        pkg = host.package(name)
        assert pkg.is_installed
        assert pkg.version.startswith(version)
# GOOD: Each package is tested
# $ pytest -v test.py
# [...]
# test.py::test_package[local-nginx-1.6] PASSED
# test.py::test_package[local-python-2.7] PASSED
# [...]
import pytest
@pytest.mark.parametrize("name,version", [
    ("nginx", "1.6"),
    ("python", "2.7"),
])
def test_packages(host, name, version):
    pkg = host.package(name)
    assert pkg.is_installed
    assert pkg.version.startswith(version)

import unittest
import testinfra
class Test(unittest.TestCase):
    def setUp(self):
        self.host = testinfra.get_host("paramiko://root@host")
    def test_nginx_config(self):
        self.assertEqual(self.host.run("nginx -t").rc, 0)
    def test_nginx_service(self):
        service = self.host.service("nginx")
        self.assertTrue(service.is_running)
        self.assertTrue(service.is_enabled)
if __name__ == "__main__":
    unittest.main()

$ python test.py
..
----------------------------------------------------------------
Ran 2 tests in 0.705s
OK

vagrant ssh-config > .vagrant/ssh-config
pytest --hosts=default --ssh-config=.vagrant/ssh-config tests.py

pip install pytest-testinfra paramiko
vagrant up
vagrant ssh-config > .vagrant/ssh-config
pytest --hosts=default --ssh-config=.vagrant/ssh-config --junit-xml junit.xml tests.py

$ pytest -qq --nagios --tb line test_ok.py; echo $?
TESTINFRA OK - 2 passed, 0 failed, 0 skipped in 2.30 seconds
..
0
$ pytest -qq --nagios --tb line test_fail.py; echo $?
TESTINFRA CRITICAL - 1 passed, 1 failed, 0 skipped in 2.24 seconds
.F
/usr/lib/python3/dist-packages/example/example.py:95: error: [Errno 111] error msg
2

verifier:
  name: shell
  command: pytest --hosts="paramiko://${KITCHEN_USERNAME}@${KITCHEN_HOSTNAME}:${KITCHEN_PORT}?ssh_identity_file=${KITCHEN_SSH_KEY}" --junit-xml "junit-${KITCHEN_INSTANCE}.xml" "test/integration/${KITCHEN_SUITE}"

import pytest
import subprocess
import testinfra
# scope='session' uses the same container for all the tests;
# scope='function' uses a new container per test function.
@pytest.fixture(scope='session')
def host(request):
    # build local ./Dockerfile
    subprocess.check_call(['docker', 'build', '-t', 'myimage', '.'])
    # run a container
    docker_id = subprocess.check_output(
        ['docker', 'run', '-d', 'myimage']).decode().strip()
    # return a testinfra connection to the container
    yield testinfra.get_host("docker://" + docker_id)
    # at the end of the test suite, destroy the container
    subprocess.check_call(['docker', 'rm', '-f', docker_id])
def test_myimage(host):
    # 'host' now binds to the container
    assert host.check_output('myapp -v') == 'Myapp 1.0'