schema: move ConnectionInfo to schema

lisa/node.py

@@ -27,7 +27,7 @@
 from lisa.util.logger import Logger, create_file_handler, get_logger, remove_handler
 from lisa.util.parallel import run_in_parallel
 from lisa.util.process import ExecutableResult, Process
-from lisa.util.shell import ConnectionInfo, LocalShell, Shell, SshShell
+from lisa.util.shell import LocalShell, Shell, SshShell
 
 T = TypeVar("T")
 __local_node: Optional[Node] = None
@@ -489,7 +489,7 @@ def set_connection_info(
         assert public_port
         assert port
 
-        self._connection_info: ConnectionInfo = ConnectionInfo(
+        self._connection_info: schema.ConnectionInfo = schema.ConnectionInfo(
             public_address,
             public_port,
             username,

lisa/schema.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass, field
 from enum import Enum
 from functools import partial
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Type, TypeVar, Union, cast
 
 from dataclasses_json import (
@@ -1309,6 +1310,46 @@ def type_name(cls) -> str:
         return constants.TESTCASE_TYPE_LEGACY
 
 
+@dataclass_json()
+@dataclass
+class ConnectionInfo:
+    address: str = ""
+    port: int = field(
+        default=22,
+        metadata=field_metadata(
+            field_function=fields.Int, validate=validate.Range(min=1, max=65535)
+        ),
+    )
+    username: str = constants.DEFAULT_USER_NAME
+    password: Optional[str] = ""
+    private_key_file: Optional[str] = ""
+
+    def __post_init__(self, *args: Any, **kwargs: Any) -> None:
+        add_secret(self.username, PATTERN_HEADTAIL)
+        add_secret(self.password)
+        add_secret(self.private_key_file)
+
+        if not self.password and not self.private_key_file:
+            raise LisaException(
+                "at least one of password or private_key_file need to be set when "
+                "connecting"
+            )
+        elif not self.private_key_file:
+            # use password
+            # spurplus doesn't process empty string correctly, use None
+            self.private_key_file = None
+        else:
+            if not Path(self.private_key_file).exists():
+                raise FileNotFoundError(self.private_key_file)
+            self.password = None
+
+        if not self.username:
+            raise LisaException("username must be set")
+
+    def __str__(self) -> str:
+        return f"{self.username}@{self.address}:{self.port}"
+
+
 @dataclass_json()
 @dataclass
 class Runbook:

lisa/util/shell.py

@@ -17,6 +17,7 @@
 from func_timeout import FunctionTimedOut, func_set_timeout  # type: ignore
 from paramiko.ssh_exception import NoValidConnectionsError, SSHException
 
+from lisa import schema
 from lisa.util import InitializableMixin, LisaException, TcpConnectionException
 
 from .logger import Logger
@@ -57,42 +58,6 @@ def wait_tcp_port_ready(
     return is_ready, result
 
 
-class ConnectionInfo:
-    def __init__(
-        self,
-        address: str = "",
-        port: int = 22,
-        username: str = "root",
-        password: Optional[str] = "",
-        private_key_file: Optional[str] = None,
-    ) -> None:
-        self.address = address
-        self.port = port
-        self.username = username
-        self.password = password
-        self.private_key_file = private_key_file
-
-        if not self.password and not self.private_key_file:
-            raise LisaException(
-                "at least one of password or private_key_file need to be set when "
-                "connecting"
-            )
-        elif not self.private_key_file:
-            # use password
-            # spurplus doesn't process empty string correctly, use None
-            self.private_key_file = None
-        else:
-            if not Path(self.private_key_file).exists():
-                raise FileNotFoundError(self.private_key_file)
-            self.password = None
-
-        if not self.username:
-            raise LisaException("username must be set")
-
-    def __str__(self) -> str:
-        return f"{self.username}@{self.address}:{self.port}"
-
-
 class WindowsShellType(object):
     """
     Windows command generator
@@ -137,7 +102,10 @@ def generate_run_command(
 
 
 # retry strategy is the same as spurplus.connect_with_retries.
-def try_connect(connection_info: ConnectionInfo, ssh_timeout: int = 300) -> Any:
+def try_connect(
+    connection_info: schema.ConnectionInfo,
+    ssh_timeout: int = 300,
+) -> Any:
     # spur always run a posix command and will fail on Windows.
     # So try with paramiko firstly.
     paramiko_client = paramiko.SSHClient()
@@ -207,7 +175,7 @@ def _spawn_ssh_process(shell: spur.ssh.SshShell, **kwargs: Any) -> spur.ssh.SshP
 
 
 class SshShell(InitializableMixin):
-    def __init__(self, connection_info: ConnectionInfo) -> None:
+    def __init__(self, connection_info: schema.ConnectionInfo) -> None:
         super().__init__()
         self.is_remote = True
         self._connection_info = connection_info

microsoft/testsuites/nested/common.py

@@ -3,15 +3,15 @@
 
 from typing import Any, Dict, List, Optional, Tuple
 
-from lisa import RemoteNode
+from lisa import RemoteNode, schema
 from lisa.features.network_interface import Synthetic
 from lisa.operating_system import Debian, Fedora, Suse
 from lisa.schema import Node
 from lisa.tools import HyperV, Lscpu, Qemu, Wget
 from lisa.tools.rm import Rm
 from lisa.util import SkippedException, fields_to_dict
 from lisa.util.logger import Logger
-from lisa.util.shell import ConnectionInfo, try_connect
+from lisa.util.shell import try_connect
 
 QEMU_NESTED_VM_IMAGE_NAME = "image.qcow2"
 HYPERV_NESTED_VM_IMAGE_NAME = "image.vhdx"
@@ -81,7 +81,7 @@ def qemu_connect_nested_vm(
     )
 
     # setup connection to nested vm
-    connection_info = ConnectionInfo(
+    connection_info = schema.ConnectionInfo(
         address=host.public_address,
         port=guest_port,
         username=guest_username,
@@ -139,7 +139,7 @@ def hyperv_connect_nested_vm(
     hyperv.setup_port_forwarding(nat_name, port, local_ip)
 
     # setup connection to nested vm
-    connection_info = ConnectionInfo(
+    connection_info = schema.ConnectionInfo(
         address=host.public_address,
         port=port,
         username=guest_username,

microsoft/testsuites/performance/nestedperf.py

@@ -33,7 +33,7 @@
     Sysctl,
 )
 from lisa.util.logger import Logger
-from lisa.util.shell import ConnectionInfo, try_connect
+from lisa.util.shell import try_connect
 from microsoft.testsuites.nested.common import (
     HYPERV_NAT_NAME,
     NESTED_VM_REQUIRED_DISK_SIZE_IN_GB,
@@ -588,7 +588,7 @@ def _linux_setup_nat(
 
         # wait till nested vm is up
         try_connect(
-            ConnectionInfo(
+            schema.ConnectionInfo(
                 address=node.public_address,
                 port=guest_port,
                 username=guest_username,