How to use testinfra - 10 common examples

# A literal IPv6 address might be like
        #  [fe80:0::a:b:c]:80
        # thus, below in words; if this starts with a '[' assume it
        # encloses an ipv6 address with a closing ']', with a possible
        # trailing port after a colon
        if name.startswith('['):
            name, port = name.split(']')
            name = name[1:]
            if port.startswith(':'):
                port = port[1:]
            else:
                port = None
        else:
            if ':' in name:
                name, port = name.split(':', 1)
        name = testinfra.utils.urlunquote(name)
        if user is not None:
            user = testinfra.utils.urlunquote(user)
        if password is not None:
            password = testinfra.utils.urlunquote(password)
        return HostSpec(name, port, user, password)

current = None
    for line in data.splitlines():
        if not current:
            current = line.split("'")[1]
            state[current] = {}
        elif current and line == "}":
            current = None
        elif current:
            key, value = line.split(' => ')
            key = key.strip()
            value = value.split("'")[1]
            state[current][key] = value
    return state


class PuppetResource(InstanceModule):
    """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',
        },
    }
    """

# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import unicode_literals

from testinfra.modules.base import InstanceModule


class Environment(InstanceModule):
    """Get Environment variables

    Example:

     >>> host.environment()
    {
        "EDITOR": "vim",
        "SHELL": "/bin/bash",
        [...]
    }
    """

    def __call__(self):
        ret_val = dict(
            i.split('=', 1) for i in self.check_output('env -0').split(
                '\x00') if i

    @cached_property
    def _data(self):
        header = ['RO', 'RA', 'SSZ', 'BSZ', 'StartSec', 'Size', 'Device']
        command = 'blockdev  --report %s'
        blockdev = self.run(command % self.device)
        if blockdev.rc != 0 or blockdev.stderr:
            raise RuntimeError("Failed to gather data: %s" % blockdev.stderr)
        output = blockdev.stdout.splitlines()
        if len(output) < 2:
            raise RuntimeError("No data from %s" % self.device)
        if output[0].split() != header:
            raise RuntimeError('Unknown output of blockdev: %s' % output[0])
        fields = output[1].split()
        return {
            'rw_mode': str(fields[0]),
            'read_ahead': int(fields[1]),
            'sector_size': int(fields[2]),

if host.system_info.type == "freebsd":
            return FreeBSDPackage
        if host.system_info.type in ("openbsd", "netbsd"):
            return OpenBSDPackage
        if host.exists("dpkg-query"):
            return DebianPackage
        if host.exists("rpm"):
            return RpmPackage
        if host.exists("apk"):
            return AlpinePackage
        if host.system_info.distribution == "arch":
            return ArchPackage
        raise NotImplementedError


class DebianPackage(Package):

    @property
    def is_installed(self):
        result = self.run_test("dpkg-query -f '${Status}' -W %s", self.name)
        if result.rc == 1:
            return False
        out = result.stdout.strip().split()
        installed_status = ["ok", "installed"]
        return out[0] in ["install", "hold"] and out[1:3] == installed_status

    @property
    def release(self):
        raise NotImplementedError

    @property
    def version(self):

    @property
    def release(self):
        raise NotImplementedError

    @property
    def version(self):
        out = self.check_output("dpkg-query -f '${Status} ${Version}' -W %s",
                                self.name)
        splitted = out.split()
        assert splitted[0].lower() in ('install', 'hold'), (
            "The package %s is not installed, dpkg-query output: %s" % (
                self.name, out))
        return splitted[3]


class FreeBSDPackage(Package):

    @property
    def is_installed(self):
        EX_UNAVAILABLE = 69
        return self.run_expect(
            [0, EX_UNAVAILABLE], "pkg query %%n %s", self.name).rc == 0

    @property
    def release(self):
        raise NotImplementedError

    @property
    def version(self):
        return self.check_output("pkg query %%v %s", self.name)

    @property
    def is_installed(self):
        EX_UNAVAILABLE = 69
        return self.run_expect(
            [0, EX_UNAVAILABLE], "pkg query %%n %s", self.name).rc == 0

    @property
    def release(self):
        raise NotImplementedError

    @property
    def version(self):
        return self.check_output("pkg query %%v %s", self.name)


class OpenBSDPackage(Package):

    @property
    def is_installed(self):
        return self.run_test("pkg_info -e %s", "%s-*" % (self.name,)).rc == 0

    @property
    def release(self):
        raise NotImplementedError

    @property
    def version(self):
        out = self.check_output("pkg_info -e %s", "%s-*" % (self.name,))
        # OpenBSD: inst:zsh-5.0.5p0
        # NetBSD: zsh-5.0.7nb1
        return out.split(self.name + "-", 1)[1]

def is_installed(self):
        return self.run_test("pkg_info -e %s", "%s-*" % (self.name,)).rc == 0

    @property
    def release(self):
        raise NotImplementedError

    @property
    def version(self):
        out = self.check_output("pkg_info -e %s", "%s-*" % (self.name,))
        # OpenBSD: inst:zsh-5.0.5p0
        # NetBSD: zsh-5.0.7nb1
        return out.split(self.name + "-", 1)[1]


class RpmPackage(Package):

    @property
    def is_installed(self):
        return self.run_test("rpm -q %s", self.name).rc == 0

    @property
    def version(self):
        return self.check_output('rpm -q --queryformat="%%{VERSION}" %s',
                                 self.name)

    @property
    def release(self):
        return self.check_output('rpm -q --queryformat="%%{RELEASE}" %s',
                                 self.name)

    @property
    def is_installed(self):
        return self.run_test("apk -e info %s", self.name).rc == 0

    @property
    def version(self):
        out = self.check_output("apk -e -v info %s", self.name).split("-")
        return out[-2]

    @property
    def release(self):
        out = self.check_output("apk -e -v info %s", self.name).split("-")
        return out[-1]


class ArchPackage(Package):

    @property
    def is_installed(self):
        return self.run_test("pacman -Q %s", self.name).rc == 0

    @property
    def version(self):
        out = self.check_output("pacman -Q %s", self.name).split(" ")
        return out[1]

    @property
    def release(self):
        raise NotImplementedError


class ChocolateyPackage(Package):

    @property
    def is_installed(self):
        return self.run_test("pacman -Q %s", self.name).rc == 0

    @property
    def version(self):
        out = self.check_output("pacman -Q %s", self.name).split(" ")
        return out[1]

    @property
    def release(self):
        raise NotImplementedError


class ChocolateyPackage(Package):

    @property
    def is_installed(self):
        return self.run_test("choco info -lo %s", self.name).rc == 0

    @property
    def version(self):
        _, version = self.check_output(
            "choco info -lo %s -r", self.name).split("|", 1)
        return version

    @property
    def release(self):
        raise NotImplementedError