Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#
# 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 Module
from testinfra.utils import cached_property
class Service(Module):
"""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)
"""
def __init__(self, name):
self.name = name
super(Service, self).__init__()
# 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 Module
class Package(Module):
"""Test packages status and version"""
def __init__(self, name):
self.name = name
super(Package, self).__init__()
@property
def is_installed(self):
"""Test if the package is installed
>>> host.package("nginx").is_installed
True
Supported package systems:
- apk (Alpine)
# you may not use this file except in compliance with the License.
# 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.
import json
from testinfra.modules.base import Module
class Docker(Module):
"""Test docker containers running on system.
Example:
>>> nginx = host.docker("app_nginx")
>>> nginx.is_running
True
>>> nginx.id
'7e67dc7495ca8f451d346b775890bdc0fb561ecdc97b68fb59ff2f77b509a8fe'
>>> nginx.name
'app_nginx'
"""
def __init__(self, name):
self._name = name
@property
def is_reachable(self):
# pylint: disable=protected-access
if not self._addr._host.exists('nc'):
# Fallback to bash if netcat is not available
return self._addr.run_expect(
[0, 1, 124],
"timeout 1 bash -c 'cat < /dev/null > /dev/tcp/%s/%s'",
self._addr.name, self._port).rc == 0
return self._addr.run(
"nc -w 1 -z %s %s", self._addr.name, self._port).rc == 0
class Addr(Module):
"""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
"""
#
# 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 Module
from testinfra.utils import cached_property
class Interface(Module):
"""Test network interfaces"""
def __init__(self, name):
self.name = name
super(Interface, self).__init__()
@property
def exists(self):
raise NotImplementedError
@property
def speed(self):
raise NotImplementedError
@property
def addresses(self):
#
# 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 Module
from testinfra.utils import cached_property
class BlockDevice(Module):
"""Information for block device.
Should be used with sudo or under root.
If device is not a block device, RuntimeError is raised.
"""
def _data(self):
raise NotImplementedError
def __init__(self, device):
self.device = device
super(BlockDevice, self).__init__()
@property
def is_partition(self):
klass = cls.get_module_class(_host)
return type(klass.__name__, (klass,), {
"_host": _host,
"run": _host.run,
"run_expect": _host.run_expect,
"run_test": _host.run_test,
"check_output": _host.check_output,
"find_command": _host.find_command,
})
@classmethod
def get_module_class(cls, host):
return cls
class InstanceModule(Module):
@classmethod
def get_module(cls, _host):
klass = super(InstanceModule, cls).get_module(_host)
return klass()
#
# 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
import datetime
import six
from testinfra.modules.base import Module
class File(Module):
"""Test various files attributes"""
def __init__(self, path):
self.path = path
super(File, self).__init__()
@property
def exists(self):
"""Test if file exists
>>> host.file("/etc/passwd").exists
True
>>> host.file("/nonexistent").exists
False
"""
family = f
break
if family is None:
raise RuntimeError("Cannot validate ip address '%s'" % (host,))
if port is not None:
try:
port = int(port)
except ValueError:
raise RuntimeError("Cannot validate port '%s'" % (port,))
return protocol, host, port
class Socket(Module):
"""Test listening tcp/udp and unix sockets
``socketspec`` must be specified as ``://:``
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``
"""
_command = None
# 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 Module
class MountPoint(Module):
"""Test Mount Points"""
def __init__(self, path, _attrs_cache=None):
self.path = path
self._attrs_cache = _attrs_cache
super(MountPoint, self).__init__()
@classmethod
def _iter_mountpoints(cls):
raise NotImplementedError
@property
def exists(self):
"""Return True if the mountpoint exists
>>> host.mount_point("/").exists