From 4e12b08459825d896d954a568e1eea0d1e0812a4 Mon Sep 17 00:00:00 2001
From: mrbobbytables <mrbobbytables@users.noreply.github.com>
Date: Sun, 6 Sep 2026 19:09:51 +0000
Subject: [PATCH 1/4] refactor(ssh): route dx, flatcar, kde-smoke, software,
 vanilla-gnome, image_cache, screenshot, and run_ssh onto
 ssh_config.ssh_argv()
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Phase 2 of #772: extract the remaining raw ssh argv builders onto the
single transport policy in tests/shared/ssh_config.py.

- Add ssh_argv(quiet=...) so callers that need LogLevel=ERROR (previously
  hand-rolled at each site) opt in explicitly instead of restating it.
- Fold ssh_steps.run_ssh onto ssh_argv(quiet=True); it now always resolves
  the port through the shared precedence chain (context > userdata > env >
  default), so unset ssh_port emits an explicit '-p 22' instead of silently
  relying on ssh's own default.
- Migrate tests/dx and tests/flatcar's raw ssh argv (both gain the -p flag,
  fixing silent port loss under port-forwarded lanes).
- Fix tests/kde-smoke/features/steps/steps.py:_run_host, which read a
  context.kde['ssh'] dict that before_all never populated — it always fell
  through to raw os.environ reads and silently ignored userdata-driven
  runs. It now resolves through ssh_argv(context), honouring the context
  attributes before_all does set.
- Migrate tests/software's _flatpak/_run_in_session/_has_bazaar, which
  already used resolve_ssh_details, onto ssh_argv() for a single argv
  builder instead of two.
- Migrate tests/shared/image_cache.py and tests/shared/screenshot.py.
  screenshot._ssh_run previously read only os.environ; it now resolves via
  the behave context bound by configure_screenshot_context.
- Fix tests/vanilla-gnome/features/steps/steps.py, which passed the
  command's own timeout as ssh's ConnectTimeout, conflating the connect
  deadline with the command deadline (a 900s long-command timeout became a
  900s connect timeout). It now uses ssh_argv()'s fixed 10s connect timeout
  and threads context through so future context-based resolution works.
- Widen tests/unit/test_ssh_transport_contract.py's MIGRATED_MODULES to
  cover all of the above, so any of them re-growing a private ssh argv
  copy fails CI.

tests/smoke/features/steps/offline_boot_steps.py remains unmigrated per
the issue's Phase 1 exclusion (PR #768 covers its unit tests).

Fixes #772

Signed-off-by: mrbobbytables <mrbobbytables@users.noreply.github.com>
---
 tests/dx/features/steps/steps.py            | 13 ++----
 tests/flatcar/features/steps/steps.py       | 12 +----
 tests/kde-smoke/features/steps/steps.py     | 25 +++-------
 tests/shared/image_cache.py                 | 17 ++-----
 tests/shared/screenshot.py                  | 23 ++++------
 tests/shared/ssh_config.py                  | 13 +++++-
 tests/shared/ssh_steps.py                   | 21 ++-------
 tests/software/features/environment.py      | 13 +-----
 tests/software/features/steps/steps.py      | 15 ++----
 tests/unit/test_ssh_config.py               | 13 ++----
 tests/unit/test_ssh_steps.py                | 14 +++++-
 tests/unit/test_ssh_transport_contract.py   | 21 +++++++++
 tests/unit/test_vanilla_gnome_steps.py      |  2 +-
 tests/vanilla-gnome/features/steps/steps.py | 51 +++++++++++----------
 14 files changed, 108 insertions(+), 145 deletions(-)

diff --git a/tests/dx/features/steps/steps.py b/tests/dx/features/steps/steps.py
index 141f69bc7..9bb9620af 100644
--- a/tests/dx/features/steps/steps.py
+++ b/tests/dx/features/steps/steps.py
@@ -14,20 +14,13 @@
 from behave import step
 from qecore.common_steps import *  # noqa: F401,F403
 
+from tests.shared.ssh_config import ssh_argv
+
 
 def _ssh(context, cmd, timeout=60):
     """Run a command on the DX VM over SSH and record stdout + return code."""
     result = subprocess.run(
-        [
-            "ssh",
-            "-i", context.ssh_key,
-            "-o", "StrictHostKeyChecking=no",
-            "-o", "UserKnownHostsFile=/dev/null",
-            "-o", "ConnectTimeout=10",
-            "-o", "LogLevel=ERROR",
-            f"{context.ssh_user}@{context.vm_ip}",
-            cmd,
-        ],
+        ssh_argv(context, quiet=True) + [cmd],
         capture_output=True,
         text=True,
         timeout=timeout,
diff --git a/tests/flatcar/features/steps/steps.py b/tests/flatcar/features/steps/steps.py
index 640cea2c6..640c6eef4 100644
--- a/tests/flatcar/features/steps/steps.py
+++ b/tests/flatcar/features/steps/steps.py
@@ -11,6 +11,7 @@
 
 from behave import step
 
+from tests.shared.ssh_config import ssh_argv
 from tests.shared.ssh_steps import *  # noqa: F401,F403,F405
 from tests.shared.ssh_steps import run_ssh, ssh_output_is, ssh_return_code_is  # noqa: F401
 
@@ -102,16 +103,7 @@ def reboot_vm_from_target_disk(context) -> None:
     )
     try:
         subprocess.run(
-            [
-                "ssh",
-                "-i", context.ssh_key,
-                "-o", "StrictHostKeyChecking=no",
-                "-o", "UserKnownHostsFile=/dev/null",
-                "-o", "ConnectTimeout=10",
-                "-o", "LogLevel=ERROR",
-                f"{context.ssh_user}@{context.vm_ip}",
-                reboot_command,
-            ],
+            ssh_argv(context, quiet=True) + [reboot_command],
             capture_output=True,
             text=True,
             timeout=15,
diff --git a/tests/kde-smoke/features/steps/steps.py b/tests/kde-smoke/features/steps/steps.py
index 60cdc3820..2a1d2f4c2 100644
--- a/tests/kde-smoke/features/steps/steps.py
+++ b/tests/kde-smoke/features/steps/steps.py
@@ -14,6 +14,7 @@
 import shlex
 import subprocess
 from behave import step
+from tests.shared.ssh_config import ssh_argv
 from tests.shared.ssh_steps import run_ssh
 
 from tests.shared.kde_shell_steps import wait_until as _shared_wait_until
@@ -61,25 +62,13 @@ def _run(cmd: str, timeout: int = 30):
 def _run_host(cmd: str, timeout: int = 30, context=None):
     """Run cmd on the host VM via SSH when inside the runner container."""
     if _IN_CONTAINER:
-        # Prefer the connection settings resolved in before_all (which honour
-        # behave -D userdata); fall back to the environment. Reading env only
-        # meant userdata-configured runs probed the wrong host.
-        conn = getattr(context, "kde", {}).get("ssh", {}) if context is not None else {}
-        ssh_key = conn.get("key") or os.environ.get("SSH_KEY", "/home/bluefin-test/.ssh/id_ed25519")
-        vm_ip = conn.get("ip") or os.environ.get("VM_IP", "127.0.0.1")
-        vm_user = conn.get("user") or os.environ.get("VM_USER", "bluefin-test")
-        ssh_port = str(conn.get("port") or os.environ.get("SSH_PORT", "22"))
+        # ssh_argv() resolves through context attributes (set in before_all,
+        # which honours behave -D userdata) before falling back to the
+        # environment. Previously this read a never-populated
+        # ``context.kde["ssh"]`` dict, so it always fell through to raw env
+        # reads and silently ignored userdata-configured runs.
         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(context) + [cmd],
             capture_output=True, text=True, timeout=timeout,
         )
     else:
