diff options
author | Matt Clay <matt@mystile.com> | 2023-03-13 18:23:04 -0700 |
---|---|---|
committer | GitHub <noreply@github.com> | 2023-03-13 18:23:04 -0700 |
commit | d36f52614d15d763f9be95abdb2da5488ad2ceb5 (patch) | |
tree | d5341599b4c2a2769409369fa8b927c391cad547 /test/lib/ansible_test/_internal/connections.py | |
parent | ff91a95ef11acedd0ff8367143a882c72f7637ae (diff) | |
download | ansible-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.py | 102 |
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 = [] |