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 __future__ import absolute_import
from testinfra.backend import base
class KubectlBackend(base.BaseBackend):
NAME = "kubectl"
def __init__(self, name, *args, **kwargs):
self.name = name
self.container = kwargs.get('container')
self.namespace = kwargs.get('namespace')
self.kubeconfig = kwargs.get('kubeconfig')
super(KubectlBackend, self).__init__(self.name, *args, **kwargs)
def run(self, command, *args, **kwargs):
cmd = self.get_command(command, *args)
# `kubectl exec` does not support specifying the user to run as.
# See https://github.com/kubernetes/kubernetes/issues/30656
kcmd = 'kubectl '
kcmd_args = []
if self.kubeconfig is not 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.backend import base
class LocalBackend(base.BaseBackend):
NAME = "local"
def __init__(self, *args, **kwargs):
super(LocalBackend, self).__init__("local", **kwargs)
def get_pytest_id(self):
return "local"
@classmethod
def get_hosts(cls, host, **kwargs):
return [host]
def run(self, command, *args, **kwargs):
return self.run_local(self.get_command(command, *args))
"You must install paramiko package (pip install paramiko) "
"to use the paramiko backend"))
import paramiko.ssh_exception
from testinfra.backend import base
from testinfra.utils import cached_property
class IgnorePolicy(paramiko.MissingHostKeyPolicy):
"""Policy for ignoring missing host key."""
def missing_host_key(self, client, hostname, key):
pass
class ParamikoBackend(base.BaseBackend):
NAME = "paramiko"
def __init__(
self, hostspec, ssh_config=None, ssh_identity_file=None,
timeout=10, *args, **kwargs):
self.host = self.parse_hostspec(hostspec)
self.ssh_config = ssh_config
self.ssh_identity_file = ssh_identity_file
self.get_pty = False
self.timeout = int(timeout)
super(ParamikoBackend, self).__init__(self.host.name, *args, **kwargs)
def _load_ssh_config(self, client, cfg, ssh_config):
for key, value in ssh_config.lookup(self.host.name).items():
if key == "hostname":
cfg[key] = value
# 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
import base64
from testinfra.backend import base
class SshBackend(base.BaseBackend):
"""Run command through ssh command"""
NAME = "ssh"
def __init__(self, hostspec, ssh_config=None, ssh_identity_file=None,
timeout=10, controlpersist=60, ssh_extra_args=None,
*args, **kwargs):
self.host = self.parse_hostspec(hostspec)
self.ssh_config = ssh_config
self.ssh_identity_file = ssh_identity_file
self.timeout = int(timeout)
self.controlpersist = int(controlpersist)
self.ssh_extra_args = ssh_extra_args
super(SshBackend, self).__init__(self.host.name, *args, **kwargs)
def run(self, command, *args, **kwargs):
return self.run_ssh(self.get_command(command, *args))
#
# 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 __future__ import absolute_import
from testinfra.backend import base
class DockerBackend(base.BaseBackend):
NAME = "docker"
def __init__(self, name, *args, **kwargs):
self.name, self.user = self.parse_containerspec(name)
super(DockerBackend, self).__init__(self.name, *args, **kwargs)
def run(self, command, *args, **kwargs):
cmd = self.get_command(command, *args)
if self.user is not None:
out = self.run_local(
"docker exec -u %s %s /bin/sh -c %s",
self.user, self.name, cmd)
else:
out = self.run_local(
"docker exec %s /bin/sh -c %s", self.name, cmd)
out.command = self.encode(cmd)
#
# 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 __future__ import absolute_import
from testinfra.backend import base
class LxcBackend(base.BaseBackend):
NAME = "lxc"
def __init__(self, name, *args, **kwargs):
self.name = name
super(LxcBackend, self).__init__(self.name, *args, **kwargs)
def run(self, command, *args, **kwargs):
cmd = self.get_command(command, *args)
out = self.run_local("lxc exec %s --mode=non-interactive -- "
"/bin/sh -c %s", self.name, cmd)
out.command = self.encode(cmd)
return out
#
# 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 __future__ import absolute_import
from testinfra.backend import base
class OpenShiftBackend(base.BaseBackend):
NAME = "openshift"
def __init__(self, name, *args, **kwargs):
self.name = name
self.container = kwargs.get('container')
self.namespace = kwargs.get('namespace')
self.kubeconfig = kwargs.get('kubeconfig')
super(OpenShiftBackend, self).__init__(self.name, *args, **kwargs)
def run(self, command, *args, **kwargs):
cmd = self.get_command(command, *args)
# `oc exec` does not support specifying the user to run as.
# See https://github.com/kubernetes/kubernetes/issues/30656
oscmd = 'oc '
oscmd_args = []
if self.kubeconfig is not None:
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import unicode_literals
from __future__ import absolute_import
import logging
import pprint
from testinfra.backend import base
from testinfra.utils.ansible_runner import AnsibleRunner
logger = logging.getLogger("testinfra")
class AnsibleBackend(base.BaseBackend):
NAME = "ansible"
HAS_RUN_ANSIBLE = True
def __init__(self, host, ansible_inventory=None, ssh_config=None,
ssh_identity_file=None, force_ansible=False,
*args, **kwargs):
self.host = host
self.ansible_inventory = ansible_inventory
self.ssh_config = ssh_config
self.ssh_identity_file = ssh_identity_file
self.force_ansible = force_ansible
super(AnsibleBackend, self).__init__(host, *args, **kwargs)
@property
def ansible_runner(self):
return AnsibleRunner.get_runner(self.ansible_inventory)
return s
else:
for c in s:
if c not in _safechars:
break
else:
if not s:
return "''"
return s
# use single quotes, and put single quotes into double quotes
# the string $'b is then quoted as '$'"'"'b'
return '"' + s.replace('"', '"\'"\'"') + '"'
class WinRMBackend(base.BaseBackend):
"""Run command through winrm command"""
NAME = "winrm"
def __init__(self, hostspec, no_ssl=False, no_verify_ssl=False,
read_timeout_sec=None, operation_timeout_sec=None,
*args, **kwargs):
self.host = self.parse_hostspec(hostspec)
self.conn_args = {
'endpoint': '{}://{}{}/wsman'.format(
'http' if no_ssl else 'https',
self.host.name,
':{}'.format(self.host.port) if self.host.port else ''),
'transport': 'ntlm',
'username': self.host.user,
'password': self.host.password,
}
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import unicode_literals
from __future__ import absolute_import
try:
import salt.client
except ImportError:
raise RuntimeError(
"You must install salt package to use the salt backend")
from testinfra.backend import base
class SaltBackend(base.BaseBackend):
HAS_RUN_SALT = True
NAME = "salt"
def __init__(self, host, *args, **kwargs):
self.host = host
self._client = None
super(SaltBackend, self).__init__(self.host, *args, **kwargs)
@property
def client(self):
if self._client is None:
self._client = salt.client.LocalClient()
return self._client
def run(self, command, *args, **kwargs):
command = self.get_command(command, *args)