diff --git a/tests/shared/image_cache.py b/tests/shared/image_cache.py
index 1201d8636..ec39f7528 100644
--- a/tests/shared/image_cache.py
+++ b/tests/shared/image_cache.py
@@ -89,20 +89,9 @@ def _ssh_returncode(context, command: str, timeout: int) -> int:
     Unlike a step, this probe leaves ``context`` untouched: it runs before the
     per-scenario state reset and must not smear command output across scenarios.
     """
-    from tests.shared.ssh_config import resolve_ssh_details
-
-    details = resolve_ssh_details(context)
-    argv = [
-        "ssh",
-        "-i", details["ssh_key"],
-        "-o", "StrictHostKeyChecking=no",
-        "-o", "UserKnownHostsFile=/dev/null",
-        "-o", "ConnectTimeout=10",
-        "-o", "LogLevel=ERROR",
-    ]
-    if details.get("ssh_port"):
-        argv += ["-p", str(details["ssh_port"])]
-    argv += [f"{details['ssh_user']}@{details['vm_ip']}", command]
+    from tests.shared.ssh_config import ssh_argv
+
+    argv = ssh_argv(context, quiet=True) + [command]
     return subprocess.run(argv, capture_output=True, text=True, timeout=timeout).returncode
 
 
diff --git a/tests/shared/screenshot.py b/tests/shared/screenshot.py
index 0f873b563..90cc528bd 100644
--- a/tests/shared/screenshot.py
+++ b/tests/shared/screenshot.py
@@ -56,21 +56,16 @@ def _screenshot_path(label: str, context: Any | None = None) -> str:
 
 
 def _ssh_run(cmd: str, timeout: int = 15) -> "subprocess.CompletedProcess[str]":
-    """Run a shell command on the VM via SSH (used when inside the runner 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")
+    """Run a shell command on the VM via SSH (used when inside the runner container).
+
+    Resolves connection details via ``tests.shared.ssh_config.ssh_argv``, which
+    honours the bound behave context (``configure_screenshot_context``) before
+    falling back to environment variables.
+    """
+    from tests.shared.ssh_config import ssh_argv
+
     return 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(_CURRENT_CONTEXT) + [cmd],
         capture_output=True, text=True, timeout=timeout,
     )
 
diff --git a/tests/shared/ssh_config.py b/tests/shared/ssh_config.py
index 76accb205..d23b3938f 100644
--- a/tests/shared/ssh_config.py
+++ b/tests/shared/ssh_config.py
@@ -78,23 +78,32 @@ def resolve_ssh_details(context=None) -> dict:
     }
 
 
-def ssh_argv(context=None, *, connect_timeout: int = 10) -> list[str]:
+def ssh_argv(context=None, *, connect_timeout: int = 10, quiet: bool = False) -> 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.
+
+    ``quiet=True`` adds ``LogLevel=ERROR`` so ssh's own diagnostics do not leak
+    into captured command output — set it for callers that previously built
+    that option inline instead of restating it at every call site.
     """
     details = resolve_ssh_details(context)
-    return [
+    argv = [
         "ssh",
         "-i", details["ssh_key"],
         "-o", "StrictHostKeyChecking=no",
         "-o", "UserKnownHostsFile=/dev/null",
         "-o", f"ConnectTimeout={connect_timeout}",
+    ]
+    if quiet:
+        argv += ["-o", "LogLevel=ERROR"]
+    argv += [
         "-p", str(details["ssh_port"]),
         f"{details['ssh_user']}@{details['vm_ip']}",
     ]
+    return argv
 
 
 def populate_ssh_context(context) -> None:
diff --git a/tests/shared/ssh_steps.py b/tests/shared/ssh_steps.py
index cbfe7134d..edb74f6a1 100644
--- a/tests/shared/ssh_steps.py
+++ b/tests/shared/ssh_steps.py
@@ -12,6 +12,8 @@
 
 from behave import step
 
+from tests.shared.ssh_config import ssh_argv
+
 
 def _is_local_target(context) -> bool:
     """Return True if tests are executing directly on the target host/container."""
@@ -29,7 +31,6 @@ def run_ssh(context, cmd, timeout=60):
     final_cmd = cmd
     if prefix:
         final_cmd = f"bash -c {shlex.quote(f'{prefix}; {cmd}')}"
-
     if _is_local_target(context):
         try:
             result = subprocess.run(
@@ -49,23 +50,7 @@ def run_ssh(context, cmd, timeout=60):
         context.last_ssh_result = result
         return stdout, result.returncode
 
-    ssh_opts = [
-        "ssh",
-        "-i",
-        context.ssh_key,
-        "-o",
-        "StrictHostKeyChecking=no",
-        "-o",
-        "UserKnownHostsFile=/dev/null",
-        "-o",
-        "ConnectTimeout=10",
-        "-o",
-        "LogLevel=ERROR",
-    ]
-    if getattr(context, "ssh_port", None):
-        ssh_opts += ["-p", str(context.ssh_port)]
-    ssh_opts.append(f"{context.ssh_user}@{context.vm_ip}")
-    ssh_opts.append(final_cmd)
+    ssh_opts = ssh_argv(context, quiet=True) + [final_cmd]
     try:
         result = subprocess.run(ssh_opts, capture_output=True, text=True, timeout=timeout)
     except subprocess.TimeoutExpired:
diff --git a/tests/software/features/environment.py b/tests/software/features/environment.py
index 5cd627954..bebbaf4b4 100644
--- a/tests/software/features/environment.py
+++ b/tests/software/features/environment.py
@@ -8,7 +8,7 @@
 from qecore.sandbox import TestSandbox
 from qecore.common_steps import *  # noqa: F401,F403
 
-from tests.shared.ssh_config import populate_ssh_context, resolve_ssh_details
+from tests.shared.ssh_config import populate_ssh_context, ssh_argv
 
 try:
     from tests.shared.timing import record_end, record_start
@@ -50,16 +50,7 @@ def take_fastfetch_screenshot():
 def _has_bazaar(context) -> bool:
     """Return True when Bazaar (io.github.kolunmi.Bazaar) is installed on the VM."""
     import subprocess
-    ssh = resolve_ssh_details(context)
-    ssh_args = [
-        'ssh',
-        '-i', ssh['ssh_key'],
-        '-o', 'StrictHostKeyChecking=no',
-        '-o', 'UserKnownHostsFile=/dev/null',
-        '-o', 'ConnectTimeout=10',
-        '-o', 'LogLevel=ERROR',
-        '-p', ssh['ssh_port'],
-        f"{ssh['ssh_user']}@{ssh['vm_ip']}",
+    ssh_args = ssh_argv(context, quiet=True) + [
         'flatpak list --app --columns=application 2>/dev/null | grep -q io.github.kolunmi.Bazaar',
     ]
     try:
diff --git a/tests/software/features/steps/steps.py b/tests/software/features/steps/steps.py
index ad25218d0..c42ce7daf 100644
--- a/tests/software/features/steps/steps.py
+++ b/tests/software/features/steps/steps.py
@@ -11,7 +11,7 @@
 except Exception:  # noqa: BLE001
     tree = None  # type: ignore[assignment]
 from qecore.common_steps import *  # noqa: F401,F403
-from tests.shared.ssh_config import resolve_ssh_details
+from tests.shared.ssh_config import ssh_argv
 from tests.shared.ssh_steps import *  # noqa: F401,F403
 from tests.smoke.features.steps.app_support import atspi_click, launch_background
 
@@ -48,16 +48,12 @@ def _flatpak(context, args: list[str], timeout: int = 10) -> subprocess.Complete
     """Run flatpak via SSH when inside the runner container.
 
     Connection details come from the same source as the shared SSH steps
