Type testinfra.backend

Author: philpep

testinfra/backend/__init__.py

@@ -44,8 +44,8 @@ def get_backend_class(connection: str) -> type["testinfra.backend.base.BaseBacke
     return getattr(importlib.import_module(module), name)  # type: ignore[no-any-return]
 
 
-def parse_hostspec(hostspec):
-    kw = {}
+def parse_hostspec(hostspec: str) -> tuple[str, dict[str, Any]]:
+    kw: dict[str, Any] = {}
     if hostspec is not None and "://" in hostspec:
         url = urllib.parse.urlparse(hostspec)
         kw["connection"] = url.scheme

testinfra/backend/ansible.py

@@ -12,6 +12,7 @@
 
 import logging
 import pprint
+from typing import Any, Optional
 
 from testinfra.backend import base
 from testinfra.utils.ansible_runner import AnsibleRunner
@@ -25,13 +26,13 @@ class AnsibleBackend(base.BaseBackend):
 
     def __init__(
         self,
-        host,
-        ansible_inventory=None,
-        ssh_config=None,
-        ssh_identity_file=None,
-        force_ansible=False,
-        *args,
-        **kwargs,
+        host: str,
+        ansible_inventory: Optional[str] = None,
+        ssh_config: Optional[str] = None,
+        ssh_identity_file: Optional[str] = None,
+        force_ansible: bool = False,
+        *args: Any,
+        **kwargs: Any,
     ):
         self.host = host
         self.ansible_inventory = ansible_inventory
@@ -41,10 +42,10 @@ def __init__(
         super().__init__(host, *args, **kwargs)
 
     @property
-    def ansible_runner(self):
+    def ansible_runner(self) -> AnsibleRunner:
         return AnsibleRunner.get_runner(self.ansible_inventory)
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         command = self.get_command(command, *args)
         if not self.force_ansible:
             host = self.ansible_runner.get_host(
@@ -64,8 +65,10 @@ def run(self, command, *args, **kwargs):
             stderr=out["stderr"],
         )
 
-    def run_ansible(self, module_name, module_args=None, **kwargs):
-        def get_encoding():
+    def run_ansible(
+        self, module_name: str, module_args: Optional[str] = None, **kwargs: Any
+    ) -> Any:
+        def get_encoding() -> str:
             return self.encoding
 
         result = self.ansible_runner.run_module(
@@ -80,10 +83,10 @@ def get_encoding():
         )
         return result
 
-    def get_variables(self):
+    def get_variables(self) -> dict[str, Any]:
         return self.ansible_runner.get_variables(self.host)
 
     @classmethod
-    def get_hosts(cls, host, **kwargs):
+    def get_hosts(cls, host: str, **kwargs: Any) -> list[str]:
         inventory = kwargs.get("ansible_inventory")
         return AnsibleRunner.get_runner(inventory).get_hosts(host or "all")

testinfra/backend/chroot.py

@@ -10,6 +10,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import os.path
+from typing import Any
 
 from testinfra.backend import base
 
@@ -23,11 +24,11 @@ class ChrootBackend(base.BaseBackend):
 
     NAME = "chroot"
 
-    def __init__(self, name, *args, **kwargs):
+    def __init__(self, name: str, *args: Any, **kwargs: Any):
         self.name = name
         super().__init__(self.name, *args, **kwargs)
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         if not os.path.exists(self.name) and os.path.isdir(self.name):
             raise RuntimeError(
                 "chroot path {} not found or not a directory".format(self.name)

testinfra/backend/docker.py

@@ -10,17 +10,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+
 from testinfra.backend import base
 
 
 class DockerBackend(base.BaseBackend):
     NAME = "docker"
 
-    def __init__(self, name, *args, **kwargs):
+    def __init__(self, name: str, *args: Any, **kwargs: Any):
         self.name, self.user = self.parse_containerspec(name)
         super().__init__(self.name, *args, **kwargs)
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         cmd = self.get_command(command, *args)
         if self.user is not None:
             out = self.run_local(

testinfra/backend/kubectl.py

@@ -10,21 +10,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+
 from testinfra.backend import base
 
 
 class KubectlBackend(base.BaseBackend):
     NAME = "kubectl"
 
-    def __init__(self, name, *args, **kwargs):
+    def __init__(self, name: str, *args: Any, **kwargs: Any):
         self.name = name
         self.container = kwargs.get("container")
         self.namespace = kwargs.get("namespace")
         self.kubeconfig = kwargs.get("kubeconfig")
         self.context = kwargs.get("context")
         super().__init__(self.name, *args, **kwargs)
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         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

testinfra/backend/local.py

@@ -10,21 +10,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+
 from testinfra.backend import base
 
 
 class LocalBackend(base.BaseBackend):
     NAME = "local"
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any):
         super().__init__("local", **kwargs)
 
-    def get_pytest_id(self):
+    def get_pytest_id(self) -> str:
         return "local"
 
     @classmethod
-    def get_hosts(cls, host, **kwargs):
+    def get_hosts(cls, host: str, **kwargs: Any) -> list[str]:
         return [host]
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         return self.run_local(self.get_command(command, *args))

testinfra/backend/lxc.py

@@ -10,17 +10,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+
 from testinfra.backend import base
 
 
 class LxcBackend(base.BaseBackend):
     NAME = "lxc"
 
-    def __init__(self, name, *args, **kwargs):
+    def __init__(self, name: str, *args: Any, **kwargs: Any):
         self.name = name
         super().__init__(self.name, *args, **kwargs)
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         cmd = self.get_command(command, *args)
         out = self.run_local(
             "lxc exec %s --mode=non-interactive -- " "/bin/sh -c %s", self.name, cmd

testinfra/backend/openshift.py

@@ -10,20 +10,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+
 from testinfra.backend import base
 
 
 class OpenShiftBackend(base.BaseBackend):
     NAME = "openshift"
 
-    def __init__(self, name, *args, **kwargs):
+    def __init__(self, name: str, *args: Any, **kwargs: Any):
         self.name = name
         self.container = kwargs.get("container")
         self.namespace = kwargs.get("namespace")
         self.kubeconfig = kwargs.get("kubeconfig")
         super().__init__(self.name, *args, **kwargs)
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         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

testinfra/backend/paramiko.py

@@ -23,7 +23,9 @@
     )
 
 import functools
+from typing import Any, Optional
 
+import paramiko.pkey
 import paramiko.ssh_exception
 
 from testinfra.backend import base
@@ -32,7 +34,9 @@
 class IgnorePolicy(paramiko.MissingHostKeyPolicy):
     """Policy for ignoring missing host key."""
 
-    def missing_host_key(self, client, hostname, key):
+    def missing_host_key(
+        self, client: paramiko.SSHClient, hostname: str, key: paramiko.pkey.PKey
+    ) -> None:
         pass
 
 
@@ -41,12 +45,12 @@ class ParamikoBackend(base.BaseBackend):
 
     def __init__(
         self,
-        hostspec,
-        ssh_config=None,
-        ssh_identity_file=None,
-        timeout=10,
-        *args,
-        **kwargs,
+        hostspec: str,
+        ssh_config: Optional[str] = None,
+        ssh_identity_file: Optional[str] = None,
+        timeout: int = 10,
+        *args: Any,
+        **kwargs: Any,
     ):
         self.host = self.parse_hostspec(hostspec)
         self.ssh_config = ssh_config
@@ -55,7 +59,13 @@ def __init__(
         self.timeout = int(timeout)
         super().__init__(self.host.name, *args, **kwargs)
 
-    def _load_ssh_config(self, client, cfg, ssh_config, ssh_config_dir="~/.ssh"):
+    def _load_ssh_config(
+        self,
+        client: paramiko.SSHClient,
+        cfg: dict[str, Any],
+        ssh_config: paramiko.SSHConfig,
+        ssh_config_dir: str = "~/.ssh",
+    ) -> None:
         for key, value in ssh_config.lookup(self.host.name).items():
             if key == "hostname":
                 cfg[key] = value
@@ -85,7 +95,7 @@ def _load_ssh_config(self, client, cfg, ssh_config, ssh_config_dir="~/.ssh"):
                     self._load_ssh_config(client, cfg, new_ssh_config, ssh_config_dir)
 
     @functools.cached_property
-    def client(self):
+    def client(self) -> paramiko.SSHClient:
         client = paramiko.SSHClient()
         client.set_missing_host_key_policy(paramiko.WarningPolicy())
         cfg = {
@@ -118,11 +128,13 @@ def client(self):
 
         if self.ssh_identity_file:
             cfg["key_filename"] = self.ssh_identity_file
-        client.connect(**cfg)
+        client.connect(**cfg)  # type: ignore[arg-type]
         return client
 
-    def _exec_command(self, command):
-        chan = self.client.get_transport().open_session()
+    def _exec_command(self, command: bytes) -> tuple[int, bytes, bytes]:
+        transport = self.client.get_transport()
+        assert transport is not None
+        chan = transport.open_session()
         if self.get_pty:
             chan.get_pty()
         chan.exec_command(command)
@@ -131,17 +143,19 @@ def _exec_command(self, command):
         stderr = b"".join(chan.makefile_stderr("rb"))
         return rc, stdout, stderr
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         command = self.get_command(command, *args)
-        command = self.encode(command)
+        cmd = self.encode(command)
         try:
-            rc, stdout, stderr = self._exec_command(command)
+            rc, stdout, stderr = self._exec_command(cmd)
         except paramiko.ssh_exception.SSHException:
-            if not self.client.get_transport().is_active():
+            transport = self.client.get_transport()
+            assert transport is not None
+            if not transport.is_active():
                 # try to reinit connection (once)
                 del self.client
-                rc, stdout, stderr = self._exec_command(command)
+                rc, stdout, stderr = self._exec_command(cmd)
             else:
                 raise
 
-        return self.result(rc, command, stdout, stderr)
+        return self.result(rc, cmd, stdout, stderr)

testinfra/backend/podman.py

@@ -10,17 +10,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+
 from testinfra.backend import base
 
 
 class PodmanBackend(base.BaseBackend):
     NAME = "podman"
 
-    def __init__(self, name, *args, **kwargs):
+    def __init__(self, name: str, *args: Any, **kwargs: Any):
         self.name, self.user = self.parse_containerspec(name)
         super().__init__(self.name, *args, **kwargs)
 
-    def run(self, command, *args, **kwargs):
+    def run(self, command: str, *args: str, **kwargs: Any) -> base.CommandResult:
         cmd = self.get_command(command, *args)
         if self.user is not None:
             out = self.run_local(