Fix typing for SSH utilities
Change-Id: I82bad176020949b5937e542b5312b9f8ab225d66
Reviewed-on: https://fuchsia-review.googlesource.com/c/antlion/+/865460
Reviewed-by: Kevin Sakuma <sakuma@google.com>
Fuchsia-Auto-Submit: Sam Balana <sbalana@google.com>
Commit-Queue: Auto-Submit <auto-submit@fuchsia-infra.iam.gserviceaccount.com>
diff --git a/packages/antlion/capabilities/ssh.py b/packages/antlion/capabilities/ssh.py
index 0248c2e..7abc5e9 100644
--- a/packages/antlion/capabilities/ssh.py
+++ b/packages/antlion/capabilities/ssh.py
@@ -17,7 +17,7 @@
import subprocess
import time
from dataclasses import dataclass
-from typing import BinaryIO, List, Union
+from typing import BinaryIO, List, Optional, Union
from antlion import logger, signals
from antlion.net import wait_for_port
@@ -77,7 +77,7 @@
def __init__(self, err: subprocess.TimeoutExpired):
super().__init__(
f'SSH command "{err.cmd}" timed out after {err.timeout}s, '
- f'stdout="{err.stdout}", stderr="{err.stderr}"'
+ f"stdout={err.stdout!r}, stderr={err.stderr!r}"
)
@@ -265,7 +265,7 @@
timeout_sec: int,
connect_retries: int,
force_tty: bool,
- stdin: BinaryIO,
+ stdin: Optional[BinaryIO],
) -> SSHResult:
err: Exception = ValueError("connect_retries cannot be 0")
for i in range(0, connect_retries):
@@ -277,7 +277,7 @@
raise err
def _run(
- self, command: str, timeout_sec: int, force_tty: bool, stdin: BinaryIO
+ self, command: str, timeout_sec: int, force_tty: bool, stdin: Optional[BinaryIO]
) -> SSHResult:
full_command = self.config.full_command(command, force_tty)
self.log.debug(
@@ -366,7 +366,7 @@
SSHTimeout: if there is no response within timeout_sec
"""
file = open(local_path, "rb")
- return self._run_with_retry(
+ self._run_with_retry(
f"cat > {remote_path}",
timeout_sec,
connect_retries,
diff --git a/packages/antlion/controllers/utils_lib/ssh/connection.py b/packages/antlion/controllers/utils_lib/ssh/connection.py
index 23c80d7..35550a8 100644
--- a/packages/antlion/controllers/utils_lib/ssh/connection.py
+++ b/packages/antlion/controllers/utils_lib/ssh/connection.py
@@ -20,6 +20,7 @@
import threading
import time
import uuid
+from typing import Optional
from antlion import logger
from antlion.controllers.utils_lib import host_utils
@@ -68,6 +69,10 @@
@property
def socket_path(self):
"""Returns: The os path to the master socket file."""
+ if self._master_ssh_tempdir is None:
+ raise AttributeError(
+ "socket_path is not available yet; run setup_master_ssh() first"
+ )
return os.path.join(self._master_ssh_tempdir, "socket")
def __init__(self, settings):
@@ -81,11 +86,11 @@
self._formatter = formatter.SshFormatter()
self._lock = threading.Lock()
self._master_ssh_proc = None
- self._master_ssh_tempdir = None
+ self._master_ssh_tempdir: Optional[str] = None
self._tunnels = list()
def log_line(msg):
- return "[SshConnection | %s] %s" % (self._settings.hostname, msg)
+ return f"[SshConnection | {self._settings.hostname}] {msg}"
self.log = logger.create_logger(log_line)
@@ -206,7 +211,7 @@
extra_options["ControlPath"] = self.socket_path
identifier = str(uuid.uuid4())
- full_command = 'echo "CONNECTED: %s"; %s' % (identifier, command)
+ full_command = f'echo "CONNECTED: {identifier}"; {command}'
terminal_command = self._formatter.format_command(
full_command, env, self._settings, extra_options=extra_options
@@ -224,7 +229,7 @@
# Check for a connected message to prevent false negatives.
valid_connection = re.search(
- "^CONNECTED: %s" % identifier, output, flags=re.MULTILINE
+ f"^CONNECTED: {identifier}", output, flags=re.MULTILINE
)
if valid_connection:
# Remove the first line that contains the connect message.
@@ -279,11 +284,11 @@
if unknown_host:
raise Error("Unknown host.", result)
- self.log.error("An unknown error has occurred. Job result: %s" % result)
+ self.log.error(f"An unknown error has occurred. Job result: {result}")
ping_output = job.run(
- "ping %s -c 3 -w 1" % self._settings.hostname, ignore_status=True
+ f"ping {self._settings.hostname} -c 3 -w 1", ignore_status=True
)
- self.log.error("Ping result: %s" % ping_output)
+ self.log.error(f"Ping result: {ping_output}")
if attempts > 1:
self._cleanup_master_ssh()
self.run(command, timeout, ignore_status, env, io_encoding, attempts - 1)
@@ -418,7 +423,7 @@
# TODO: This may belong somewhere else: b/32572515
user_host = self._formatter.format_host_name(self._settings)
job.run(
- "scp %s %s:%s" % (local_path, user_host, remote_path),
+ f"scp {local_path} {user_host}:{remote_path}",
ignore_status=ignore_status,
)
@@ -432,7 +437,7 @@
"""
user_host = self._formatter.format_host_name(self._settings)
job.run(
- "scp %s:%s %s" % (user_host, remote_path, local_path),
+ f"scp {user_host}:{remote_path} {local_path}",
ignore_status=ignore_status,
)