-    (``tests.shared.ssh_config.resolve_ssh_details``): context attributes,
+    (``tests.shared.ssh_config.ssh_argv``): context attributes,
     then behave userdata, then environment variables.
     """
     if _IN_CONTAINER:
-        ssh = resolve_ssh_details(context)
         return subprocess.run(
-            ["ssh", "-i", ssh["ssh_key"], "-o", "StrictHostKeyChecking=no",
-             "-o", "UserKnownHostsFile=/dev/null", "-o", "ConnectTimeout=10",
-             "-p", ssh["ssh_port"], f"{ssh['ssh_user']}@{ssh['vm_ip']}",
-             " ".join(["flatpak"] + [f"'{a}'" for a in args])],
+            ssh_argv(context) + [" ".join(["flatpak"] + [f"'{a}'" for a in args])],
             capture_output=True, text=True, timeout=timeout,
         )
     return subprocess.run(
@@ -75,11 +71,8 @@ def _run_in_session(context, cmd: str, timeout: int = 15) -> subprocess.Complete
     """
     full = f"source /tmp/session.env 2>/dev/null; {cmd}"
     if _IN_CONTAINER:
-        ssh = resolve_ssh_details(context)
         return subprocess.run(
-            ["ssh", "-i", ssh["ssh_key"], "-o", "StrictHostKeyChecking=no",
-             "-o", "UserKnownHostsFile=/dev/null", "-o", "ConnectTimeout=10",
-             "-p", ssh["ssh_port"], f"{ssh['ssh_user']}@{ssh['vm_ip']}", full],
+            ssh_argv(context) + [full],
             capture_output=True, text=True, timeout=timeout,
         )
     return subprocess.run(
diff --git a/tests/unit/test_ssh_config.py b/tests/unit/test_ssh_config.py
index d460565f5..c88748f8b 100644
--- a/tests/unit/test_ssh_config.py
+++ b/tests/unit/test_ssh_config.py
@@ -137,18 +137,13 @@ def test_before_all_values_come_from_environment(self):
     def test_bazaar_probe_uses_shared_connection_details(self):
         env_mod = self._import_software_environment()
         ctx = _full_context()
-        details = {
-            "ssh_key": "/resolved/key",
-            "vm_ip": "192.0.2.40",
-            "ssh_user": "resolved-user",
-            "ssh_port": "2224",
-        }
+        argv = ["ssh", "-i", "/resolved/key", "-p", "2224", "resolved-user@192.0.2.40"]
         result = types.SimpleNamespace(returncode=0)
-        with patch.object(env_mod, "resolve_ssh_details", return_value=details) as resolve, \
+        with patch.object(env_mod, "ssh_argv", return_value=argv) as build_argv, \
              patch("subprocess.run", return_value=result) as run:
             assert env_mod._has_bazaar(ctx)
 
-        resolve.assert_called_once_with(ctx)
+        build_argv.assert_called_once_with(ctx, quiet=True)
         command = run.call_args.args[0]
         assert "/resolved/key" in command
         assert "2224" in command
@@ -176,5 +171,5 @@ def test_steps_module_resolves_via_shared_source(self):
         import inspect
         steps_mod = _import_software_steps()
         src = inspect.getsource(steps_mod._flatpak)
-        assert "resolve_ssh_details" in src
+        assert "ssh_argv" in src
         assert "os.environ" not in src
diff --git a/tests/unit/test_ssh_steps.py b/tests/unit/test_ssh_steps.py
index d049276bb..b2dcc3c86 100644
--- a/tests/unit/test_ssh_steps.py
+++ b/tests/unit/test_ssh_steps.py
@@ -4,6 +4,7 @@
 context attribute setting) using subprocess mocks. No live SSH required.
 """
 
+import types
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -25,6 +26,11 @@ def _make_context(*, ssh_key="/tmp/test.key", ssh_user="bluefin-test",
     ctx.last_command_output = ""
     ctx.ssh_rc = None
     ctx.last_ssh_result = None
+    # run_ssh resolves connection details via tests.shared.ssh_config, which
+    # reads context.config.userdata (a real dict) after context attributes —
+    # a bare MagicMock here would auto-fabricate a truthy userdata.get(...)
+    # result and corrupt resolution.
+    ctx.config = types.SimpleNamespace(userdata={})
     return ctx
 
 
@@ -154,13 +160,17 @@ def test_ssh_port_included_when_set(self):
         assert "-p" in call_args
         assert "2222" in call_args
 
-    def test_no_port_flag_when_ssh_port_is_none(self):
+    def test_default_port_used_when_ssh_port_is_none(self):
+        """ssh_argv() always emits an explicit -p; unset context.ssh_port
+        resolves to ssh_config.DEFAULT_SSH_PORT ("22"), matching the ssh
+        default rather than omitting the flag."""
         ctx = _make_context(ssh_port=None)
         proc = _make_proc(stdout="ok\n")
         with patch("subprocess.run", return_value=proc) as mock_run:
             self.mod.run_ssh(ctx, "echo ok")
         call_args = mock_run.call_args[0][0]
-        assert "-p" not in call_args
+        assert "-p" in call_args
+        assert call_args[call_args.index("-p") + 1] == "22"
 
     def test_target_host_in_command(self):
         ctx = _make_context(ssh_user="testuser", vm_ip="10.0.0.5")
diff --git a/tests/unit/test_ssh_transport_contract.py b/tests/unit/test_ssh_transport_contract.py
index 05a531595..21fc89648 100644
--- a/tests/unit/test_ssh_transport_contract.py
+++ b/tests/unit/test_ssh_transport_contract.py
@@ -19,6 +19,9 @@
 # regression: the transport policy must stay in tests/shared/ssh_config.py.
 MIGRATED_MODULES = (
     "tests/shared/gnome_shell_steps.py",
+    "tests/shared/ssh_steps.py",
+    "tests/shared/image_cache.py",
+    "tests/shared/screenshot.py",
     "tests/smoke/features/steps/app_support.py",
     "tests/smoke/features/steps/display_scaling_steps.py",
     "tests/smoke/features/steps/gnome_apps_steps.py",
@@ -26,6 +29,12 @@
     "tests/smoke/features/steps/gnome_notifications_steps.py",
     "tests/smoke/features/steps/steps.py",
     "tests/smoke/features/steps/system_health_steps.py",
+    "tests/dx/features/steps/steps.py",
+    "tests/flatcar/features/steps/steps.py",
+    "tests/kde-smoke/features/steps/steps.py",
+    "tests/software/features/steps/steps.py",
+    "tests/software/features/environment.py",
+    "tests/vanilla-gnome/features/steps/steps.py",
 )
 
 
@@ -86,6 +95,18 @@ def test_resolves_without_a_context(self):
         with patch.dict(os.environ, _clean_env(VM_IP="10.0.0.3"), clear=True):
             assert "bluefin-test@10.0.0.3" in ssh_config.ssh_argv(None)
 
+    def test_quiet_defaults_off(self):
+        """Callers that never asked for LogLevel=ERROR keep byte-identical argv."""
+        with patch.dict(os.environ, _clean_env(), clear=True):
+            assert "LogLevel=ERROR" not in ssh_config.ssh_argv()
+
+    def test_quiet_adds_log_level_error(self):
+        """quiet=True is the single opt-in for suppressing ssh's own diagnostics."""
+        with patch.dict(os.environ, _clean_env(), clear=True):
+            argv = ssh_config.ssh_argv(quiet=True)
+        assert "LogLevel=ERROR" in argv
+        assert argv[0] == "ssh"
+
 
 class TestNoPrivateTransportCopies:
     def test_migrated_modules_do_not_rebuild_ssh_argv(self):
diff --git a/tests/unit/test_vanilla_gnome_steps.py b/tests/unit/test_vanilla_gnome_steps.py
index 4b329dc03..94198048f 100644
--- a/tests/unit/test_vanilla_gnome_steps.py
+++ b/tests/unit/test_vanilla_gnome_steps.py
@@ -123,7 +123,7 @@ def test_gnome_files_checks_gnome_files_and_nautilus(self):
         """PR #388: gnome-files added as fallback alongside nautilus."""
         m = _import_vanilla_gnome_steps()
         checked_commands = []
-        def track_cmd(cmd):
+        def track_cmd(cmd, context=None):
             checked_commands.append(cmd)
             return cmd == "gnome-files"
         with patch.object(m, "_command_exists", side_effect=track_cmd), \
diff --git a/tests/vanilla-gnome/features/steps/steps.py b/tests/vanilla-gnome/features/steps/steps.py
index 2655ece32..88c1dc86b 100644
--- a/tests/vanilla-gnome/features/steps/steps.py
+++ b/tests/vanilla-gnome/features/steps/steps.py
@@ -25,6 +25,7 @@
 from qecore.common_steps import *  # noqa: F401,F403
 from tests.shared.gnome_shell_steps import *  # noqa: F401,F403
 from tests.shared.gnome_shell_steps import _shell_eval, _eval_bool, _wait_eval_bool
+from tests.shared.ssh_config import ssh_argv
 
 
 # ── Shell.Eval helpers (GNOME 50: uinput Super + AT-SPI toggle click broken) ──
@@ -133,36 +134,32 @@ def overview_search_bar_contains(context, text) -> None:
     assert text in entry.text, f"Search bar text '{entry.text}' does not contain '{text}'"
 
 
-def _ssh_run(cmd: str, timeout: int = 15) -> subprocess.CompletedProcess:
-    """Run a command on the VM via SSH using the standard connection env vars."""
-    import os
-    ssh_args = [
-        'ssh',
-        '-i', os.environ.get('SSH_KEY', '/home/bluefin-test/.ssh/id_ed25519'),
-        '-o', 'StrictHostKeyChecking=no',
-        '-o', 'UserKnownHostsFile=/dev/null',
-        '-o', f"ConnectTimeout={timeout}",
-        '-o', 'LogLevel=ERROR',
-        '-p', os.environ.get('SSH_PORT', '22'),
-        f"{os.environ.get('VM_USER', 'bluefin-test')}@{os.environ.get('VM_IP', '127.0.0.1')}",
-        cmd,
-    ]
-    return subprocess.run(ssh_args, capture_output=True, text=True, timeout=timeout)
-
-
-def _command_exists(command: str) -> bool:
+def _ssh_run(cmd: str, timeout: int = 15, context=None) -> subprocess.CompletedProcess:
+    """Run a command on the VM via SSH.
+
+    ``timeout`` bounds only the local wait for the command to finish; the SSH
+    *connect* deadline is fixed at ``ssh_argv``'s default (10s) so a long
+    command timeout no longer inflates how long a dead connection is retried.
+    """
+    return subprocess.run(
+        ssh_argv(context, quiet=True) + [cmd],
+        capture_output=True, text=True, timeout=timeout,
+    )
+
+
+def _command_exists(command: str, context=None) -> bool:
     """Check whether a command is available on the VM."""
     try:
-        result = _ssh_run(f'command -v {command}')
+        result = _ssh_run(f'command -v {command}', context=context)
     except (FileNotFoundError, subprocess.TimeoutExpired):
         return False
     return result.returncode == 0 and bool(result.stdout.strip())
 
 
-def _flatpak_app_exists(app_id: str) -> bool:
+def _flatpak_app_exists(app_id: str, context=None) -> bool:
     """Check whether a Flatpak app is installed on the VM."""
     try:
-        result = _ssh_run('flatpak list --app --columns=application 2>/dev/null', timeout=20)
+        result = _ssh_run('flatpak list --app --columns=application 2>/dev/null', timeout=20, context=context)
     except (FileNotFoundError, subprocess.TimeoutExpired):
         return False
     if result.returncode != 0:
@@ -171,9 +168,9 @@ def _flatpak_app_exists(app_id: str) -> bool:
     return app_id in installed
 
 
-def _assert_any_app_present(label: str, commands: tuple[str, ...], flatpaks: tuple[str, ...]) -> None:
-    found_commands = [command for command in commands if _command_exists(command)]
-    found_flatpaks = [app_id for app_id in flatpaks if _flatpak_app_exists(app_id)]
+def _assert_any_app_present(label: str, commands: tuple[str, ...], flatpaks: tuple[str, ...], context=None) -> None:
+    found_commands = [command for command in commands if _command_exists(command, context=context)]
+    found_flatpaks = [app_id for app_id in flatpaks if _flatpak_app_exists(app_id, context=context)]
     assert found_commands or found_flatpaks, (
         f"{label} not found. Commands checked: {commands}; flatpaks checked: {flatpaks}"
     )
@@ -209,6 +206,7 @@ def files_application_is_installed(context) -> None:
         # 'gnome-files' is the binary name since GNOME 47; 'nautilus' is the classic name
         ('gnome-files', 'nautilus'),
         ('org.gnome.Nautilus',),
+        context=context,
     )
 
 
@@ -219,6 +217,7 @@ def text_editor_application_is_installed(context) -> None:
         # gnome-text-editor (GNOME 42+); gedit (classic fallback)
         ('gnome-text-editor', 'gedit'),
         ('org.gnome.TextEditor', 'org.gnome.gedit'),
+        context=context,
     )
 
 
@@ -228,6 +227,7 @@ def web_browser_application_is_installed(context) -> None:
         'Web browser application',
         ('epiphany', 'firefox', 'chromium', 'chromium-browser'),
         ('org.gnome.Epiphany', 'org.mozilla.firefox', 'org.chromium.Chromium'),
+        context=context,
     )
 
 
@@ -237,6 +237,7 @@ def terminal_application_is_installed(context) -> None:
         'Terminal application',
         ('kgx', 'ptyxis', 'gnome-terminal'),
         ('org.gnome.Console', 'app.devsuite.Ptyxis', 'org.gnome.Terminal'),
+        context=context,
     )
 
 
@@ -279,7 +280,7 @@ def close_screenshot_tool(context) -> None:
 @step('Bluefin-specific extensions are absent on vanilla-gnome')
 def bluefin_extensions_absent(context) -> None:
     try:
-        result = _ssh_run("gnome-extensions list")
+        result = _ssh_run("gnome-extensions list", context=context)
     except (FileNotFoundError, subprocess.TimeoutExpired):
         return
 

From 47dfa7c83a67749edbcdb76df79f265f0c2b6785 Mon Sep 17 00:00:00 2001
From: v <a@l>
Date: Fri, 18 Sep 2026 13:41:20 -0400
Subject: [PATCH 2/4] docs(skills): record the ssh_argv transport contract

AGENTS.md:146 wants a matching docs/skills/** update when a change alters
a contract. This PR changes one across five suites - `-p` is now always
emitted and `quiet` is opt-in per call site - with no note. Records both
properties and the MIGRATED_MODULES guard in the shared-ssh reference.

Assisted-by: Claude Opus 4.5 via pi
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../behave/references/shared-ssh.md              | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/docs/skills/test-authoring/behave/references/shared-ssh.md b/docs/skills/test-authoring/behave/references/shared-ssh.md
index 5de5a0e94..2d997e572 100644
--- a/docs/skills/test-authoring/behave/references/shared-ssh.md
+++ b/docs/skills/test-authoring/behave/references/shared-ssh.md
@@ -100,3 +100,19 @@ When a scenario is meant to fail on a bad command, never append `; true` (or
 similar success-forcing trailers) to the SSH command. That masks the real exit
 status and turns `SSH command return code is "0"` into a no-op. Use `2>&1` to
 capture diagnostics, but preserve the original command's exit code.
+
+## `ssh_argv()` owns the SSH argv for every suite
+
+`tests/shared/ssh_config.ssh_argv(context, *, connect_timeout=10, quiet=False)` is the
+single builder for SSH argv. Two properties of that contract are easy to get wrong:
+
+- **`-p` is always emitted**, including when `SSH_PORT` is unset (it falls back to `22`).
+  Do not write a test asserting the port flag is absent.
+- **`quiet=True` adds `-o LogLevel=ERROR`, and it is opt-in per call site.** Only pass it
+  where the old hand-rolled argv already suppressed the banner (`run_ssh`, `image_cache`,
+  the software `_has_bazaar` probe, dx/flatcar/vanilla-gnome). Passing it everywhere hides
+  diagnostics that kde-smoke and the screenshot helper rely on.
+
+Suites must not hand-roll `ssh` argv. `tests/unit/test_ssh_transport_contract.py`
+enforces this for the modules listed in its `MIGRATED_MODULES` set — add new suites there
+rather than copying an argv list.

From 92b38a62bf14aae3a818323c86d36c349d7b192d Mon Sep 17 00:00:00 2001
From: "sec-check[bot]" <sec-check[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 01:50:45 -0400
Subject: [PATCH 3/4] fix(ssh): preserve kde-smoke default key, read
 TMT_SSH_PORT, refresh contract docs
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review follow-ups for the Phase 2 ssh_argv() routing:

- kde-smoke before_all no longer hardcodes an always-truthy TMT_SSH_KEY
  default. With no key env var set, _default_ssh_key() prefers the
  tmt-provisioned /etc/ssh/test-key/id_ed25519 when it exists and otherwise
  falls back to the runner key /home/bluefin-test/.ssh/id_ed25519 — the path
  _run_host used before it moved onto ssh_argv(), so env-default local runs
  keep authenticating.
- resolve_ssh_details() now reads TMT_SSH_PORT after SSH_PORT/VM_PORT, so
  tmt-provisioned lanes (dx, flatcar) on a forwarded port no longer connect to
  22. Those suites resolve host/user/key from TMT_SSH_* and never set
  context.ssh_port.
- populate_ssh_context() docstring and the behave SSH docs now describe the
  real post-refactor contract: skipping the call no longer raises
  AttributeError, it silently falls back to default credentials.

Unit tests cover both new behaviours.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 docs/skills/test-authoring/behave/SKILL.md    |  2 +-
 .../behave/references/shared-ssh.md           | 12 +++++----
 tests/kde-smoke/features/environment.py       | 19 ++++++++++++--
 tests/shared/ssh_config.py                    | 16 +++++++++---
 tests/unit/test_kde_smoke_environment.py      | 25 +++++++++++++++++++
 tests/unit/test_ssh_config.py                 | 16 +++++++++++-
 6 files changed, 77 insertions(+), 13 deletions(-)

diff --git a/docs/skills/test-authoring/behave/SKILL.md b/docs/skills/test-authoring/behave/SKILL.md
index 0c081fe32..3500ca1a1 100644
--- a/docs/skills/test-authoring/behave/SKILL.md
+++ b/docs/skills/test-authoring/behave/SKILL.md
@@ -39,7 +39,7 @@ metadata:
 
 1. Read the target `.feature` file and the suite's `steps/*.py` before adding phrases.
 2. Reuse `tests/shared/ssh_steps.py` for generic SSH command/assertion steps instead of duplicating helpers.
-3. Star-importing `tests/shared/ssh_steps` obligates the suite to set the context attributes those steps read. `run_ssh()` dereferences `context.ssh_key`, `context.ssh_user`, `context.vm_ip` and (optionally) `context.ssh_port`; a suite that skips this fails every SSH scenario with `AttributeError` on the first step. Call `tests.shared.ssh_config.populate_ssh_context(context)` from the suite's `before_all` (see `tests/software/features/environment.py`) — it resolves context attributes → behave userdata → `SSH_KEY`/`VM_IP`/`VM_USER`/`SSH_PORT` env vars → runner defaults, which is the same source suite-local SSH helpers (e.g. the software suite's `_flatpak()`) must use. Never let a suite keep a second, env-only SSH path alongside the shared steps.
+3. Star-importing `tests/shared/ssh_steps` obligates the suite to set the context attributes those steps read. `run_ssh()` builds its argv via `ssh_config.ssh_argv(context)`, which prefers `context.ssh_key`, `context.ssh_user`, `context.vm_ip`, `context.ssh_port` and then falls back to userdata/env/defaults; a suite that skips this no longer fails loudly — it silently runs against default credentials. Call `tests.shared.ssh_config.populate_ssh_context(context)` from the suite's `before_all` (see `tests/software/features/environment.py`) — it resolves context attributes → behave userdata → `SSH_KEY`/`VM_IP`/`VM_USER`/`SSH_PORT`/`TMT_SSH_PORT` env vars → runner defaults, which is the same source suite-local SSH helpers (e.g. the software suite's `_flatpak()`) must use. Never let a suite keep a second, env-only SSH path alongside the shared steps.
 4. Keep step phrases unique within the loaded suite and check for collisions before committing.
 5. Choose assertions that match the command shape: equality for single-line output, substring for multiline output.
 6. Run `behave --dry-run` on the touched suite before pushing so undefined or ambiguous phrases fail locally.
diff --git a/docs/skills/test-authoring/behave/references/shared-ssh.md b/docs/skills/test-authoring/behave/references/shared-ssh.md
index 2d997e572..37be73904 100644
--- a/docs/skills/test-authoring/behave/references/shared-ssh.md
+++ b/docs/skills/test-authoring/behave/references/shared-ssh.md
@@ -38,11 +38,13 @@ from tests.shared.ssh_steps import *  # noqa: F401,F403
 
 ## Importing the steps is only half the contract
 
-`run_ssh()` reads its connection details from **`context`**, not from the
-environment. A suite that star-imports `ssh_steps` must also populate
-`context.vm_ip`, `context.ssh_user`, `context.ssh_key` and optionally
-`context.ssh_port` in `before_all`, or every SSH step raises `AttributeError`
-at runtime.
+`run_ssh()` resolves its connection details through
+`ssh_config.ssh_argv(context)`, which prefers **`context`** attributes
+(`vm_ip`, `ssh_user`, `ssh_key`, `ssh_port`) and only then falls back to behave
+userdata, environment variables and runner defaults. A suite that star-imports
+`ssh_steps` without populating those attributes no longer raises
+`AttributeError` — it silently connects with *default* credentials, which is
+worse. Populate them in `before_all`.
 
 Resolve them through the shared helper rather than hand-rolling per suite:
 
diff --git a/tests/kde-smoke/features/environment.py b/tests/kde-smoke/features/environment.py
index 36e4ca369..ebf668590 100644
--- a/tests/kde-smoke/features/environment.py
+++ b/tests/kde-smoke/features/environment.py
@@ -16,6 +16,7 @@
 import traceback
 
 from tests.shared.ssh_steps import *  # noqa: F401,F403 — register shared SSH steps
+from tests.shared.ssh_config import DEFAULT_SSH_KEY
 from tests.shared.timing import record_end, record_start
 from tests.shared.kde_faillog import collect_on_failure
 from tests.shared.kde_preconditions import (
@@ -31,6 +32,20 @@
 # Session environment file injected by the e2e runner.
 _SESSION_ENV_FILE = "/tmp/session.env"
 
+# Key provisioned by tmt inside the runner container.
+_TMT_SSH_KEY = "/etc/ssh/test-key/id_ed25519"
+
+
+def _default_ssh_key() -> str:
+    """Key path to use when no key is configured by userdata or environment.
+
+    Prefer the tmt-provisioned key when it is actually present (CI lanes), and
+    otherwise fall back to the runner container's own key — the path
+    ``_run_host`` used before it moved onto ``ssh_argv()``, so env-default
+    local runs keep authenticating.
+    """
+    return _TMT_SSH_KEY if os.path.exists(_TMT_SSH_KEY) else DEFAULT_SSH_KEY
+
 
 def _first_value(*values: str) -> str:
     for value in values:
@@ -105,8 +120,8 @@ def before_all(context) -> None:
         userdata.get("key", ""),
         os.environ.get("SSH_KEY", ""),
         os.environ.get("SSH_KEY_PATH", ""),
-        os.environ.get("TMT_SSH_KEY", "/etc/ssh/test-key/id_ed25519"),
-    )
+        os.environ.get("TMT_SSH_KEY", ""),
+    ) or _default_ssh_key()
     context.ssh_port = _first_value(
         userdata.get("ssh_port", ""),
         os.environ.get("SSH_PORT", ""),
diff --git a/tests/shared/ssh_config.py b/tests/shared/ssh_config.py
index d23b3938f..9d5110ab7 100644
--- a/tests/shared/ssh_config.py
+++ b/tests/shared/ssh_config.py
@@ -17,7 +17,10 @@
    ``ssh_port``) — behave's ``userdata`` is a plain dict; on a mock context
    without userdata, attribute lookup falls through to step 3 automatically.
 3. Environment variables (``SSH_KEY``/``SSH_KEY_PATH``, ``VM_IP``,
-   ``VM_USER``/``SSH_USER``, ``SSH_PORT``/``VM_PORT``).
+   ``VM_USER``/``SSH_USER``, ``SSH_PORT``/``VM_PORT``/``TMT_SSH_PORT``).
+   ``TMT_SSH_PORT`` is read so tmt-provisioned lanes on a forwarded port do
+   not silently connect to 22 — suite ``environment.py`` hooks resolve host,
+   user and key from ``TMT_SSH_*`` but mostly never set ``context.ssh_port``.
 4. Built-in defaults matching the runner container layout.
 """
 
@@ -74,6 +77,7 @@ def resolve_ssh_details(context=None) -> dict:
             userdata.get("ssh_port", ""),
             os.environ.get("SSH_PORT", ""),
             os.environ.get("VM_PORT", ""),
+            os.environ.get("TMT_SSH_PORT", ""),
         ) or DEFAULT_SSH_PORT,
     }
 
@@ -109,9 +113,13 @@ def ssh_argv(context=None, *, connect_timeout: int = 10, quiet: bool = False) ->
 def populate_ssh_context(context) -> None:
     """Set the context attributes ``run_ssh`` requires.
 
-    Any suite whose steps star-import ``tests.shared.ssh_steps`` must call
-    this from ``before_all`` — otherwise the shared steps raise
-    ``AttributeError`` on first use.
+    Suites whose steps star-import ``tests.shared.ssh_steps`` should call this
+    from ``before_all`` (or set the attributes themselves). The shared steps no
+    longer raise ``AttributeError`` when it is skipped: ``run_ssh`` builds its
+    argv through ``ssh_argv(context)``, whose ``resolve_ssh_details`` falls back
+    to userdata, environment and built-in defaults. So a suite that skips this
+    call connects with *default* credentials instead of failing loudly — call it
+    to pin the resolved values once, in one place, for the whole run.
     """
     details = resolve_ssh_details(context)
     context.ssh_key = details["ssh_key"]
diff --git a/tests/unit/test_kde_smoke_environment.py b/tests/unit/test_kde_smoke_environment.py
index 781f23f8d..1f59bc3a6 100644
--- a/tests/unit/test_kde_smoke_environment.py
+++ b/tests/unit/test_kde_smoke_environment.py
@@ -141,3 +141,28 @@ def test_environment_uses_correct_webdriver_module():
     mod = importlib.import_module("tests.kde-smoke.features.environment")
     assert hasattr(mod.kde_webdriver, "new_session")
     assert not hasattr(mod.kde_webdriver, "start_driver")
+
+
+# ---------------------------------------------------------------------------
+# 4. Default SSH key resolution (no key configured by userdata or environment)
+# ---------------------------------------------------------------------------
+
+
+class TestDefaultSshKey:
+    """_run_host now resolves context.ssh_key through ssh_argv(); the env-default
+    key must stay usable in both the tmt-provisioned and the plain runner case."""
+
+    @pytest.fixture(autouse=True)
+    def _load_module(self):
+        self.mod = importlib.import_module("tests.kde-smoke.features.environment")
+
+    def test_prefers_tmt_key_when_present(self, monkeypatch):
+        monkeypatch.setattr(self.mod.os.path, "exists", lambda path: True)
+        assert self.mod._default_ssh_key() == "/etc/ssh/test-key/id_ed25519"
+
+    def test_falls_back_to_runner_key_when_tmt_key_absent(self, monkeypatch):
+        from tests.shared.ssh_config import DEFAULT_SSH_KEY
+
+        monkeypatch.setattr(self.mod.os.path, "exists", lambda path: False)
+        assert self.mod._default_ssh_key() == DEFAULT_SSH_KEY
+        assert DEFAULT_SSH_KEY == "/home/bluefin-test/.ssh/id_ed25519"
diff --git a/tests/unit/test_ssh_config.py b/tests/unit/test_ssh_config.py
index c88748f8b..26719d079 100644
--- a/tests/unit/test_ssh_config.py
+++ b/tests/unit/test_ssh_config.py
@@ -33,7 +33,7 @@ def _full_context(userdata=None):
 _ENV_CLEAR = {
     k: "" for k in (
         "SSH_KEY", "SSH_KEY_PATH", "VM_IP", "VM_USER", "SSH_USER",
-        "SSH_PORT", "VM_PORT",
+        "SSH_PORT", "VM_PORT", "TMT_SSH_PORT",
     )
 }
 
@@ -59,6 +59,20 @@ def test_environment_used_when_no_context_or_userdata(self):
         assert details["ssh_user"] == "envuser"
         assert details["ssh_port"] == "2222"
 
+    def test_tmt_ssh_port_used_when_no_other_port_set(self):
+        """tmt-provisioned lanes (dx/flatcar) export TMT_SSH_* only; without
+        this the forwarded port was dropped and ssh connected to 22."""
+        env = dict(_ENV_CLEAR, TMT_SSH_PORT="2022")
+        with patch.dict(os.environ, env, clear=False):
+            details = ssh_config.resolve_ssh_details(_bare_context())
+        assert details["ssh_port"] == "2022"
+
+    def test_ssh_port_beats_tmt_ssh_port(self):
+        env = dict(_ENV_CLEAR, SSH_PORT="2222", TMT_SSH_PORT="2022")
+        with patch.dict(os.environ, env, clear=False):
+            details = ssh_config.resolve_ssh_details(_bare_context())
+        assert details["ssh_port"] == "2222"
+
     def test_userdata_beats_environment(self):
         env = dict(_ENV_CLEAR, SSH_KEY="/env/key", VM_IP="192.0.2.10")
         userdata = {"ssh_key": "/ud/key", "vm_ip": "10.9.9.9"}

From fa1d374aa17940e0b0bb114348330fd01fd6ca5e Mon Sep 17 00:00:00 2001
From: Copilot <223556219+Copilot@users.noreply.github.com>
Date: Sun, 20 Sep 2026 03:14:52 -0400
Subject: [PATCH 4/4] fix(ssh): log resolved SSH target and restore 20s flatpak
 connect window

Address review findings on the ssh_argv() routing work:

- resolve_ssh_details() can no longer fail loudly when a suite never
  populates context: it falls back to the built-in runner defaults. Add
  resolve_ssh_details_with_sources() which reports where each field came
  from, and have ssh_argv() print the resolved destination once per
  distinct target, plus a warning when every field fell back to a
  built-in default (pointing at populate_ssh_context). A misconfigured
  suite now says which target it dialled instead of producing an opaque
  rc-255 against the wrong host.
- vanilla-gnome _ssh_run() takes an explicit connect_timeout, and the
  Flatpak probe passes connect_timeout=20 so its connect budget stays at
  the pre-refactor 20s instead of halving to ssh_argv's 10s default.

Docs and unit tests updated for both.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .../behave/references/shared-ssh.md           |  17 +++
 tests/shared/ssh_config.py                    | 134 +++++++++++++-----
 tests/unit/test_ssh_config.py                 |  51 +++++++
 tests/unit/test_vanilla_gnome_steps.py        |  17 +++
 tests/vanilla-gnome/features/steps/steps.py   |  15 +-
 5 files changed, 198 insertions(+), 36 deletions(-)

diff --git a/docs/skills/test-authoring/behave/references/shared-ssh.md b/docs/skills/test-authoring/behave/references/shared-ssh.md
index 37be73904..b65d69a21 100644
--- a/docs/skills/test-authoring/behave/references/shared-ssh.md
+++ b/docs/skills/test-authoring/behave/references/shared-ssh.md
@@ -46,6 +46,20 @@ userdata, environment variables and runner defaults. A suite that star-imports
 `AttributeError` — it silently connects with *default* credentials, which is
 worse. Populate them in `before_all`.
 
+To keep that failure mode diagnosable, `ssh_argv()` prints the resolved
+destination once per distinct target, together with where each field came from:
+
+```
+SSH target: bluefin-test@127.0.0.1:22 key=/home/bluefin-test/.ssh/id_ed25519 (ssh_key=default,ssh_port=default,ssh_user=default,vm_ip=default)
+WARNING: no SSH connection details were configured — using built-in runner defaults. ...
+```
+
+The warning fires only when *every* field fell back to a built-in default,
+i.e. nothing (context, userdata, or environment) configured the run. Use
+`ssh_config.resolve_ssh_details_with_sources(context)` when a suite or test
+needs to assert the details came from a non-default source.
+
+
 Resolve them through the shared helper rather than hand-rolling per suite:
 
 ```python
@@ -114,6 +128,9 @@ single builder for SSH argv. Two properties of that contract are easy to get wro
   where the old hand-rolled argv already suppressed the banner (`run_ssh`, `image_cache`,
   the software `_has_bazaar` probe, dx/flatcar/vanilla-gnome). Passing it everywhere hides
   diagnostics that kde-smoke and the screenshot helper rely on.
+- **`connect_timeout` is per call site.** The default is 10s; a probe that previously used
+  a longer connect window (e.g. the vanilla-gnome Flatpak probe, 20s) must pass
+  `connect_timeout=` explicitly rather than relying on the command timeout.
 
 Suites must not hand-roll `ssh` argv. `tests/unit/test_ssh_transport_contract.py`
 enforces this for the modules listed in its `MIGRATED_MODULES` set — add new suites there
diff --git a/tests/shared/ssh_config.py b/tests/shared/ssh_config.py
index 9d5110ab7..c2efc7db2 100644
--- a/tests/shared/ssh_config.py
+++ b/tests/shared/ssh_config.py
@@ -22,6 +22,11 @@
    not silently connect to 22 — suite ``environment.py`` hooks resolve host,
    user and key from ``TMT_SSH_*`` but mostly never set ``context.ssh_port``.
 4. Built-in defaults matching the runner container layout.
+
+Because step 4 always yields a usable destination, a suite that never
+populated its context cannot fail loudly any more. ``ssh_argv`` therefore logs
+the resolved destination once per run and warns when *every* field came from
+the built-in defaults — see ``log_resolved_ssh_target``.
 """
 
 import os
@@ -31,12 +36,25 @@
 DEFAULT_VM_USER = "bluefin-test"
 DEFAULT_SSH_PORT = "22"
 
+SOURCE_CONTEXT = "context"
+SOURCE_USERDATA = "userdata"
+SOURCE_ENVIRONMENT = "environment"
+SOURCE_DEFAULT = "default"
+
+_logged_targets = set()
+
 
-def _first_value(*values: str) -> str:
-    for value in values:
+def _pick(context_value, userdata_values, env_values, default):
+    """Return ``(value, source)`` for one field, honouring the priority order."""
+    if context_value:
+        return context_value, SOURCE_CONTEXT
+    for value in userdata_values:
         if value:
-            return value
-    return ""
+            return value, SOURCE_USERDATA
+    for value in env_values:
+        if value:
+            return value, SOURCE_ENVIRONMENT
+    return default, SOURCE_DEFAULT
 
 
 def _userdata(context) -> dict:
@@ -45,43 +63,92 @@ def _userdata(context) -> dict:
     return userdata if hasattr(userdata, "get") else {}
 
 
-def resolve_ssh_details(context=None) -> dict:
-    """Return SSH connection details for the current run.
+def resolve_ssh_details_with_sources(context=None) -> dict:
+    """Return ``{field: (value, source)}`` for the current run.
 
-    Keys: ``ssh_key``, ``vm_ip``, ``ssh_user``, ``ssh_port`` (all strings).
+    ``source`` is one of ``SOURCE_CONTEXT``/``SOURCE_USERDATA``/
+    ``SOURCE_ENVIRONMENT``/``SOURCE_DEFAULT`` so callers can tell a configured
+    destination apart from the built-in runner fallback.
     """
     userdata = _userdata(context)
     return {
-        "ssh_key": _first_value(
+        "ssh_key": _pick(
             getattr(context, "ssh_key", ""),
-            userdata.get("ssh_key", ""),
-            userdata.get("key", ""),
-            os.environ.get("SSH_KEY", ""),
-            os.environ.get("SSH_KEY_PATH", ""),
-        ) or DEFAULT_SSH_KEY,
-        "vm_ip": _first_value(
+            (userdata.get("ssh_key", ""), userdata.get("key", "")),
+            (os.environ.get("SSH_KEY", ""), os.environ.get("SSH_KEY_PATH", "")),
+            DEFAULT_SSH_KEY,
+        ),
+        "vm_ip": _pick(
             getattr(context, "vm_ip", ""),
-            userdata.get("vm_ip", ""),
-            userdata.get("host", ""),
-            os.environ.get("VM_IP", ""),
-        ) or DEFAULT_VM_IP,
-        "ssh_user": _first_value(
+            (userdata.get("vm_ip", ""), userdata.get("host", "")),
+            (os.environ.get("VM_IP", ""),),
+            DEFAULT_VM_IP,
+        ),
+        "ssh_user": _pick(
             getattr(context, "ssh_user", ""),
-            userdata.get("vm_user", ""),
-            userdata.get("user", ""),
-            os.environ.get("VM_USER", ""),
-            os.environ.get("SSH_USER", ""),
-        ) or DEFAULT_VM_USER,
-        "ssh_port": _first_value(
+            (userdata.get("vm_user", ""), userdata.get("user", "")),
+            (os.environ.get("VM_USER", ""), os.environ.get("SSH_USER", "")),
+            DEFAULT_VM_USER,
+        ),
+        "ssh_port": _pick(
             getattr(context, "ssh_port", ""),
-            userdata.get("ssh_port", ""),
-            os.environ.get("SSH_PORT", ""),
-            os.environ.get("VM_PORT", ""),
-            os.environ.get("TMT_SSH_PORT", ""),
-        ) or DEFAULT_SSH_PORT,
+            (userdata.get("ssh_port", ""),),
+            (
+                os.environ.get("SSH_PORT", ""),
+                os.environ.get("VM_PORT", ""),
+                os.environ.get("TMT_SSH_PORT", ""),
+            ),
+            DEFAULT_SSH_PORT,
+        ),
+    }
+
+
+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).
+    """
+    return {
+        field: value
+        for field, (value, _source) in
+        resolve_ssh_details_with_sources(context).items()
     }
 
 
+def log_resolved_ssh_target(context=None) -> None:
+    """Print the resolved SSH destination once per distinct target.
+
+    ``resolve_ssh_details`` cannot fail loudly on a suite that never populated
+    its context — it falls back to the runner defaults. Printing the resolved
+    destination (and warning when *nothing* configured it) turns an otherwise
+    silent connect failure against the wrong host into a diagnosable one.
+    """
+    resolved = resolve_ssh_details_with_sources(context)
+    details = {field: value for field, (value, _s) in resolved.items()}
+    destination = (
+        f"{details['ssh_user']}@{details['vm_ip']}:{details['ssh_port']}"
+    )
+    signature = (destination, details["ssh_key"])
+    if signature in _logged_targets:
+        return
+    _logged_targets.add(signature)
+    sources = ",".join(
+        f"{field}={source}" for field, (_v, source) in sorted(resolved.items())
+    )
+    print(
+        f"SSH target: {destination} key={details['ssh_key']} ({sources})",
+        flush=True,
+    )
+    if all(source == SOURCE_DEFAULT for _v, source in resolved.values()):
+        print(
+            "WARNING: no SSH connection details were configured — using "
+            "built-in runner defaults. If this suite targets a VM, call "
+            "tests.shared.ssh_config.populate_ssh_context(context) from "
+            "before_all (or set VM_IP/VM_USER/SSH_KEY/SSH_PORT).",
+            flush=True,
+        )
+
+
 def ssh_argv(context=None, *, connect_timeout: int = 10, quiet: bool = False) -> list[str]:
     """Return the canonical ``ssh`` argv prefix for the current run.
 
@@ -94,6 +161,7 @@ def ssh_argv(context=None, *, connect_timeout: int = 10, quiet: bool = False) ->
     that option inline instead of restating it at every call site.
     """
     details = resolve_ssh_details(context)
+    log_resolved_ssh_target(context)
     argv = [
         "ssh",
         "-i", details["ssh_key"],
@@ -118,8 +186,10 @@ def populate_ssh_context(context) -> None:
     longer raise ``AttributeError`` when it is skipped: ``run_ssh`` builds its
     argv through ``ssh_argv(context)``, whose ``resolve_ssh_details`` falls back
     to userdata, environment and built-in defaults. So a suite that skips this
-    call connects with *default* credentials instead of failing loudly — call it
-    to pin the resolved values once, in one place, for the whole run.
+    call connects with *default* credentials instead of failing loudly (though
+    ``ssh_argv`` logs the resolved destination and warns when every field is a
+    built-in default) — call it to pin the resolved values once, in one place,
+    for the whole run.
     """
     details = resolve_ssh_details(context)
     context.ssh_key = details["ssh_key"]
diff --git a/tests/unit/test_ssh_config.py b/tests/unit/test_ssh_config.py
index 26719d079..ef6f02652 100644
--- a/tests/unit/test_ssh_config.py
+++ b/tests/unit/test_ssh_config.py
@@ -92,6 +92,57 @@ def test_context_attributes_beat_everything(self):
         assert details["ssh_port"] == "2200"
 
 
+class TestResolvedTargetLogging:
+    """Silent fallback to runner defaults must at least be diagnosable: the
+    resolved destination is logged and an all-defaults resolution warns."""
+
+    def setup_method(self):
+        ssh_config._logged_targets.clear()
+
+    def test_sources_reported_per_field(self):
+        env = dict(_ENV_CLEAR, VM_IP="192.0.2.50")
+        with patch.dict(os.environ, env, clear=False):
+            resolved = ssh_config.resolve_ssh_details_with_sources(
+                _bare_context({"vm_user": "uduser"})
+            )
+        assert resolved["vm_ip"] == ("192.0.2.50", ssh_config.SOURCE_ENVIRONMENT)
+        assert resolved["ssh_user"] == ("uduser", ssh_config.SOURCE_USERDATA)
+        assert resolved["ssh_key"] == (
+            ssh_config.DEFAULT_SSH_KEY, ssh_config.SOURCE_DEFAULT
+        )
+
+    def test_context_attributes_reported_as_context_source(self):
+        with patch.dict(os.environ, _ENV_CLEAR, clear=False):
+            resolved = ssh_config.resolve_ssh_details_with_sources(_full_context())
+        assert all(
+            source == ssh_config.SOURCE_CONTEXT for _v, source in resolved.values()
+        )
+
+    def test_argv_logs_destination_and_warns_on_all_defaults(self, capsys):
+        with patch.dict(os.environ, _ENV_CLEAR, clear=False):
+            ssh_config.ssh_argv(_bare_context())
+        out = capsys.readouterr().out
+        assert (
+            f"{ssh_config.DEFAULT_VM_USER}@{ssh_config.DEFAULT_VM_IP}:"
+            f"{ssh_config.DEFAULT_SSH_PORT}" in out
+        )
+        assert "WARNING" in out
+        assert "populate_ssh_context" in out
+
+    def test_no_warning_when_target_is_configured(self, capsys):
+        with patch.dict(os.environ, _ENV_CLEAR, clear=False):
+            ssh_config.ssh_argv(_full_context())
+        out = capsys.readouterr().out
+        assert "ctxuser@10.1.1.1:2200" in out
+        assert "WARNING" not in out
+
+    def test_destination_logged_once_per_target(self, capsys):
+        with patch.dict(os.environ, _ENV_CLEAR, clear=False):
+            ssh_config.ssh_argv(_full_context())
+            ssh_config.ssh_argv(_full_context())
+        assert capsys.readouterr().out.count("SSH target:") == 1
+
+
 class TestPopulateSshContext:
     def test_sets_all_attributes_run_ssh_requires(self):
         ctx = _bare_context()
diff --git a/tests/unit/test_vanilla_gnome_steps.py b/tests/unit/test_vanilla_gnome_steps.py
index 94198048f..0c8c8eba8 100644
--- a/tests/unit/test_vanilla_gnome_steps.py
+++ b/tests/unit/test_vanilla_gnome_steps.py
@@ -93,6 +93,23 @@ def test_returns_false_when_ssh_not_available(self):
         with patch.object(m, "_ssh_run", side_effect=FileNotFoundError):
             assert m._flatpak_app_exists("org.gnome.Nautilus") is False
 
+    def test_keeps_20s_connect_window(self):
+        """Regression: the probe's connect budget was 20s before the argv was
+        centralised; ssh_argv's 10s default must not silently halve it."""
+        m = _import_vanilla_gnome_steps()
+        mock_result = MagicMock(returncode=0, stdout="")
+        with patch.object(m, "_ssh_run", return_value=mock_result) as ssh_run:
+            m._flatpak_app_exists("org.gnome.Nautilus")
+        assert ssh_run.call_args.kwargs["timeout"] == 20
+        assert ssh_run.call_args.kwargs["connect_timeout"] == 20
+
+    def test_ssh_run_passes_connect_timeout_to_argv(self):
+        m = _import_vanilla_gnome_steps()
+        with patch.object(m, "ssh_argv", return_value=["ssh"]) as argv, \
+             patch.object(m.subprocess, "run", return_value=MagicMock()):
+            m._ssh_run("true", timeout=20, connect_timeout=20)
+        assert argv.call_args.kwargs["connect_timeout"] == 20
+
 
 # ---------------------------------------------------------------------------
 # _assert_any_app_present
diff --git a/tests/vanilla-gnome/features/steps/steps.py b/tests/vanilla-gnome/features/steps/steps.py
index 0f01735a7..1a7d3c71b 100644
--- a/tests/vanilla-gnome/features/steps/steps.py
+++ b/tests/vanilla-gnome/features/steps/steps.py
@@ -134,15 +134,18 @@ def overview_search_bar_contains(context, text) -> None:
     assert text in entry.text, f"Search bar text '{entry.text}' does not contain '{text}'"
 
 
-def _ssh_run(cmd: str, timeout: int = 15, context=None) -> subprocess.CompletedProcess:
+def _ssh_run(cmd: str, timeout: int = 15, context=None,
+             connect_timeout: int = 10) -> subprocess.CompletedProcess:
     """Run a command on the VM via SSH.
 
     ``timeout`` bounds only the local wait for the command to finish; the SSH
-    *connect* deadline is fixed at ``ssh_argv``'s default (10s) so a long
+    *connect* deadline is passed separately as ``connect_timeout`` so a long
     command timeout no longer inflates how long a dead connection is retried.
+    Call sites that need a longer connect window on a slow-booting VM raise
+    ``connect_timeout`` explicitly.
     """
     return subprocess.run(
-        ssh_argv(context, quiet=True) + [cmd],
+        ssh_argv(context, quiet=True, connect_timeout=connect_timeout) + [cmd],
         capture_output=True, text=True, timeout=timeout,
     )
 
@@ -200,7 +203,11 @@ def _command_exists(command: str, context=None) -> bool:
 def _flatpak_app_exists(app_id: str, context=None) -> bool:
     """Check whether a Flatpak app is installed on the VM."""
     try:
-        result = _ssh_run('flatpak list --app --columns=application 2>/dev/null', timeout=20, context=context)
+        result = _ssh_run(
+            'flatpak list --app --columns=application 2>/dev/null',
+            timeout=20, connect_timeout=20, context=context,
+        )
+
     except (FileNotFoundError, subprocess.TimeoutExpired):
         return False
     if result.returncode != 0: