[architect] refactor: single source of truth for SSH transport — smoke suite + shared/gnome_shell_steps onto ssh_config.ssh_argv()

tests/shared/gnome_shell_steps.py

@@ -9,6 +9,8 @@
 import subprocess
 from time import sleep
 
+from tests.shared.ssh_config import ssh_argv
+
 from behave import step
 from behave.runner import Context
 
@@ -19,15 +21,8 @@
 
 
 def _ssh_args() -> list[str]:
-    return [
-        "ssh",
-        "-i", os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519"),
-        "-o", "StrictHostKeyChecking=no",
-        "-o", "UserKnownHostsFile=/dev/null",
-        "-o", "ConnectTimeout=10",
-        "-p", os.environ.get("SSH_PORT", "22"),
-        f"{os.environ.get('VM_USER', 'bluefin-test')}@{os.environ.get('VM_IP', '127.0.0.1')}",
-    ]
+    """Canonical SSH argv — see tests/shared/ssh_config.py."""
+    return ssh_argv()
 
 
 def _shell_eval(js: str, timeout: int = 5) -> str:

tests/shared/ssh_config.py

@@ -5,6 +5,10 @@
 commands on the VM (e.g. ``_flatpak`` in the software suite), and by suite
 ``environment.py`` hooks that probe the VM directly.
 
