summaryrefslogtreecommitdiff
path: root/test/lib/ansible_test/_internal/connections.py
diff options
context:
space:
mode:
authorMatt Clay <matt@mystile.com>2023-03-13 18:23:04 -0700
committerGitHub <noreply@github.com>2023-03-13 18:23:04 -0700
commitd36f52614d15d763f9be95abdb2da5488ad2ceb5 (patch)
treed5341599b4c2a2769409369fa8b927c391cad547 /test/lib/ansible_test/_internal/connections.py
parentff91a95ef11acedd0ff8367143a882c72f7637ae (diff)
downloadansible-d36f52614d15d763f9be95abdb2da5488ad2ceb5.tar.gz
[stable-2.14] ansible-test - Improve code formatting (#79983) (#80206)
* ansible-test - Add blank lines after docstrings * ansible-test - Preserve formatting of arg pairs * ansible-test - Remove unused string * ansible-test - Remove pointless dict() usage * ansible-test - Clean up initial func arg indenting * ansible-test - Clean up constructor arg indenting * ansible-test - Clean up func arg wrapping * ansible-test - Clean up comma and paren placement (cherry picked from commit 715ab99462b1799f4a0c1caeddf161e930adf13f)
Diffstat (limited to 'test/lib/ansible_test/_internal/connections.py')
-rw-r--r--test/lib/ansible_test/_internal/connections.py102
1 files changed, 56 insertions, 46 deletions
diff --git a/test/lib/ansible_test/_internal/connections.py b/test/lib/ansible_test/_internal/connections.py
index 4823b1a476..84dc84b208 100644
--- a/test/lib/ansible_test/_internal/connections.py
+++ b/test/lib/ansible_test/_internal/connections.py
@@ -44,33 +44,37 @@ from .become import (
class Connection(metaclass=abc.ABCMeta):
"""Base class for connecting to a host."""
+
@abc.abstractmethod
- def run(self,
- command: list[str],
- capture: bool,
- interactive: bool = False,
- data: t.Optional[str] = None,
- stdin: t.Optional[t.IO[bytes]] = None,
- stdout: t.Optional[t.IO[bytes]] = None,
- output_stream: t.Optional[OutputStream] = None,
- ) -> tuple[t.Optional[str], t.Optional[str]]:
+ def run(
+ self,
+ command: list[str],
+ capture: bool,
+ interactive: bool = False,
+ data: t.Optional[str] = None,
+ stdin: t.Optional[t.IO[bytes]] = None,
+ stdout: t.Optional[t.IO[bytes]] = None,
+ output_stream: t.Optional[OutputStream] = None,
+ ) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified command and return the result."""
- def extract_archive(self,
- chdir: str,
- src: t.IO[bytes],
- ):
+ def extract_archive(
+ self,
+ chdir: str,
+ src: t.IO[bytes],
+ ):
"""Extract the given archive file stream in the specified directory."""
tar_cmd = ['tar', 'oxzf', '-', '-C', chdir]
retry(lambda: self.run(tar_cmd, stdin=src, capture=True))
- def create_archive(self,
- chdir: str,
- name: str,
- dst: t.IO[bytes],
- exclude: t.Optional[str] = None,
- ):
+ def create_archive(
+ self,
+ chdir: str,
+ name: str,
+ dst: t.IO[bytes],
+ exclude: t.Optional[str] = None,
+ ):
"""Create the specified archive file stream from the specified directory, including the given name and optionally excluding the given name."""
tar_cmd = ['tar', 'cf', '-', '-C', chdir]
gzip_cmd = ['gzip']
@@ -90,18 +94,20 @@ class Connection(metaclass=abc.ABCMeta):
class LocalConnection(Connection):
"""Connect to localhost."""
+
def __init__(self, args: EnvironmentConfig) -> None:
self.args = args
- def run(self,
- command: list[str],
- capture: bool,
- interactive: bool = False,
- data: t.Optional[str] = None,
- stdin: t.Optional[t.IO[bytes]] = None,
- stdout: t.Optional[t.IO[bytes]] = None,
- output_stream: t.Optional[OutputStream] = None,
- ) -> tuple[t.Optional[str], t.Optional[str]]:
+ def run(
+ self,
+ command: list[str],
+ capture: bool,
+ interactive: bool = False,
+ data: t.Optional[str] = None,
+ stdin: t.Optional[t.IO[bytes]] = None,
+ stdout: t.Optional[t.IO[bytes]] = None,
+ output_stream: t.Optional[OutputStream] = None,
+ ) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified command and return the result."""
return run_command(
args=self.args,
@@ -117,6 +123,7 @@ class LocalConnection(Connection):
class SshConnection(Connection):
"""Connect to a host using SSH."""
+
def __init__(self, args: EnvironmentConfig, settings: SshConnectionDetail, become: t.Optional[Become] = None) -> None:
self.args = args
self.settings = settings
@@ -136,15 +143,16 @@ class SshConnection(Connection):
self.options.extend(ssh_options_to_list(ssh_options))
- def run(self,
- command: list[str],
- capture: bool,
- interactive: bool = False,
- data: t.Optional[str] = None,
- stdin: t.Optional[t.IO[bytes]] = None,
- stdout: t.Optional[t.IO[bytes]] = None,
- output_stream: t.Optional[OutputStream] = None,
- ) -> tuple[t.Optional[str], t.Optional[str]]:
+ def run(
+ self,
+ command: list[str],
+ capture: bool,
+ interactive: bool = False,
+ data: t.Optional[str] = None,
+ stdin: t.Optional[t.IO[bytes]] = None,
+ stdout: t.Optional[t.IO[bytes]] = None,
+ output_stream: t.Optional[OutputStream] = None,
+ ) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified command and return the result."""
options = list(self.options)
@@ -213,20 +221,22 @@ class SshConnection(Connection):
class DockerConnection(Connection):
"""Connect to a host using Docker."""
+
def __init__(self, args: EnvironmentConfig, container_id: str, user: t.Optional[str] = None) -> None:
self.args = args
self.container_id = container_id
self.user: t.Optional[str] = user
- def run(self,
- command: list[str],
- capture: bool,
- interactive: bool = False,
- data: t.Optional[str] = None,
- stdin: t.Optional[t.IO[bytes]] = None,
- stdout: t.Optional[t.IO[bytes]] = None,
- output_stream: t.Optional[OutputStream] = None,
- ) -> tuple[t.Optional[str], t.Optional[str]]:
+ def run(
+ self,
+ command: list[str],
+ capture: bool,
+ interactive: bool = False,
+ data: t.Optional[str] = None,
+ stdin: t.Optional[t.IO[bytes]] = None,
+ stdout: t.Optional[t.IO[bytes]] = None,
+ output_stream: t.Optional[OutputStream] = None,
+ ) -> tuple[t.Optional[str], t.Optional[str]]:
"""Run the specified command and return the result."""
options = []