refactor(ssh): Phase 2 — route dx/flatcar/kde-smoke/software/vanilla-gnome onto ssh_config.ssh_argv()

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.

docs/skills/test-authoring/behave/references/shared-ssh.md

@@ -38,11 +38,27 @@ 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`.
+
+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:
 
@@ -100,3 +116,22 @@ 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.
+- **`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
+rather than copying an argv list.

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,

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,

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", ""),

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:

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
 
 

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,
     )