+``ssh_argv(context)`` builds the canonical ``ssh`` argument vector from those
+details so that no caller has to restate the transport policy (host-key
+handling, connect timeout, port flag) inline.
+
 ``resolve_ssh_details(context)`` reads, in priority order:
 
 1. Behave ``context`` attributes (``ssh_key``, ``vm_ip``, ``ssh_user``,
@@ -32,12 +36,18 @@ def _first_value(*values: str) -> str:
     return ""
 
 
-def resolve_ssh_details(context) -> dict:
+def _userdata(context) -> dict:
+    """Return behave userdata, or an empty dict when there is no context."""
+    userdata = getattr(getattr(context, "config", None), "userdata", None)
+    return userdata if hasattr(userdata, "get") else {}
+
+
+def resolve_ssh_details(context=None) -> dict:
     """Return SSH connection details for the current run.
 
     Keys: ``ssh_key``, ``vm_ip``, ``ssh_user``, ``ssh_port`` (all strings).
     """
-    userdata = context.config.userdata
+    userdata = _userdata(context)
     return {
         "ssh_key": _first_value(
             getattr(context, "ssh_key", ""),
@@ -68,6 +78,25 @@ def resolve_ssh_details(context) -> dict:
     }
 
 
+def ssh_argv(context=None, *, connect_timeout: int = 10) -> list[str]:
+    """Return the canonical ``ssh`` argv prefix for the current run.
+
+    Callers append the remote command:  ``subprocess.run(ssh_argv() + [cmd])``.
+    This is the single place where SSH transport policy (host-key handling,
+    connect timeout, port flag, destination) is expressed.
+    """
+    details = resolve_ssh_details(context)
+    return [
+        "ssh",
+        "-i", details["ssh_key"],
+        "-o", "StrictHostKeyChecking=no",
+        "-o", "UserKnownHostsFile=/dev/null",
+        "-o", f"ConnectTimeout={connect_timeout}",
+        "-p", str(details["ssh_port"]),
+        f"{details['ssh_user']}@{details['vm_ip']}",
+    ]
+
+
 def populate_ssh_context(context) -> None:
     """Set the context attributes ``run_ssh`` requires.
 

tests/smoke/features/steps/app_support.py

@@ -3,6 +3,8 @@
 import shutil
 import subprocess
 
+from tests.shared.ssh_config import ssh_argv
+
 
 # When behave runs inside the runner container the host VM filesystem is not
 # visible: /usr/share/applications, flatpak, etc. are absent from the image.
@@ -22,15 +24,8 @@
 
 
 def _ssh_args() -> list[str]:
-    return [
-        "ssh",
-        "-i", os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519"),
-        "-o", "StrictHostKeyChecking=no",
-        "-o", "UserKnownHostsFile=/dev/null",
-        "-o", "ConnectTimeout=10",
-        "-p", os.environ.get("SSH_PORT", "22"),
-        f"{os.environ.get('VM_USER', 'bluefin-test')}@{os.environ.get('VM_IP', '127.0.0.1')}",
-    ]
+    """Canonical SSH argv — see tests/shared/ssh_config.py."""
+    return ssh_argv()
 
 
 def _ssh_run(cmd: str, timeout: int = 30) -> subprocess.CompletedProcess:

tests/smoke/features/steps/display_scaling_steps.py

@@ -13,6 +13,8 @@
 import subprocess
 from time import sleep
 
+from tests.shared.ssh_config import ssh_argv
+
 from behave import step
 
 try:
@@ -30,15 +32,8 @@
 
 
 def _ssh_args() -> list[str]:
-    return [
-        "ssh",
-        "-i", os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519"),
-        "-o", "StrictHostKeyChecking=no",
-        "-o", "UserKnownHostsFile=/dev/null",
-        "-o", "ConnectTimeout=10",
-        "-p", os.environ.get("SSH_PORT", "22"),
-        f"{os.environ.get('VM_USER', 'bluefin-test')}@{os.environ.get('VM_IP', '127.0.0.1')}",
-    ]
+    """Canonical SSH argv — see tests/shared/ssh_config.py."""
+    return ssh_argv()
 
 
 def _run_host(cmd: str, timeout: int = 30):

tests/smoke/features/steps/gnome_apps_steps.py

@@ -3,6 +3,8 @@
 import subprocess
 from time import sleep
 
+from tests.shared.ssh_config import ssh_argv
+
 from behave import step
 try:
     from dogtail import tree
@@ -138,15 +140,8 @@ def _shell_eval_force_close(app_names: tuple[str, ...]) -> None:
 
 
 def _ssh_args() -> list[str]:
-    return [
-        "ssh",
-        "-i", os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519"),
-        "-o", "StrictHostKeyChecking=no",
-        "-o", "UserKnownHostsFile=/dev/null",
-        "-o", "ConnectTimeout=10",
-        "-p", os.environ.get("SSH_PORT", "22"),
-        f"{os.environ.get('VM_USER', 'bluefin-test')}@{os.environ.get('VM_IP', '127.0.0.1')}",
-    ]
+    """Canonical SSH argv — see tests/shared/ssh_config.py."""
+    return ssh_argv()
 
 
 def _launch_app(app_id: str) -> None:

tests/smoke/features/steps/gnome_extensions_steps.py

@@ -7,6 +7,7 @@
 from time import sleep
 
 from behave import step
+from tests.shared.ssh_config import ssh_argv
 try:
     from dogtail import tree
 except Exception:  # noqa: BLE001
@@ -34,21 +35,8 @@ def _run_host(cmd: list[str] | str):
     import shlex
     cmd_str = cmd if isinstance(cmd, str) else " ".join(shlex.quote(a) for a in cmd)
     if _IN_CONTAINER:
-        ssh_key = os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519")
-        vm_ip = os.environ.get("VM_IP", "127.0.0.1")
-        vm_user = os.environ.get("VM_USER", "bluefin-test")
-        ssh_port = os.environ.get("SSH_PORT", "22")
         result = subprocess.run(
-            [
-                "ssh",
-                "-i", ssh_key,
-                "-o", "StrictHostKeyChecking=no",
-                "-o", "UserKnownHostsFile=/dev/null",
-                "-o", "ConnectTimeout=10",
-                "-p", ssh_port,
-                f"{vm_user}@{vm_ip}",
-                cmd_str,
-            ],
+            ssh_argv() + [cmd_str],
             capture_output=True, text=True, timeout=30, check=False,
         )
     else:
@@ -231,18 +219,12 @@ def launch_extensions_preferences_via_command(context) -> None:
     if _IN_CONTAINER:
         # Inside the runner container the desktop file is absent from the container
         # filesystem — launch via SSH on the VM where the session is running.
-        ssh_key = os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519")
-        vm_ip = os.environ.get("VM_IP", "127.0.0.1")
-        vm_user = os.environ.get("VM_USER", "bluefin-test")
-        ssh_port = os.environ.get("SSH_PORT", "22")
         cmd = (
             "source /tmp/session.env 2>/dev/null; "
             f"nohup gio launch {EXTENSIONS_DESKTOP_FILE} </dev/null &>/dev/null & disown"
         )
         result = subprocess.run(
-            ["ssh", "-i", ssh_key, "-o", "StrictHostKeyChecking=no",
-             "-o", "UserKnownHostsFile=/dev/null", "-o", "ConnectTimeout=10",
-             "-p", ssh_port, f"{vm_user}@{vm_ip}", cmd],
+            ssh_argv() + [cmd],
             capture_output=True, text=True, timeout=15,
         )
         if result.returncode != 0:

tests/smoke/features/steps/gnome_notifications_steps.py

@@ -5,6 +5,7 @@
 from time import sleep
 
 from behave import step
+from tests.shared.ssh_config import ssh_argv
 try:
     from qecore.common_steps import *  # noqa: F401,F403
 except Exception:  # noqa: BLE001
@@ -27,21 +28,8 @@ def _run(cmd: str):
 def _run_host(cmd: str):
     """Run cmd on the host VM via SSH when inside the runner container."""
     if _IN_CONTAINER:
-        ssh_key = os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519")
-        vm_ip = os.environ.get("VM_IP", "127.0.0.1")
-        vm_user = os.environ.get("VM_USER", "bluefin-test")
-        ssh_port = os.environ.get("SSH_PORT", "22")
         result = subprocess.run(
-            [
-                "ssh",
-                "-i", ssh_key,
-                "-o", "StrictHostKeyChecking=no",
-                "-o", "UserKnownHostsFile=/dev/null",
-                "-o", "ConnectTimeout=10",
-                "-p", ssh_port,
-                f"{vm_user}@{vm_ip}",
-                cmd,
-            ],
+            ssh_argv() + [cmd],
             capture_output=True, text=True, timeout=30,
         )
     else:

tests/smoke/features/steps/steps.py

@@ -28,6 +28,7 @@
 except Exception:  # noqa: BLE001
     pass
 from tests.shared.gnome_shell_steps import *  # noqa: F401,F403
+from tests.shared.ssh_config import ssh_argv
 
 # Same container detection as system_health_steps — /proc/1/ns/mnt is a symlink
 # to a kernel namespace object so lexists() is required (isfile() returns False).
@@ -37,21 +38,8 @@
 def _run_host(cmd: str, timeout: int = 30):
     """Run cmd on the host VM via SSH when inside the runner container."""
     if _IN_CONTAINER:
-        ssh_key = os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519")
-        vm_ip = os.environ.get("VM_IP", "127.0.0.1")
-        vm_user = os.environ.get("VM_USER", "bluefin-test")
-        ssh_port = os.environ.get("SSH_PORT", "22")
         result = subprocess.run(
-            [
-                "ssh",
-                "-i", ssh_key,
-                "-o", "StrictHostKeyChecking=no",
-                "-o", "UserKnownHostsFile=/dev/null",
-                "-o", "ConnectTimeout=10",
-                "-p", ssh_port,
-                f"{vm_user}@{vm_ip}",
-                cmd,
-            ],
+            ssh_argv() + [cmd],
             capture_output=True, text=True, timeout=timeout,
         )
     else:

tests/smoke/features/steps/system_health_steps.py

@@ -5,6 +5,7 @@
 import subprocess
 
 from behave import step
+from tests.shared.ssh_config import ssh_argv
 try:
     from qecore.common_steps import *  # noqa: F401,F403
 except Exception:  # noqa: BLE001
@@ -64,21 +65,8 @@ def _run_host(cmd: str, timeout: int = 30):
     (which requires host-level CAP_SYS_ADMIN that rootless podman cannot grant).
     """
     if _IN_CONTAINER:
-        ssh_key = os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519")
-        vm_ip = os.environ.get("VM_IP", "127.0.0.1")
-        vm_user = os.environ.get("VM_USER", "bluefin-test")
-        ssh_port = os.environ.get("SSH_PORT", "22")
         result = subprocess.run(
-            [
-                "ssh",
-                "-i", ssh_key,
-                "-o", "StrictHostKeyChecking=no",
-                "-o", "UserKnownHostsFile=/dev/null",
-                "-o", "ConnectTimeout=10",
-                "-p", ssh_port,
-                f"{vm_user}@{vm_ip}",
-                cmd,
-            ],
+            ssh_argv() + [cmd],
             capture_output=True, text=True, timeout=timeout,
         )
     else: