tests/support/pkg.py - master - githubmirror/salt

tests/support/pkg.py
2 326 строк89 KB
Daniel A. Wozniak
Merge branch '3008.x' into merge/3008.x/master-26-06-24
26 июн 2026, 00:39
26 июн 2026, 00:390ab117d
import atexit
import configparser
import contextlib
import errno
import logging
import os
import pathlib
import pprint
import re
import shutil
import subprocess
import textwrap
import time
from typing import TYPE_CHECKING

import attr
import distro
import packaging.version
import psutil
import pytest
import requests
import saltfactories.cli
from pytestshellutils.shell import DaemonImpl, Subprocess
from pytestshellutils.utils.processes import (
    ProcessResult,
    _get_cmdline,
    terminate_process,
    terminate_process_list,
)
from pytestskipmarkers.utils import platform
from saltfactories.bases import SystemdSaltDaemonImpl
from saltfactories.cli import call, key
from saltfactories.daemons import api, master, minion
from saltfactories.utils import cli_scripts

import salt.utils.files
from tests.conftest import CODE_DIR
from tests.support.pytest.helpers import TestAccount, download_file

ARTIFACTS_DIR = CODE_DIR / "artifacts" / "pkg"

log = logging.getLogger(__name__)


def _macos_salt_exe_command_v(path_prefix: str):
    """
    Resolve ``salt`` using only *path_prefix* in ``PATH`` (bash ``command -v``).

    ``shutil.which`` can still pick a stale or unrelated ``salt`` when the
    process ``PATH`` is polluted; an isolated lookup matches CI runners more
    reliably after ``installer`` drops files under ``/opt``.
    """
    try:
        env = os.environ.copy()
        env["PATH"] = path_prefix
        proc = subprocess.run(
            ["/bin/bash", "-lc", "command -v salt"],
            check=False,
            capture_output=True,
            text=True,
            timeout=20,
            env=env,
        )
    except (OSError, subprocess.TimeoutExpired):
        return None
    if proc.returncode != 0:
        return None
    lines = [ln.strip() for ln in proc.stdout.splitlines() if ln.strip()]
    if not lines:
        return None
    return lines[0]


def _macos_prefix_from_salt_exe(salt_exe: str):
    if not salt_exe:
        return None
    try:
        wp = pathlib.Path(salt_exe).resolve()
    except OSError:
        return None
    if "saltstack" not in str(wp) and "/opt/salt" not in str(wp):
        return None
    if wp.parent.name == "bin":
        return wp.parent.parent
    return wp.parent


def _macos_salt_onedir_prefix():
    """
    Return the on-disk Salt onedir prefix on macOS, if installed.

    Newer macOS packages install under ``/opt/saltstack/salt`` (same layout as
    Linux onedir). Older releases used ``/opt/salt``. Use ``exists()`` so
    symlinks and executable wrappers are detected the same way as in CI.
    If layout probes miss (e.g. before ``PATH`` is updated), fall back to
    ``shutil.which`` with common install locations prepended.
    """
    for candidate in (
        pathlib.Path("/opt/saltstack/salt"),
        pathlib.Path("/opt/salt"),
    ):
        if (candidate / "bin" / "salt").exists():
            return candidate
        if (candidate / "salt").exists():
            return candidate
    mac_bins = "/opt/saltstack/salt/bin:/opt/salt/bin:/opt/saltstack/salt:/opt/salt"
    path_for_which = mac_bins + os.pathsep + os.environ.get("PATH", "")
    prefix = _macos_prefix_from_salt_exe(shutil.which("salt", path=path_for_which))
    if prefix is not None:
        return prefix
    return _macos_prefix_from_salt_exe(_macos_salt_exe_command_v(mac_bins))


def pep440_public_equal(reported: str, expected: str) -> bool:
    """
    True when *reported* and *expected* match on release/pre/post/dev, ignoring
    local when one side omits it (``salt --version`` often drops ``+g``).
    """
    try:
        pr = packaging.version.parse(reported)
        pe = packaging.version.parse(expected)
    except packaging.version.InvalidVersion:
        return reported == expected

    def _pub(v):
        return (v.release, v.pre, v.post, v.dev)

    return _pub(pr) == _pub(pe)


def pep440_version_to_rpm_nevra_version(version: str) -> str:
    """
    Map a PEP440-style version string to the RPM ``Version`` field spelling.

    Published RPMs use a tilde before pre-release labels (``3008.0~rc1``) while
    CI and pytest often pass ``3008.0rc1``. :command:`yum` / :command:`tdnf`
    then cannot resolve ``salt-3008.0rc1``.

    If *version* already contains ``~`` (RPM-shaped), it is returned unchanged.
    Non-pre-release versions are returned unchanged.
    """
    if not version or "~" in version:
        return version
    try:
        parsed = packaging.version.parse(version)
    except packaging.version.InvalidVersion:
        return version
    if parsed.pre is None and parsed.dev is None:
        return version
    release = ".".join(str(p) for p in parsed.release)
    out = release
    if parsed.pre is not None:
        pre_l, pre_n = parsed.pre
        out = f"{out}~{pre_l}{pre_n}"
    else:
        out = f"{out}~dev{parsed.dev}"
    if parsed.post is not None:
        out = f"{out}.post{parsed.post}"
    if parsed.local is not None:
        out = f"{out}+{parsed.local}"
    return out


import pytestshellutils.shell
import pytestshellutils.utils.processes

_original_terminate = pytestshellutils.shell.SubprocessImpl._terminate


def _patched_terminate(self):
    if not platform.is_darwin():
        return _original_terminate(self)

    from tests.support.mock import patch

    with patch("psutil.Process.children", return_value=[]):
        return _original_terminate(self)


pytestshellutils.shell.SubprocessImpl._terminate = _patched_terminate


@attr.s(kw_only=True, slots=True)
class SaltPkgInstall:
    pkg_system_service: bool = attr.ib(default=False)
    proc: Subprocess = attr.ib(init=False, repr=False)

    # Paths
    root: pathlib.Path = attr.ib(default=None)
    run_root: pathlib.Path = attr.ib(default=None)
    ssm_bin: pathlib.Path = attr.ib(default=None)
    bin_dir: pathlib.Path = attr.ib(default=None)
    install_dir: pathlib.Path = attr.ib(init=False)
    binary_paths: dict[str, list[pathlib.Path]] = attr.ib(init=False)
    config_path: str = attr.ib(init=False)
    conf_dir: pathlib.Path = attr.ib()

    # Test selection flags
    upgrade: bool = attr.ib(default=False)
    downgrade: bool = attr.ib(default=False)
    classic: bool = attr.ib(default=False)

    # Installing flags
    no_install: bool = attr.ib(default=False)
    no_uninstall: bool = attr.ib(default=False)

    # Distribution/system information
    distro_id: str = attr.ib(init=False)
    distro_codename: str = attr.ib(init=False)
    distro_name: str = attr.ib(init=False)
    distro_version: str = attr.ib(init=False)

    # Version information
    prev_version: str = attr.ib()
    use_prev_version: str = attr.ib()
    artifact_version: str = attr.ib(init=False)
    version: str = attr.ib(init=False)

    # Package (and management) metadata
    pkg_mngr: str = attr.ib(init=False)
    rm_pkg: str = attr.ib(init=False)
    dbg_pkg: str = attr.ib(init=False)
    salt_pkgs: list[str] = attr.ib(init=False)
    pkgs: list[str] = attr.ib(factory=list)
    file_ext: bool = attr.ib(default=None)
    relenv: bool = attr.ib(default=True)
    installer_timeout: int = attr.ib(default=attr.NOTHING)
    install_env: dict = attr.ib(factory=dict)

    @proc.default
    def _default_proc(self):
        return Subprocess(timeout=240)

    @installer_timeout.default
    def _default_installer_timeout(self):
        if platform.is_windows():
            return 600
        return 240

    @distro_id.default
    def _default_distro_id(self):
        return distro.id().lower()

    @distro_codename.default
    def _default_distro_codename(self):
        return distro.codename().lower()

    @distro_name.default
    def _default_distro_name(self):
        name = distro.name()
        if name:
            if "vmware" in name.lower():
                return name.split()[1].lower()
            return name.split()[0].lower()

    @distro_version.default
    def _default_distro_version(self):
        if self.distro_name in ("photon", "rocky"):
            return distro.version().split(".")[0]
        return distro.version().lower()

    @pkg_mngr.default
    def _default_pkg_mngr(self):
        if self.distro_id in (
            "almalinux",
            "rocky",
            "centos",
            "redhat",
            "amzn",
            "fedora",
            "photon",
        ):
            return "yum"
        elif self.distro_id in ("ubuntu", "debian"):
            ret = self.proc.run("apt-get", "update")
            self._check_retcode(ret)
            return "apt-get"

    @rm_pkg.default
    def _default_rm_pkg(self):
        if self.distro_id in (
            "almalinux",
            "rocky",
            "centos",
            "redhat",
            "amzn",
            "fedora",
            "photon",
        ):
            return "remove"
        elif self.distro_id in ("ubuntu", "debian"):
            return "purge"

    @dbg_pkg.default
    def _default_dbg_pkg(self):
        dbg_pkg = None
        if self.distro_id in (
            "almalinux",
            "rocky",
            "centos",
            "redhat",
            "amzn",
            "fedora",
            "photon",
        ):
            dbg_pkg = "salt-debuginfo"
        elif self.distro_id in ("ubuntu", "debian"):
            dbg_pkg = "salt-dbg"
        return dbg_pkg

    @salt_pkgs.default
    def _default_salt_pkgs(self):
        salt_pkgs = [
            "salt-api",
            "salt-syndic",
            "salt-ssh",
            "salt-master",
            "salt-cloud",
            "salt-minion",
        ]
        if self.distro_id in (
            "almalinux",
            "rocky",
            "centos",
            "redhat",
            "amzn",
            "fedora",
            "photon",
        ):
            salt_pkgs.append("salt")
        elif self.distro_id in ("ubuntu", "debian"):
            salt_pkgs.append("salt-common")
        if packaging.version.parse(self.version) >= packaging.version.parse("3006.3"):
            if self.dbg_pkg:
                salt_pkgs.append(self.dbg_pkg)
        return salt_pkgs

    @install_dir.default
    def _default_install_dir(self):
        if platform.is_windows():
            install_dir = pathlib.Path(
                os.getenv("ProgramFiles"), "Salt Project", "Salt"
            ).resolve()
        elif platform.is_darwin():
            found = _macos_salt_onedir_prefix()
            install_dir = found if found is not None else pathlib.Path("/opt", "salt")
        else:
            install_dir = pathlib.Path("/opt", "saltstack", "salt")
        return install_dir

    @config_path.default
    def _default_config_path(self):
        """
        Default location for salt configurations
        """
        if platform.is_windows():
            config_path = pathlib.Path("C:\\salt", "etc", "salt")
        else:
            config_path = pathlib.Path("/etc", "salt")
        return config_path

    @version.default
    def _default_version(self):
        """
        The version to be installed at the start
        """
        if not self.upgrade and not self.use_prev_version:
            version = self.artifact_version
        else:
            if self.prev_version is None:
                raise ValueError(
                    "prev_version must be provided for upgrade tests. "
                    "Use --prev-version option to specify the previous version."
                )
            if self.use_prev_version:
                # Post-downgrade integration: ``salt --version`` reports the full
                # previous release (e.g. 3008.0rc1), not ``major.minor`` only.
                version = self.prev_version
            else:
                version = self.prev_version
                parsed = packaging.version.parse(version)
                version = f"{parsed.major}.{parsed.minor}"
        # ensure services stopped on Debian/Ubuntu (minic install for RedHat - non-starting)
        if self.distro_id in ("ubuntu", "debian"):
            self.stop_services()
        return version

    @artifact_version.default
    def _default_artifact_version(self):
        """
        The version of the local salt artifacts being tested, based on regex matching
        """
        version = ""
        artifacts = list(ARTIFACTS_DIR.glob("**/*.*"))
        for artifact in artifacts:
            version = re.search(
                r"([0-9].*)(\-[0-9].fc|\-[0-9].el|\+ds|\_all|\_any|\_amd64|\_arm64|\-[0-9].am|(\-[0-9]-[a-z]*-[a-z]*[0-9_]*.|\-[0-9]*.*)(exe|msi|pkg|rpm|deb))",
                artifact.name,
            )
            if version:
                version = version.groups()[0].replace("_", "-").replace("~", "")
                version = version.split("-")[0]
                break
        if not version:
            pytest.fail(
                f"Failed to find package artifacts in '{ARTIFACTS_DIR}'. "
                f"Directory Contents:\n{pprint.pformat(artifacts)}"
            )
        return version

    def update_process_path(self):
        # The installer updates the path for the system, but that doesn't
        # make it to this python session, so we need to update that
        if platform.is_windows():
            os.environ["PATH"] = ";".join([str(self.install_dir), os.getenv("path")])
        elif platform.is_darwin():
            # On macOS, salt executables are in install_dir (/opt/salt)
            # while Python executables are in bin_dir (/opt/salt/bin)
            path_parts = [str(self.install_dir), str(self.bin_dir), os.getenv("PATH")]
            os.environ["PATH"] = ":".join(path_parts)
        else:
            os.environ["PATH"] = ":".join([str(self.bin_dir), os.getenv("PATH")])
        # Update the proc's captured environment so run() calls pick up the new PATH
        if self.proc is not None:
            self.proc.environ["PATH"] = os.environ["PATH"]

    def __attrs_post_init__(self):
        self.relenv = packaging.version.parse(self.version) >= packaging.version.parse(
            "3006.0"
        )
        if self.installer_timeout:
            self.proc.timeout = self.installer_timeout

        file_ext_re = "rpm|deb"
        if platform.is_darwin():
            file_ext_re = "pkg"
        if platform.is_windows():
            file_ext_re = "exe|msi"

        for f_path in ARTIFACTS_DIR.glob("**/*.*"):
            f_path = str(f_path)
            if re.search(f"salt-(.*).({file_ext_re})$", f_path, re.IGNORECASE):
                self.file_ext = os.path.splitext(f_path)[1].strip(".")
                self.pkgs.append(f_path)
                if platform.is_windows():
                    self.root = pathlib.Path(os.getenv("LocalAppData")).resolve()
                    if self.file_ext in ["exe", "msi"]:
                        self.root = self.install_dir.parent
                        self.bin_dir = self.install_dir
                        self.ssm_bin = self.install_dir / "ssm.exe"
                        self.run_root = self.bin_dir / "bin" / "salt.exe"
                        if not self.relenv and not self.classic:
                            self.ssm_bin = self.bin_dir / "bin" / "ssm.exe"
                    else:
                        log.error("Unexpected file extension: %s", self.file_ext)
                    if self.use_prev_version:
                        self.bin_dir = self.install_dir / "bin"
                        self.run_root = self.bin_dir / "salt.exe"
                        self.ssm_bin = self.bin_dir / "ssm.exe"
                        if self.file_ext == "msi" or self.relenv:
                            self.ssm_bin = self.install_dir / "ssm.exe"
                        if (
                            self.install_dir / "salt-minion.exe"
                        ).exists() and not self.relenv:
                            log.debug(
                                "Removing %s", self.install_dir / "salt-minion.exe"
                            )
                            (self.install_dir / "salt-minion.exe").unlink()

                elif platform.is_darwin():
                    self.root = pathlib.Path("/opt")
                    if self.file_ext == "pkg":
                        self.bin_dir = self.install_dir / "bin"
                        self.run_root = self.bin_dir / "run"
                    else:
                        log.error("Unexpected file extension: %s", self.file_ext)
                log.debug("root: %s", self.root)
                log.debug("bin_dir: %s", self.bin_dir)
                log.debug("ssm_bin: %s", self.ssm_bin)
                log.debug("run_root: %s", self.run_root)

        if not self.pkgs:
            pytest.fail("Could not find Salt Artifacts")

        python_bin = self.install_dir / "bin" / "python3"
        if platform.is_windows():
            python_bin = self.install_dir / "Scripts" / "python.exe"
            if self.relenv:
                self.binary_paths = {
                    "call": ["salt-call.exe"],
                    "cp": ["salt-cp.exe"],
                    "minion": ["salt-minion.exe"],
                    "pip": ["salt-pip.exe"],
                    "python": [python_bin],
                }
            elif self.classic:
                self.binary_paths = {
                    "call": [self.install_dir / "salt-call.bat"],
                    "cp": [self.install_dir / "salt-cp.bat"],
                    "minion": [self.install_dir / "salt-minion.bat"],
                    "python": [self.bin_dir / "python.exe"],
                }
                self.binary_paths["pip"] = self.binary_paths["python"] + ["-m", "pip"]
            else:
                self.binary_paths = {
                    "call": [str(self.run_root), "call"],
                    "cp": [str(self.run_root), "cp"],
                    "minion": [str(self.run_root), "minion"],
                    "pip": [str(self.run_root), "pip"],
                    "python": [str(self.run_root), "shell"],
                }

        else:
            if os.path.exists(self.install_dir / "bin" / "salt"):
                install_dir = self.install_dir / "bin"
            else:
                install_dir = self.install_dir
            if self.relenv:
                self.binary_paths = {
                    "salt": [install_dir / "salt"],
                    "api": [install_dir / "salt-api"],
                    "call": [install_dir / "salt-call"],
                    "cloud": [install_dir / "salt-cloud"],
                    "cp": [install_dir / "salt-cp"],
                    "key": [install_dir / "salt-key"],
                    "master": [install_dir / "salt-master"],
                    "minion": [install_dir / "salt-minion"],
                    "proxy": [install_dir / "salt-proxy"],
                    "run": [install_dir / "salt-run"],
                    "ssh": [install_dir / "salt-ssh"],
                    "syndic": [install_dir / "salt-syndic"],
                    "spm": [install_dir / "spm"],
                    "pip": [install_dir / "salt-pip"],
                    "python": [python_bin],
                }
            else:
                self.binary_paths = {
                    "salt": [shutil.which("salt")],
                    "api": [shutil.which("salt-api")],
                    "call": [shutil.which("salt-call")],
                    "cloud": [shutil.which("salt-cloud")],
                    "cp": [shutil.which("salt-cp")],
                    "key": [shutil.which("salt-key")],
                    "master": [shutil.which("salt-master")],
                    "minion": [shutil.which("salt-minion")],
                    "proxy": [shutil.which("salt-proxy")],
                    "run": [shutil.which("salt-run")],
                    "ssh": [shutil.which("salt-ssh")],
                    "syndic": [shutil.which("salt-syndic")],
                    "spm": [shutil.which("spm")],
                    "python": [str(pathlib.Path("/usr/bin/python3"))],
                }
                if self.classic:
                    if platform.is_darwin():
                        # `which` is not catching the right paths on downgrades, explicitly defining them here
                        self.binary_paths = {
                            "salt": [self.bin_dir / "salt"],
                            "api": [self.bin_dir / "salt-api"],
                            "call": [self.bin_dir / "salt-call"],
                            "cloud": [self.bin_dir / "salt-cloud"],
                            "cp": [self.bin_dir / "salt-cp"],
                            "key": [self.bin_dir / "salt-key"],
                            "master": [self.bin_dir / "salt-master"],
                            "minion": [self.bin_dir / "salt-minion"],
                            "proxy": [self.bin_dir / "salt-proxy"],
                            "run": [self.bin_dir / "salt-run"],
                            "ssh": [self.bin_dir / "salt-ssh"],
                            "syndic": [self.bin_dir / "salt-syndic"],
                            "spm": [self.bin_dir / "spm"],
                            "python": [str(self.bin_dir / "python3")],
                            "pip": [str(self.bin_dir / "pip3")],
                        }
                    else:
                        self.binary_paths["pip"] = [str(pathlib.Path("/usr/bin/pip3"))]
                        self.proc.run(*self.binary_paths["pip"], "install", "-U", "pip")
                        self.proc.run(
                            *self.binary_paths["pip"], "install", "-U", "pyopenssl"
                        )
                else:
                    self.binary_paths["python"] = [shutil.which("salt"), "shell"]
                    if platform.is_darwin():
                        self.binary_paths["pip"] = [self.run_root, "pip"]
                        self.binary_paths["spm"] = [shutil.which("salt-spm")]
                    else:
                        self.binary_paths["pip"] = [shutil.which("salt-pip")]
        log.debug("python_bin: %s", python_bin)
        log.debug("binary_paths: %s", self.binary_paths)
        log.debug("install_dir: %s", self.install_dir)

    def _refresh_macos_binary_paths(self):
        """
        Re-resolve ``install_dir`` / ``binary_paths`` after a ``.pkg`` install.

        ``__attrs_post_init__`` runs before installers populate ``/opt/...``,
        so upgrade/downgrade sessions need a second pass once files exist.
        """
        if not platform.is_darwin():
            return
        # Prepends so :func:`shutil.which` inside prefix detection can see
        # ``/opt`` layouts before a stale default ``/opt/salt`` mis-seeds
        # ``$PATH`` from :meth:`update_process_path`.
        opt_first = [
            p
            for p in (
                "/opt/saltstack/salt/bin",
                "/opt/saltstack/salt",
                "/opt/salt/bin",
                "/opt/salt",
            )
            if pathlib.Path(p).exists()
        ]
        _old = os.environ.get("PATH", "")
        if opt_first:
            os.environ["PATH"] = os.pathsep.join(opt_first) + os.pathsep + _old
        try:
            found = _macos_salt_onedir_prefix()
        finally:
            os.environ["PATH"] = _old
        if found is None:
            return
        self.install_dir = found
        self.bin_dir = found / "bin"
        self.run_root = self.bin_dir / "run"
        python_bin = self.install_dir / "bin" / "python3"
        if os.path.exists(self.install_dir / "bin" / "salt"):
            install_dir = self.install_dir / "bin"
        else:
            install_dir = self.install_dir
        if self.relenv:
            self.binary_paths = {
                "salt": [install_dir / "salt"],
                "api": [install_dir / "salt-api"],
                "call": [install_dir / "salt-call"],
                "cloud": [install_dir / "salt-cloud"],
                "cp": [install_dir / "salt-cp"],
                "key": [install_dir / "salt-key"],
                "master": [install_dir / "salt-master"],
                "minion": [install_dir / "salt-minion"],
                "proxy": [install_dir / "salt-proxy"],
                "run": [install_dir / "salt-run"],
                "ssh": [install_dir / "salt-ssh"],
                "syndic": [install_dir / "salt-syndic"],
                "spm": [install_dir / "spm"],
                "pip": [install_dir / "salt-pip"],
                "python": [python_bin],
            }
        else:
            self.binary_paths = {
                "salt": [shutil.which("salt")],
                "api": [shutil.which("salt-api")],
                "call": [shutil.which("salt-call")],
                "cloud": [shutil.which("salt-cloud")],
                "cp": [shutil.which("salt-cp")],
                "key": [shutil.which("salt-key")],
                "master": [shutil.which("salt-master")],
                "minion": [shutil.which("salt-minion")],
                "proxy": [shutil.which("salt-proxy")],
                "run": [shutil.which("salt-run")],
                "ssh": [shutil.which("salt-ssh")],
                "syndic": [shutil.which("salt-syndic")],
                "spm": [shutil.which("spm")],
                "python": [str(pathlib.Path("/usr/bin/python3"))],
            }
            if self.classic:
                self.binary_paths = {
                    "salt": [self.bin_dir / "salt"],
                    "api": [self.bin_dir / "salt-api"],
                    "call": [self.bin_dir / "salt-call"],
                    "cloud": [self.bin_dir / "salt-cloud"],
                    "cp": [self.bin_dir / "salt-cp"],
                    "key": [self.bin_dir / "salt-key"],
                    "master": [self.bin_dir / "salt-master"],
                    "minion": [self.bin_dir / "salt-minion"],
                    "proxy": [self.bin_dir / "salt-proxy"],
                    "run": [self.bin_dir / "salt-run"],
                    "ssh": [self.bin_dir / "salt-ssh"],
                    "syndic": [self.bin_dir / "salt-syndic"],
                    "spm": [self.bin_dir / "spm"],
                    "python": [str(self.bin_dir / "python3")],
                    "pip": [str(self.bin_dir / "pip3")],
                }
            else:
                self.binary_paths["python"] = [shutil.which("salt"), "shell"]
                self.binary_paths["pip"] = [self.run_root, "pip"]
                self.binary_paths["spm"] = [shutil.which("salt-spm")]
        log.debug(
            "Refreshed macOS binary_paths (install_dir=%s): %s",
            self.install_dir,
            self.binary_paths,
        )

    @staticmethod
    def salt_factories_root_dir(system_service: bool = False) -> pathlib.Path:
        if system_service is False:
            return None
        if platform.is_windows():
            return pathlib.Path("C:\\salt")
        if platform.is_darwin():
            found = _macos_salt_onedir_prefix()
            return found if found is not None else pathlib.Path("/opt/salt")
        return pathlib.Path("/")

    def _check_retcode(self, ret):
        """
        Helper function to check subprocess.run returncode equals 0
        If not raise AssertionError
        """
        if ret.returncode != 0:
            log.error(ret)
            # Provide better error message for Windows access violation
            # 0xC0000005 as signed 32-bit integer is 3221225477
            if platform.is_windows() and ret.returncode in (0xC0000005, 3221225477):
                log.error(
                    "Windows installer crashed with access violation (0xC0000005). "
                    "This may indicate a file locking issue or that services weren't fully stopped."
                )
        assert ret.returncode == 0
        return True

    def _install_pkgs(self, upgrade=False, downgrade=False):
        if downgrade:
            self.install_previous(downgrade=downgrade)
            return True
        pkg = None
        if platform.is_windows() and self.file_ext:
            for p in self.pkgs:
                if p.endswith(self.file_ext):
                    pkg = str(pathlib.Path(p).resolve())
                    break
        if pkg is None:
            pkg = str(pathlib.Path(self.pkgs[0]).resolve())
        if platform.is_windows():
            if upgrade:
                self.root = self.install_dir.parent
                self.bin_dir = self.install_dir
                self.ssm_bin = self.install_dir / "ssm.exe"
                self._ensure_windows_services_stopped()
                time.sleep(3)
            if pkg.endswith("exe"):
                log.info("Installing: %s", str(pkg))
                ret = self.proc.run(
                    str(pkg),
                    "/start-minion=0",
                    "/S",
                    _timeout=self.installer_timeout,
                )
                self._check_retcode(ret)
            elif pkg.endswith("msi"):
                log.info("Installing: %s", str(pkg))
                # self.proc.run always makes the command a list even when shell
                # is true, meaning shell being true will never work correctly.
                msi_cmd = f'msiexec.exe /qn /i "{pkg}" /norestart START_MINION=""'
                ret = subprocess.run(
                    msi_cmd,
                    shell=True,  # nosec
                    check=False,
                )
                log.info("MSI returncode: %s", ret.returncode)
                assert ret.returncode in [0, 3010]

                if upgrade:
                    # MSI major upgrades with mismatched component GUIDs can
                    # remove files that should be kept. Running a repair
                    # ensures all files from the new product are on disk.
                    repair_cmd = f'msiexec.exe /qn /fa "{pkg}" /norestart'
                    repair_ret = subprocess.run(
                        repair_cmd,
                        shell=True,  # nosec
                        check=False,
                    )
                    log.info("MSI repair returncode: %s", repair_ret.returncode)
            else:
                log.error("Invalid package: %s", pkg)
                return False

            log.debug("Removing installed salt-minion service")
            self.proc.run(str(self.ssm_bin), "stop", "salt-minion", "confirm")
            subprocess.run(
                "sc.exe delete salt-minion",
                shell=True,  # nosec
                check=False,
            )
            # Wait for Windows to fully purge the service entry
            for _ in range(30):
                ret = subprocess.run(
                    "sc.exe query salt-minion",
                    shell=True,  # nosec
                    check=False,
                    capture_output=True,
                )
                if ret.returncode != 0:
                    break
                time.sleep(1)

            # Add installation to the path
            self.update_process_path()

            if self.pkg_system_service:
                self._install_ssm_service()

        elif platform.is_darwin():
            daemons_dir = pathlib.Path("/Library", "LaunchDaemons")
            service_name = "com.saltstack.salt.minion"
            plist_file = daemons_dir / f"{service_name}.plist"
            log.debug("Installing: %s", str(pkg))
            ret = self.proc.run("installer", "-pkg", str(pkg), "-target", "/")
            self._check_retcode(ret)

            # Stop the service installed by the installer

            try:
                subprocess.run(
                    ["launchctl", "disable", f"system/{service_name}"],
                    check=False,
                    timeout=30,
                )
                subprocess.run(
                    ["launchctl", "bootout", "system", str(plist_file)],
                    check=False,
                    timeout=30,
                )
            except subprocess.TimeoutExpired:
                log.warning("launchctl command timed out")

            self._refresh_macos_binary_paths()

        elif upgrade:
            env = os.environ.copy()
            extra_args = []
            if self.distro_id in ("ubuntu", "debian"):

                pref_file = pathlib.Path(
                    "/etc", "apt", "preferences.d", "salt-pin-1001"
                )
                pref_file.parent.mkdir(exist_ok=True)
                deb_artifact = pep440_version_to_rpm_nevra_version(
                    self.artifact_version
                )
                pin = f"{deb_artifact.rsplit('.', 1)[0]}.*"
                with salt.utils.files.fopen(pref_file, "w") as fp:
                    fp.write(
                        f"Package: salt-*\n"
                        f"Pin: version {pin}\n"
                        f"Pin-Priority: 1001"
                    )
                log.error("Pin to %s", pin)

                env["DEBIAN_FRONTEND"] = "noninteractive"
                extra_args = [
                    "-o",
                    "DPkg::Options::=--force-confdef",
                    "-o",
                    "DPkg::Options::=--force-confold",
                    # Downgrade leaves (e.g.) ``salt-dbg`` newer than pinned mains until
                    # the next full install; ``apt upgrade`` needs this on Debian/Ubuntu.
                    "--allow-downgrades",
                ]
            log.info("Installing packages:\n%s", pprint.pformat(self.pkgs))
            args = extra_args + self.pkgs
            upgrade_cmd = "upgrade"
            if self.distro_id == "photon":
                # tdnf does not detect nightly build versions to be higher version
                # than release versions
                upgrade_cmd = "install"
                if "+" in self.pkgs[0]:
                    # self.pkgs are not signed unless this is a release.
                    args.append("--nogpgcheck")
            ret = self.proc.run(
                self.pkg_mngr,
                upgrade_cmd,
                "-y",
                *args,
                env=env,
            )
        else:
            # Fresh install path
            env = os.environ.copy()
            # Add any custom install environment variables
            if self.install_env:
                env.update(self.install_env)

            args = ["install", "-y"]
            if self.distro_id == "photon":
                ret = self.proc.run(
                    "rpm",
                    "--import",
                    "https://packages.broadcom.com/artifactory/api/security/keypair/SaltProjectKey/public",
                )
                self._check_retcode(ret)
                if "+" in self.pkgs[0]:
                    # self.pkgs are not signed unless this is a release.
                    args.append("--nogpgcheck")
            log.info("Installing packages:\n%s", pprint.pformat(self.pkgs))
            args += self.pkgs
            ret = self.proc.run(self.pkg_mngr, *args, env=env)

        if not platform.is_darwin() and not platform.is_windows():
            # Make sure we don't have any trailing references to old package file locations
            assert ret.returncode == 0
            assert "/saltstack/salt/run" not in ret.stdout
            log.info(ret)
            self._check_retcode(ret)

    def _ensure_windows_services_stopped(self):
        """
        Stop Salt Windows services prior to running the installer and wait for
        their processes to exit to avoid upgrade timeouts.
        """
        if not platform.is_windows():
            return
        if not getattr(self, "ssm_bin", None):
            return
        if not pathlib.Path(self.ssm_bin).exists():
            log.debug("Windows SSM binary not found at %s", self.ssm_bin)
            return

        stop = self.proc.run(
            str(self.ssm_bin), "stop", "salt-minion", "confirm", _timeout=120
        )
        # 1062: The service has not been started.
        if stop.returncode not in (0, 1062):
            log.debug(
                "Stopping service salt-minion returned %s",
                stop.returncode,
            )

        deadline = time.time() + 120
        while time.time() < deadline:
            running = {
                (proc.info["name"] or "").lower()
                for proc in psutil.process_iter(["name"])
                if (proc.info["name"] or "").lower() == "salt-minion.exe"
            }
            if not running:
                break
            log.debug("Waiting for salt-minion process to exit before upgrade")
            time.sleep(2)
        else:
            log.warning("salt-minion process still running before upgrade")

    def _install_ssm_service(self, service="minion"):
        """
        This function installs the service on Windows using SSM but does not
        start it.

        Args:

            service (str):
                The name of the service. Default is ``minion``
        """
        service_name = f"salt-{service}"
        binary = self.install_dir / f"{service_name}.exe"
        ret = self.proc.run(
            str(self.ssm_bin),
            "install",
            service_name,
            binary,
            "-c",
            f'"{str(self.conf_dir)}"',
        )
        self._check_retcode(ret)
        ret = self.proc.run(
            str(self.ssm_bin),
            "set",
            service_name,
            "Description",
            "Salt Minion for testing",
        )
        self._check_retcode(ret)
        # This doesn't start the service. It will start automatically on reboot
        # It is set here to make it the same as what the installer does
        ret = self.proc.run(
            str(self.ssm_bin), "set", service_name, "Start", "SERVICE_AUTO_START"
        )
        self._check_retcode(ret)
        ret = self.proc.run(
            str(self.ssm_bin), "set", service_name, "AppStopMethodConsole", "24000"
        )
        self._check_retcode(ret)
        ret = self.proc.run(
            str(self.ssm_bin), "set", service_name, "AppStopMethodWindow", "2000"
        )
        self._check_retcode(ret)
        ret = self.proc.run(
            str(self.ssm_bin), "set", service_name, "AppRestartDelay", "60000"
        )
        self._check_retcode(ret)

    def package_python_version(self):
        return self.proc.run(
            str(self.binary_paths["python"][0]),
            "-c",
            "import sys; print('{}.{}'.format(*sys.version_info))",
        ).stdout.strip()

    def install(self, upgrade=False, downgrade=False, stop_services=True):
        self._install_pkgs(upgrade=upgrade, downgrade=downgrade)
        if platform.is_darwin():
            self._refresh_macos_binary_paths()
            self.update_process_path()
            if stop_services:
                # The Salt .pkg loads ``com.saltstack.salt.minion`` (and on
                # some installs also ``master``/``api``/``syndic``) at install
                # time via ``RunAtLoad=true``. Those daemons start with the
                # default minion config (``master: salt``, which does not
                # resolve), so they never authenticate to the test fixture's
                # master. Worse, saltfactories' ``PkgLaunchdSaltDaemonImpl``
                # checks the plist label when its own ``salt_minion`` fixture
                # tries to ``launchctl bootstrap`` -- sees the auto-loaded
                # daemon, declares the test minion "already running", and
                # silently uses the misconfigured launchd-managed process
                # instead. Fan-out then finds zero connected minions and
                # ``api_request`` returns ``{'return': [{}]}``.
                # Bootout the auto-installed daemons here so the label is
                # free when the test fixture brings up its own.
                self._stop_macos_pkg_daemons()
        if self.distro_id in ("ubuntu", "debian") and stop_services:
            self.stop_services()
        elif (
            upgrade
            and self.pkg_system_service
            and not platform.is_windows()
            and not platform.is_darwin()
        ):
            # RPM/DEB upgrade replaces on disk while systemd units can keep old
            # processes until restart; ``salt-call --local test.version`` then
            # still reports the previous release (see upgrade systemd teardown).
            self.restart_services()

    def _stop_macos_pkg_daemons(self):
        """
        Bootout each ``com.saltstack.salt.*`` plist the .pkg installed so the
        labels are free for the test fixtures to load their own configurations.
        Idempotent: ``launchctl bootout`` on a not-loaded service exits with a
        non-zero status which we deliberately ignore. Also re-enables the
        service after bootout so a previous test run that left it disabled
        (via ``launchctl disable`` in fixture teardown) does not block the
        next ``launchctl bootstrap`` -- ``disable`` persists on disk in
        ``/var/db/com.apple.xpc.launchd/disabled.plist`` across reboots.
        """
        plist_dir = pathlib.Path("/Library/LaunchDaemons")
        for service in ("salt-master", "salt-minion", "salt-api", "salt-syndic"):
            label = f"com.saltstack.{service.replace('-', '.')}"
            plist = plist_dir / f"{label}.plist"
            if not plist.exists():
                continue
            for cmd in (
                ("launchctl", "bootout", "system", str(plist)),
                ("launchctl", "enable", f"system/{label}"),
            ):
                try:
                    subprocess.run(
                        ["sudo", *cmd], check=False, capture_output=True, timeout=30
                    )
                except subprocess.TimeoutExpired:
                    log.warning("launchctl %s timed out for %s", cmd[1], service)

    def stop_services(self):
        """
        Debian/Ubuntu distros automatically start the services on install
        We want to ensure our tests start with the config settings we have set.
        This will also verify the expected services are up and running.
        """
        retval = True
        for service in ["salt-syndic", "salt-master", "salt-minion"]:
            check_run = self.proc.run("systemctl", "status", service)
            if check_run.returncode != 0:
                # The system was not started automatically and
                # we are expecting it to be on install on Debian/Ubuntu systems
                log.debug("The service %s was not started on install.", service)
                retval = False
            else:
                stop_service = self.proc.run("systemctl", "stop", service)
                self._check_retcode(stop_service)
        return retval

    def restart_services(self):
        """
        Debian/Ubuntu distros automatically start the services
        We want to ensure our tests start with the config settings we have set,
        for example: after install the services are stopped (similar to RedHat not starting services on install)
        This will also verify the expected services are up and running.
        """
        for service in ["salt-minion", "salt-master", "salt-syndic"]:
            check_run = self.proc.run("systemctl", "status", service)
            log.debug(
                "The restart_services status, before restart, for service %s is %s.",
                service,
                check_run,
            )
            restart_service = self.proc.run("systemctl", "restart", service)
            self._check_retcode(restart_service)

    def _salt_yum_repo_path(self) -> pathlib.Path:
        """
        Path to the Broadcom ``salt.repo`` copy under ``/etc/yum.repos.d`` (see
        :meth:`install_previous`).
        """
        distro_name = self.distro_name
        if distro_name in ("almalinux", "rocky", "centos", "fedora"):
            distro_name = "redhat"
        return pathlib.Path("/etc/yum.repos.d") / f"salt-{distro_name}.repo"

    def _rpm_repo_set_enabled_photon(self, repo_id: str, enabled: bool) -> None:
        """
        PhotonOS uses ``tdnf`` behind ``yum``; it does not implement
        ``dnf config-manager --enable``. Toggle ``enabled=`` in the repo file and
        refresh metadata.
        """
        repo_path = self._salt_yum_repo_path()
        parser = configparser.ConfigParser(interpolation=None)
        if not parser.read(repo_path, encoding="utf-8"):
            log.error("Could not read Salt repo file at %s", repo_path)
            assert False, f"Missing or unreadable repo file: {repo_path}"
        if not parser.has_section(repo_id):
            log.error(
                "Salt repo file %s has no [%s] stanza (upstream salt.repo changed?)",
                repo_path,
                repo_id,
            )
            assert False, f"Repo file {repo_path} missing [{repo_id}]"
        parser.set(repo_id, "enabled", "1" if enabled else "0")
        with salt.utils.files.fopen(repo_path, "w", encoding="utf-8") as fp:
            parser.write(fp)
        ret = self.proc.run(self.pkg_mngr, "makecache", "-y")
        self._check_retcode(ret)

    def install_previous(self, downgrade=False):
        """
        Install previous version. This is used for upgrade tests.
        """
        major_ver = packaging.version.parse(self.prev_version).major
        relenv = packaging.version.parse(self.prev_version) >= packaging.version.parse(
            "3006.0"
        )
        distro_name = self.distro_name
        if distro_name in ("almalinux", "rocky", "centos", "fedora"):
            distro_name = "redhat"
        root_url = "https://packages.broadcom.com/artifactory"

        if self.distro_name in [
            "almalinux",
            "rocky",
            "redhat",
            "centos",
            "amazon",
            "fedora",
            "vmware",
            "photon",
        ]:
            # Removing EPEL repo files
            for fp in pathlib.Path("/etc", "yum.repos.d").glob("epel*"):
                fp.unlink()

            if platform.is_aarch64():
                arch = "arm64"
                # Starting with 3006.5, we prioritize the aarch64 repo paths for rpm-based distros
                if packaging.version.parse(
                    self.prev_version
                ) >= packaging.version.parse("3006.5"):
                    arch = "aarch64"
            else:
                arch = "x86_64"
            ret = self.proc.run(
                "rpm",
                "--import",
                "https://packages.broadcom.com/artifactory/api/security/keypair/SaltProjectKey/public",
            )
            self._check_retcode(ret)
            download_file(
                "https://github.com/saltstack/salt-install-guide/releases/latest/download/salt.repo",
                f"/etc/yum.repos.d/salt-{distro_name}.repo",
            )

            cmd_action = "downgrade" if downgrade else "install"
            pkgs_to_install = self.salt_pkgs.copy()

            if self.distro_name == "photon":
                rpm_prev = pep440_version_to_rpm_nevra_version(self.prev_version)
                orig_pkgs = pkgs_to_install[:]
                pkgs_to_install = []
                for _ in orig_pkgs:
                    pkgs_to_install.append(f"{_}-{rpm_prev}")
                ret = self.proc.run(self.pkg_mngr, "clean", "all")
                self._check_retcode(ret)
                if major_ver >= 3008:
                    self._rpm_repo_set_enabled_photon("salt-repo-latest", True)
                else:
                    repo_path = self._salt_yum_repo_path()
                    if repo_path:
                        parser = configparser.ConfigParser(interpolation=None)
                        if parser.read(
                            repo_path, encoding="utf-8"
                        ) and parser.has_section("salt-repo-3008-lts"):
                            self._rpm_repo_set_enabled_photon(
                                "salt-repo-3008-lts", False
                            )
            else:
                if "3007" in self.prev_version:
                    ret = self.proc.run(
                        self.pkg_mngr,
                        "config-manager",
                        "--enable",
                        "salt-repo-3007-sts",
                    )
                    self._check_retcode(ret)
                elif major_ver >= 3008:
                    # Default ``salt.repo`` enables v3006 LTS only; that stanza excludes
                    # ``*3008*``. Published 3008.x RPMs (including pre-releases) are only
                    # visible when ``salt-repo-latest`` is enabled.
                    ret = self.proc.run(
                        self.pkg_mngr,
                        "config-manager",
                        "--enable",
                        "salt-repo-latest",
                    )
                    self._check_retcode(ret)
                else:
                    ret = self.proc.run(
                        self.pkg_mngr,
                        "config-manager",
                        "--disable",
                        "salt-repo-3007-sts",
                    )
                    self._check_retcode(ret)
                if "3008" in self.prev_version:
                    ret = self.proc.run(
                        self.pkg_mngr,
                        "config-manager",
                        "--enable",
                        "salt-repo-3008-lts",
                    )
                    self._check_retcode(ret)
                else:
                    ret = self.proc.run(
                        self.pkg_mngr,
                        "config-manager",
                        "--disable",
                        "salt-repo-3008-lts",
                    )
                    self._check_retcode(ret)
                ret = self.proc.run(self.pkg_mngr, "clean", "expire-cache")
                self._check_retcode(ret)
                # Unversioned ``yum downgrade`` only moves one step among *all* repo
                # versions, so a downgrade from a 3008 pre-release artifact can jump to
                # 3006.x if the exact ``--prev-version`` build is not selected. Pin the
                # RPM set like Photon already does (release suffix is distro-specific).
                if downgrade:
                    rpm_prev = pep440_version_to_rpm_nevra_version(self.prev_version)
                    orig_pkgs = pkgs_to_install[:]
                    if self.dbg_pkg:
                        orig_pkgs = [p for p in orig_pkgs if self.dbg_pkg not in p]
                    pkgs_to_install = [f"{pkg}-{rpm_prev}-*" for pkg in orig_pkgs]

            if self.distro_version == "8" and self.classic:
                # centosstream 8 doesn't downgrade properly using the downgrade command for some reason
                # So we explicitly install the correct version here
                list_ret = self.proc.run(
                    self.pkg_mngr, "list", "--available", "salt"
                ).stdout.split("\n")
                list_ret = [_.strip() for _ in list_ret]
                idx = list_ret.index("Available Packages")
                old_ver = list_ret[idx + 1].split()[1]
                pkgs_to_install = [f"{pkg}-{old_ver}" for pkg in pkgs_to_install]
                if self.dbg_pkg:
                    # self.dbg_pkg does not exist on classic packages
                    dbg_exists = [x for x in pkgs_to_install if self.dbg_pkg in x]
                    if dbg_exists:
                        pkgs_to_install.remove(dbg_exists[0])
                cmd_action = "install"
            # pkgs = [f"{_}=={self.prev_version}" for _ in pkgs_to_install]
            ret = self.proc.run(
                self.pkg_mngr,
                cmd_action,
                *pkgs_to_install,
                "-y",
            )
            self._check_retcode(ret)

        elif distro_name in ["debian", "ubuntu"]:
            self.proc.run("dpkg", "--configure", "-a")
            ret = self.proc.run(self.pkg_mngr, "install", "curl", "-y")
            self._check_retcode(ret)
            ret = self.proc.run(self.pkg_mngr, "install", "apt-transport-https", "-y")
            self._check_retcode(ret)
            ## only classic 3005 has arm64 support
            if relenv and platform.is_aarch64():
                arch = "arm64"
            elif platform.is_aarch64() and self.classic:
                arch = "arm64"
            else:
                arch = "amd64"

            pathlib.Path("/etc/apt/keyrings").mkdir(parents=True, exist_ok=True)
            gpg_full_path = "/etc/apt/keyrings/salt-archive-keyring.pgp"

            # download the gpg pub key
            download_file(
                f"{root_url}/api/security/keypair/SaltProjectKey/public",
                f"{gpg_full_path}",
            )
            with salt.utils.files.fopen(
                pathlib.Path("/etc", "apt", "sources.list.d", "salt.list"), "w"
            ) as fp:
                fp.write(
                    f"deb [signed-by={gpg_full_path} arch={arch}] "
                    f"{root_url}/saltproject-deb/ stable main"
                )
            self._check_retcode(ret)
            pref_file = pathlib.Path("/etc", "apt", "preferences.d", "salt-pin-1001")
            pref_file.parent.mkdir(exist_ok=True)
            deb_upstream = pep440_version_to_rpm_nevra_version(self.prev_version)
            # Only use the explicit ``pkg=…`` path when Debian's version spelling
            # differs from *prev_version* (e.g. ``3008.0rc1`` vs ``3008.0~rc1``).
            # For normal releases (``3007.14``), ``deb_upstream == prev_version``;
            # reusing this branch would skip ``salt-dbg`` and leave it newer than the
            # pin, so a later ``apt upgrade`` fails without ``--allow-downgrades``.
            if downgrade and relenv and deb_upstream != self.prev_version:
                # ``Pin: version 3008.0rc1`` does not match published Debian versions
                # spelled ``3008.0~rc1``.  Unversioned ``apt-get install salt-*`` then
                # leaves a locally installed nightly (``3008.0~rc1+185…``) untouched, so
                # ``salt --version`` never drops below the CI artifact (downgrade test
                # asserts ``downgraded < artifact_ver``).
                #
                # Do **not** use ``pkg=3008.0~rc1*``: the glob matches the already-installed
                # ``3008.0~rc1+185…`` builds (they sort higher), so apt keeps the artifact.
                # Pin and install the exact Broadcom repo upstream version (see
                # ``apt-cache show salt-common | ^Version:``), same as ``apt-cache madison``.
                pin = deb_upstream
                install_targets = []
                for name in self.salt_pkgs:
                    if self.dbg_pkg and name == self.dbg_pkg:
                        continue
                    install_targets.append(f"{name}={deb_upstream}")
                cmd = [self.pkg_mngr, "install", *install_targets, "-y"]
            else:
                pin = self.prev_version
                cmd = [self.pkg_mngr, "install", *self.salt_pkgs, "-y"]

            with salt.utils.files.fopen(pref_file, "w") as fp:
                fp.write(f"Package: salt-*\nPin: version {pin}\nPin-Priority: 1001\n")

            # if downgrade:
            #    pref_file = pathlib.Path("/etc", "apt", "preferences.d", "salt-pin-1001")
            #    pref_file.parent.mkdir(exist_ok=True)
            #    # TODO: There's probably something I should put in here to say what version
            #    # TODO: But maybe that's done elsewhere, hopefully in self.salt_pkgs
            #    pref_file.write_text(
            #        textwrap.dedent(
            #            f"""\
            #    Package: salt*
            #    Pin: origin "{root_url}/saltproject-deb"
            #    Pin-Priority: 1001
            #    """
            #        ),
            #        encoding="utf-8",
            #    )
            cmd.append("--allow-downgrades")
            env = os.environ.copy()
            env["DEBIAN_FRONTEND"] = "noninteractive"
            extra_args = [
                "-o",
                "DPkg::Options::=--force-confdef",
                "-o",
                "DPkg::Options::=--force-confold",
            ]
            self.proc.run(self.pkg_mngr, "update", *extra_args, env=env)

            cmd.extend(extra_args)
            log.error("Run cmd %s", cmd)
            ret = self.proc.run(*cmd, env=env)
            log.error("cmd return %r", ret)
            # Pre-relenv packages down get downgraded to cleanly programmatically
            # They work manually, and the install tests after downgrades will catch problems with the install
            self._check_retcode(ret)
            # Let's not check the returncode if this is the case
            # if not (
            #    downgrade
            #    and packaging.version.parse(self.prev_version)
            #    < packaging.version.parse("3006.0")
            # ):
            #    self._check_retcode(ret)
            if downgrade and not self.no_uninstall:
                pref_file.unlink()
            self.stop_services()
        elif platform.is_windows():
            self.bin_dir = self.install_dir / "bin"
            self.run_root = self.bin_dir / "salt.exe"
            self.ssm_bin = self.install_dir / "ssm.exe"
            pkg = str(pathlib.Path(self.pkgs[0]).resolve())

            win_pkg = None
            if self.file_ext == "exe":
                win_pkg = (
                    f"Salt-Minion-{self.prev_version}-Py3-AMD64-Setup.{self.file_ext}"
                )
            elif self.file_ext == "msi":
                win_pkg = f"Salt-Minion-{self.prev_version}-Py3-AMD64.{self.file_ext}"
            else:
                log.debug("Unknown windows file extension: %s", self.file_ext)

            win_pkg_url = (
                f"{root_url}/saltproject-generic/windows/{self.prev_version}/{win_pkg}"
            )
            pkg_path = pathlib.Path(r"C:\TEMP", win_pkg)
            pkg_path.parent.mkdir(exist_ok=True)
            download_file(win_pkg_url, pkg_path)

            if self.file_ext == "msi":

                if downgrade:
                    # MSI can not be downgraded, we must remove the newer version
                    # before installing the old one.
                    ret = subprocess.run(
                        f"msiexec.exe /qn /x {pkg} /norestart",
                        shell=True,  # nosec
                        check=False,
                    )
                    assert ret.returncode == 0

                # self.proc.run always makes the command a list even when shell
                # is true, meaning shell being true will never work correctly.
                ret = subprocess.run(
                    f'msiexec.exe /qn /i {pkg_path} /norestart START_MINION=""',
                    shell=True,  # nosec
                    check=False,
                )
                assert ret.returncode in [0, 3010]
            else:
                # ret = self.proc.run("start", "/wait", f"\"{pkg_path} /start-minion=0 /S\"")
                batch_file = pkg_path.parent / "install_nsis.cmd"
                batch_content = f"start /wait {str(pkg_path)} /start-minion=0 /S"
                with salt.utils.files.fopen(batch_file, "w") as fp:
                    fp.write(batch_content)
                # Now run the batch file
                ret = self.proc.run("cmd.exe", "/c", str(batch_file), _timeout=900)
                self._check_retcode(ret)

            log.debug("Removing installed salt-minion service")
            ret = self.proc.run(str(self.ssm_bin), "remove", "salt-minion", "confirm")
            self._check_retcode(ret)

            # Add installation to the path
            self.update_process_path()

            if self.pkg_system_service:
                self._install_ssm_service()

        elif platform.is_darwin():
            if relenv and platform.is_aarch64():
                arch = "arm64"
            elif platform.is_aarch64() and self.classic:
                arch = "arm64"
            else:
                arch = "x86_64"

            mac_pkg = f"salt-{self.prev_version}-py3-{arch}.pkg"
            mac_pkg_url = (
                f"{root_url}/saltproject-generic/macos/{self.prev_version}/{mac_pkg}"
            )

            mac_pkg_path = f"/tmp/{mac_pkg}"
            if not os.path.exists(mac_pkg_path):
                download_file(
                    f"{mac_pkg_url}",
                    f"/tmp/{mac_pkg}",
                )

            ret = self.proc.run("installer", "-pkg", mac_pkg_path, "-target", "/")
            self._check_retcode(ret)
            self._refresh_macos_binary_paths()

            # Stop services started by the old installer so the test framework
            # can bootstrap them with the correct test configuration on start.
            daemons_dir = pathlib.Path("/Library", "LaunchDaemons")
            for svc in ("minion", "master", "api", "syndic"):
                svc_name = f"com.saltstack.salt.{svc}"
                plist_file = daemons_dir / f"{svc_name}.plist"
                try:
                    subprocess.run(
                        ["launchctl", "disable", f"system/{svc_name}"],
                        check=False,
                        timeout=30,
                    )
                    subprocess.run(
                        ["launchctl", "bootout", "system", str(plist_file)],
                        check=False,
                        timeout=30,
                    )
                except subprocess.TimeoutExpired:
                    log.warning("launchctl timed out stopping %s", svc_name)

    def uninstall(self):
        pkg = self.pkgs[0]
        if platform.is_windows() and self.file_ext:
            for p in self.pkgs:
                if p.endswith(self.file_ext):
                    pkg = p
                    break
        if platform.is_windows():
            log.info("Uninstalling %s", pkg)
            if pkg.endswith("exe"):
                uninst = self.install_dir / "uninst.exe"
                ret = self.proc.run(uninst, "/S")
                self._check_retcode(ret)
            elif pkg.endswith("msi"):
                ret = self.proc.run("msiexec.exe", "/qn", "/x", pkg)
                self._check_retcode(ret)

        elif platform.is_darwin():
            # From here: https://stackoverflow.com/a/46118276/4581998
            daemons_dir = pathlib.Path("/Library", "LaunchDaemons")
            for service in ("minion", "master", "api", "syndic"):
                service_name = f"com.saltstack.salt.{service}"
                plist_file = daemons_dir / f"{service_name}.plist"
                # Stop the services

                try:
                    subprocess.run(
                        ["launchctl", "disable", f"system/{service_name}"],
                        check=False,
                        timeout=30,
                    )
                    subprocess.run(
                        ["launchctl", "bootout", "system", str(plist_file)],
                        check=False,
                        timeout=30,
                    )
                except subprocess.TimeoutExpired:
                    log.warning("launchctl command timed out")

            # Remove Symlink to salt-config
            if os.path.exists("/usr/local/sbin/salt-config"):
                os.unlink("/usr/local/sbin/salt-config")

            # Remove supporting files
            # Use shell=True for piped commands
            self.proc.run(
                "pkgutil --only-files --files com.saltstack.salt | grep -v opt | sed 's|^|/|' | tr '\\n' '\\0' | xargs -0 rm -f",
                shell=True,
            )

            # Remove directories
            if os.path.exists("/etc/salt"):
                shutil.rmtree("/etc/salt")

            # Remove path
            if os.path.exists("/etc/paths.d/salt"):
                os.remove("/etc/paths.d/salt")

            # Remove receipt
            self.proc.run("pkgutil", "--forget", "com.saltstack.salt")

            log.debug("Deleting the onedir directory: %s", self.install_dir)
            if self.install_dir.exists():
                shutil.rmtree(str(self.install_dir))
        else:
            log.debug("Un-Installing packages:\n%s", pprint.pformat(self.salt_pkgs))
            ret = self.proc.run(self.pkg_mngr, self.rm_pkg, "-y", *self.salt_pkgs)
            self._check_retcode(ret)

    def write_launchd_conf(self, service):
        service_name = f"com.saltstack.salt.{service}"
        daemons_dir = pathlib.Path("/Library", "LaunchDaemons")
        plist_file = daemons_dir / f"{service_name}.plist"
        # Always overwrite the plist installed by the Salt package. The
        # package plist launches the daemon with no -c flag (uses /etc/salt),
        # but the test suite needs -c <conf_dir> so daemons use the test
        # configuration.  This must run as root (the whole pkg test suite
        # requires root on macOS via ``sudo -E nox``).
        if plist_file.exists():
            log.debug("Overwriting existing plist for service: %s", service)
        else:
            log.debug("Creating plist for service: %s", service)

        contents = f"""\
            <?xml version="1.0" encoding="UTF-8"?>
            <!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
            <plist version="1.0">
                <dict>
                    <key>Label</key>
                    <string>{service_name}</string>
                    <key>RunAtLoad</key>
                    <true/>
                    <key>KeepAlive</key>
                    <true/>
                    <key>ProgramArguments</key>
                    <array>"""
        for part in self.binary_paths[service]:
            contents += f"""\n                        <string>{part}</string>\n"""
        contents += f"""\
                        <string>-c</string>
                        <string>{self.conf_dir}</string>
                    </array>
                    <key>SoftResourceLimits</key>
                    <dict>
                        <key>NumberOfFiles</key>
                        <integer>100000</integer>
                    </dict>
                    <key>HardResourceLimits</key>
                    <dict>
                        <key>NumberOfFiles</key>
                        <integer>100000</integer>
                    </dict>
                </dict>
            </plist>
            """
        plist_file.write_text(textwrap.dedent(contents), encoding="utf-8")
        log.debug("Wrote '%s'. Contents:\n%s", plist_file, plist_file.read_text())

        # Remove the plist on exit so we don't leave test configuration behind.
        atexit.register(plist_file.unlink)

    def write_systemd_conf(self, service, binary):
        ret = self.proc.run("systemctl", "daemon-reload")
        self._check_retcode(ret)
        ret = self.proc.run("systemctl", "status", service)
        if ret.returncode == 4:
            log.warning(
                "No systemd unit file was found for service %s. Creating one.", service
            )
            contents = textwrap.dedent(
                """\
                [Unit]
                Description={service}

                [Service]
                KillMode=process
                Type=notify
                NotifyAccess=all
                LimitNOFILE=8192
                ExecStart={tgt} -c {conf_dir}

                [Install]
                WantedBy=multi-user.target
                """
            )
            if isinstance(binary, list) and len(binary) == 1:
                binary = shutil.which(binary[0]) or binary[0]
            elif isinstance(binary, list):
                binary = " ".join(binary)
            unit_path = pathlib.Path(f"/etc/systemd/system/{service}.service")
            contents = contents.format(
                service=service, tgt=binary, conf_dir=self.conf_dir
            )
            log.info("Created '%s'. Contents:\n%s", unit_path, contents)
            unit_path.write_text(contents, encoding="utf-8")
            ret = self.proc.run("systemctl", "daemon-reload")
            atexit.register(unit_path.unlink)
            self._check_retcode(ret)

    def __enter__(self):
        self.update_process_path()
        if platform.is_darwin():
            self._refresh_macos_binary_paths()

        if self.no_install:
            return self

        if self.upgrade:
            self.install_previous()
        else:
            # assume downgrade, since no_install only used in these two cases
            self.install()
        return self

    def __exit__(self, *_):
        if not self.no_uninstall:
            self.uninstall()

        # Did we left anything running?!
        procs = []
        if not platform.is_windows():

            try:
                output = subprocess.check_output(
                    ["ps", "-eo", "pid,command"], text=True
                )
                for line in output.splitlines()[1:]:
                    parts = line.strip().split(maxsplit=1)
                    if len(parts) == 2:
                        pid_str, cmdline = parts
                        if "salt" in cmdline and (
                            "/opt/saltstack" in cmdline or "/opt/salt" in cmdline
                        ):
                            try:
                                pid = int(pid_str)
                                if pid != os.getpid():
                                    procs.append(psutil.Process(pid))
                            except (ValueError, psutil.NoSuchProcess):
                                pass
            except subprocess.CalledProcessError:
                pass
        else:
            for proc in psutil.process_iter():
                try:
                    name = proc.name()
                    if "salt" in name or "python" in name:
                        cmdl_strg = " ".join(
                            str(element) for element in _get_cmdline(proc)
                        )
                        if "salt" in name or "salt" in cmdl_strg:
                            if (
                                "/opt/saltstack" in cmdl_strg
                                or "/opt/salt" in cmdl_strg
                            ):
                                procs.append(proc)
                except (
                    psutil.NoSuchProcess,
                    psutil.AccessDenied,
                    psutil.ZombieProcess,
                ):
                    continue

        if procs:
            if platform.is_darwin():
                for proc in procs:
                    try:
                        proc.kill()
                    except psutil.NoSuchProcess:
                        pass
            else:
                try:
                    terminate_process_list(procs, kill=True, slow_stop=True)
                except OSError as exc:
                    # psutil / waitpid EINVAL on some aarch64/debian runners during teardown
                    if getattr(exc, "errno", None) == errno.EINVAL:
                        log.warning(
                            "terminate_process_list failed with EINVAL (%r); leftover "
                            "/opt/saltstack process sweep skipped.",
                            exc,
                        )
                    else:
                        raise


class PkgSystemdSaltDaemonImpl(SystemdSaltDaemonImpl):
    # pylint: disable=access-member-before-definition
    def get_service_name(self):
        if self._service_name is None:
            self._service_name = self.factory.script_name
        return self._service_name

    # pylint: enable=access-member-before-definition


@attr.s(kw_only=True)
class PkgLaunchdSaltDaemonImpl(PkgSystemdSaltDaemonImpl):

    plist_file = attr.ib()

    @plist_file.default
    def _default_plist_file(self):
        daemons_dir = pathlib.Path("/Library", "LaunchDaemons")
        return daemons_dir / f"{self.get_service_name()}.plist"

    def get_service_name(self):
        if self._service_name is None:
            service_name = super().get_service_name()
            if "-" in service_name:
                service_name = service_name.split("-")[-1]
            self._service_name = f"com.saltstack.salt.{service_name}"
        return self._service_name

    def cmdline(self, *args):  # pylint: disable=arguments-differ
        """
        Construct a list of arguments to use when starting the subprocess.

        :param str args:
            Additional arguments to use when starting the subprocess

        """
        if args:  # pragma: no cover
            log.debug(
                "%s.run() is ignoring the passed in arguments: %r",
                self.__class__.__name__,
                args,
            )
        self._internal_run(
            "sudo",
            "launchctl",
            "enable",
            f"system/{self.get_service_name()}",
        )
        return (
            "sudo",
            "launchctl",
            "bootstrap",
            "system",
            str(self.plist_file),
        )

    def is_running(self):
        """
        Returns true if the sub-process is alive.
        """
        if self._process is None:
            # System-domain launchd services require root to query on macOS 13+.
            ret = self._internal_run(
                "sudo", "launchctl", "list", self.get_service_name()
            )
            if ret.stdout == "":
                return False

            if "PID" not in ret.stdout:
                return False

            pid = None
            # PID in a line that looks like this
            # "PID" = 445;
            for line in ret.stdout.splitlines():
                if "PID" in line:
                    pid = line.rstrip(";").split(" = ")[1]

            if pid is None:
                return False

            self._process = psutil.Process(int(pid))

        return self._process.is_running()

    def _terminate(self):
        """
        This method actually terminates the started daemon.
        """
        # We completely override the parent class method because we're not using
        # the self._terminal property, it's a launchd service
        if self._process is None:  # pragma: no cover
            # pylint: disable=access-member-before-definition
            if TYPE_CHECKING:
                # Make mypy happy
                assert self._terminal_result
            return self._terminal_result
            # pylint: enable=access-member-before-definition

        atexit.unregister(self.terminate)
        log.info("Stopping %s", self.factory)
        pid = self.pid
        # psutil.Process.children() can hang on macOS when the process is a
        # launchd-managed daemon; skip child collection on Darwin.
        with contextlib.suppress(psutil.NoSuchProcess):
            if not platform.is_darwin():
                for child in psutil.Process(pid).children(recursive=True):
                    # pylint: disable=access-member-before-definition
                    if child not in self._children:
                        self._children.append(child)
                    # pylint: enable=access-member-before-definition

        if self._process.is_running():  # pragma: no cover
            cmdline = _get_cmdline(self._process)
        else:
            cmdline = []

        # Disable the service

        try:
            subprocess.run(
                ["launchctl", "disable", f"system/{self.get_service_name()}"],
                check=False,
                timeout=30,
            )
            # Unload the service
            subprocess.run(
                ["launchctl", "bootout", "system", str(self.plist_file)],
                check=False,
                timeout=30,
            )
        except subprocess.TimeoutExpired:
            log.warning("launchctl command timed out during terminate")

        if self._process.is_running():  # pragma: no cover
            try:
                self._process.wait(10)
            except psutil.TimeoutExpired:
                self._process.terminate()
                try:
                    self._process.wait(10)
                except psutil.TimeoutExpired:
                    pass

        try:
            exitcode = self._process.wait(5) or 0
        except psutil.TimeoutExpired:
            exitcode = 0

        # Dereference the internal _process attribute
        self._process = None
        # Lets log and kill any child processes left behind, including the main subprocess
        # if it failed to properly stop
        terminate_process(
            pid=pid,
            kill_children=True,
            children=self._children,  # pylint: disable=access-member-before-definition
            slow_stop=self.factory.slow_stop,
        )

        # pylint: disable=access-member-before-definition
        if self._terminal_stdout is not None:
            self._terminal_stdout.close()
        if self._terminal_stderr is not None:
            self._terminal_stderr.close()
        # pylint: enable=access-member-before-definition
        stdout = stderr = ""
        try:
            self._terminal_result = ProcessResult(
                returncode=exitcode, stdout=stdout, stderr=stderr, cmdline=cmdline
            )
            log.info("%s %s", self.factory.__class__.__name__, self._terminal_result)
            return self._terminal_result
        finally:
            self._terminal = None
            self._terminal_stdout = None
            self._terminal_stderr = None
            self._terminal_timeout = None
            self._children = []


@attr.s(kw_only=True)
class PkgSsmSaltDaemonImpl(PkgSystemdSaltDaemonImpl):
    def cmdline(self, *args):  # pylint: disable=arguments-differ
        """
        Construct a list of arguments to use when starting the subprocess.

        :param str args:
            Additional arguments to use when starting the subprocess

        """
        if args:  # pragma: no cover
            log.debug(
                "%s.run() is ignoring the passed in arguments: %r",
                self.__class__.__name__,
                args,
            )
        return (
            str(self.factory.salt_pkg_install.ssm_bin),
            "start",
            self.get_service_name(),
        )

    def is_running(self):
        """
        Returns true if the sub-process is alive.
        """
        if self._process is None:
            n = 1
            while True:
                if self._process is not None:
                    break
                time.sleep(1)
                ret = self._internal_run(
                    str(self.factory.salt_pkg_install.ssm_bin),
                    "processes",
                    self.get_service_name(),
                )
                log.debug("process result %s", ret)
                if not ret.stdout or (ret.stdout and not ret.stdout.strip()):
                    if n >= 120:
                        return False
                    n += 1
                    continue
                for line in ret.stdout.splitlines():
                    log.warning("Line: %s", line)
                    if not line.strip():
                        continue
                    mainpid = line.strip().split()[0]
                    self._process = psutil.Process(int(mainpid))
                    break
        ret = self._process.is_running()
        if not hasattr(self, "logged_running"):
            log.error("SSM processs is running %s", ret)
            self.logged_running = True
        return ret

    def _terminate(self):
        """
        This method actually terminates the started daemon.
        """
        # We completely override the parent class method because we're not using the
        # self._terminal property, it's a systemd service
        if self._process is None:  # pragma: no cover
            # pylint: disable=access-member-before-definition
            if TYPE_CHECKING:
                # Make mypy happy
                assert self._terminal_result
            return self._terminal_result
            # pylint: enable=access-member-before-definition

        atexit.unregister(self.terminate)
        log.info("Stopping %s", self.factory)
        pid = self.pid
        # Collect any child processes information before terminating the process
        with contextlib.suppress(psutil.NoSuchProcess):
            if not platform.is_darwin():
                for child in psutil.Process(pid).children(recursive=True):
                    # pylint: disable=access-member-before-definition
                    if child not in self._children:
                        self._children.append(child)
                    # pylint: enable=access-member-before-definition

        if self._process.is_running():  # pragma: no cover
            cmdline = _get_cmdline(self._process)
        else:
            cmdline = []

        # Tell ssm to stop the service
        try:
            self._internal_run(
                str(self.factory.salt_pkg_install.ssm_bin),
                "stop",
                self.get_service_name(),
            )
        except FileNotFoundError:
            pass

        if self._process.is_running():  # pragma: no cover
            try:
                self._process.wait(10)
            except psutil.TimeoutExpired:
                self._process.terminate()
                try:
                    self._process.wait(10)
                except psutil.TimeoutExpired:
                    pass

        try:
            exitcode = self._process.wait(5) or 0
        except psutil.TimeoutExpired:
            exitcode = 0

        # Dereference the internal _process attribute
        self._process = None
        # Let's log and kill any child processes left behind, including the main
        # subprocess if it failed to properly stop
        terminate_process(
            pid=pid,
            kill_children=True,
            children=self._children,  # pylint: disable=access-member-before-definition
            slow_stop=self.factory.slow_stop,
        )

        # pylint: disable=access-member-before-definition
        if self._terminal_stdout is not None:
            self._terminal_stdout.close()
        if self._terminal_stderr is not None:
            self._terminal_stderr.close()
        # pylint: enable=access-member-before-definition
        stdout = stderr = ""
        try:
            self._terminal_result = ProcessResult(
                returncode=exitcode, stdout=stdout, stderr=stderr, cmdline=cmdline
            )
            log.info("%s %s", self.factory.__class__.__name__, self._terminal_result)
            return self._terminal_result
        finally:
            self._terminal = None
            self._terminal_stdout = None
            self._terminal_stderr = None
            self._terminal_timeout = None
            self._children = []


@attr.s(kw_only=True)
class PkgMixin:
    salt_pkg_install: SaltPkgInstall = attr.ib()

    def get_script_path(self):
        if platform.is_darwin() and self.salt_pkg_install.classic:
            if self.salt_pkg_install.run_root and os.path.exists(
                self.salt_pkg_install.run_root
            ):
                return str(self.salt_pkg_install.run_root)
            elif os.path.exists(self.salt_pkg_install.bin_dir / self.script_name):
                return str(self.salt_pkg_install.bin_dir / self.script_name)
            else:
                return str(self.salt_pkg_install.install_dir / self.script_name)
        return super().get_script_path()

    def cmdline(self, *args, **kwargs):
        _cmdline = super().cmdline(*args, **kwargs)
        if _cmdline[0] == self.python_executable:
            _cmdline.pop(0)
        return _cmdline


@attr.s(kw_only=True)
class DaemonPkgMixin(PkgMixin):
    def __attrs_post_init__(self):
        if not platform.is_windows() and self.salt_pkg_install.pkg_system_service:
            if platform.is_darwin():
                self.write_launchd_conf()
            else:
                self.write_systemd_conf()

    def get_service_name(self):
        return self.script_name

    def write_launchd_conf(self):
        raise NotImplementedError

    def write_systemd_conf(self):
        raise NotImplementedError


@attr.s(kw_only=True)
class SaltMaster(DaemonPkgMixin, master.SaltMaster):
    """
    Subclassed just to tweak the binary paths if needed and factory classes.
    """

    def __attrs_post_init__(self):
        self.script_name = "salt-master"
        master.SaltMaster.__attrs_post_init__(self)
        DaemonPkgMixin.__attrs_post_init__(self)

    def _get_impl_class(self):
        if self.system_service and self.salt_pkg_install.pkg_system_service:
            if platform.is_windows():
                return PkgSsmSaltDaemonImpl
            if platform.is_darwin():
                return PkgLaunchdSaltDaemonImpl
            return PkgSystemdSaltDaemonImpl
        return DaemonImpl

    def write_launchd_conf(self):
        self.salt_pkg_install.write_launchd_conf("master")

    def write_systemd_conf(self):
        self.salt_pkg_install.write_systemd_conf(
            "salt-master", self.salt_pkg_install.binary_paths["master"]
        )

    def salt_minion_daemon(self, minion_id, **kwargs):
        return super().salt_minion_daemon(
            minion_id,
            factory_class=SaltMinion,
            salt_pkg_install=self.salt_pkg_install,
            **kwargs,
        )

    def salt_api_daemon(self, **kwargs):
        return super().salt_api_daemon(
            factory_class=SaltApi, salt_pkg_install=self.salt_pkg_install, **kwargs
        )

    def salt_key_cli(self, factory_class=None, **factory_class_kwargs):
        if not factory_class:
            factory_class = SaltKey
        factory_class_kwargs["salt_pkg_install"] = self.salt_pkg_install
        return super().salt_key_cli(
            factory_class=factory_class,
            **factory_class_kwargs,
        )

    def salt_cli(self, factory_class=None, **factory_class_kwargs):
        if not factory_class:
            factory_class = SaltCli
        factory_class_kwargs["salt_pkg_install"] = self.salt_pkg_install
        return super().salt_cli(
            factory_class=factory_class,
            **factory_class_kwargs,
        )


@attr.s(kw_only=True)
class SaltMasterWindows(SaltMaster):
    """
    Subclassed just to tweak the binary paths if needed and factory classes.
    """

    def __attrs_post_init__(self):
        super().__attrs_post_init__()
        self.script_name = cli_scripts.generate_script(
            bin_dir=self.factories_manager.scripts_dir,
            script_name="salt-master",
            code_dir=self.factories_manager.code_dir.parent,
        )

    def _get_impl_class(self):
        return DaemonImpl

    def cmdline(self, *args, **kwargs):
        cmdline_ = super().cmdline(*args, **kwargs)
        if self.python_executable:
            if cmdline_[0] != self.python_executable:
                cmdline_.insert(0, self.python_executable)
        return cmdline_


@attr.s(kw_only=True, slots=True)
class SaltMinion(DaemonPkgMixin, minion.SaltMinion):
    """
    Subclassed just to tweak the binary paths if needed and factory classes.
    """

    def __attrs_post_init__(self):
        self.script_name = "salt-minion"
        minion.SaltMinion.__attrs_post_init__(self)
        DaemonPkgMixin.__attrs_post_init__(self)

    def _get_impl_class(self):
        if self.system_service and self.salt_pkg_install.pkg_system_service:
            if platform.is_windows():
                return PkgSsmSaltDaemonImpl
            if platform.is_darwin():
                return PkgLaunchdSaltDaemonImpl
            return PkgSystemdSaltDaemonImpl
        return DaemonImpl

    def write_launchd_conf(self):
        self.salt_pkg_install.write_launchd_conf("minion")

    def write_systemd_conf(self):
        self.salt_pkg_install.write_systemd_conf(
            "salt-minion", self.salt_pkg_install.binary_paths["minion"]
        )

    def salt_call_cli(self, factory_class=None, **factory_class_kwargs):
        if not factory_class:
            factory_class = SaltCall
        factory_class_kwargs["salt_pkg_install"] = self.salt_pkg_install
        return super().salt_call_cli(
            factory_class=factory_class,
            **factory_class_kwargs,
        )


@attr.s(kw_only=True, slots=True)
class SaltApi(DaemonPkgMixin, api.SaltApi):
    """
    Subclassed just to tweak the binary paths if needed.
    """

    def __attrs_post_init__(self):
        self.script_name = "salt-api"
        api.SaltApi.__attrs_post_init__(self)
        DaemonPkgMixin.__attrs_post_init__(self)

    def _get_impl_class(self):
        if self.system_service and self.salt_pkg_install.pkg_system_service:
            if platform.is_windows():
                return PkgSsmSaltDaemonImpl
            if platform.is_darwin():
                return PkgLaunchdSaltDaemonImpl
            return PkgSystemdSaltDaemonImpl
        return DaemonImpl

    def write_launchd_conf(self):
        self.salt_pkg_install.write_launchd_conf("api")

    def write_systemd_conf(self):
        self.salt_pkg_install.write_systemd_conf(
            "salt-api",
            self.salt_pkg_install.binary_paths["api"],
        )


@attr.s(kw_only=True, slots=True)
class SaltCall(PkgMixin, call.SaltCall):
    """
    Subclassed just to tweak the binary paths if needed.
    """

    def __attrs_post_init__(self):
        call.SaltCall.__attrs_post_init__(self)
        self.script_name = "salt-call"


@attr.s(kw_only=True, slots=True)
class SaltCli(PkgMixin, saltfactories.cli.salt.SaltCli):
    """
    Subclassed just to tweak the binary paths if needed.
    """

    def __attrs_post_init__(self):
        self.script_name = "salt"
        saltfactories.cli.salt.SaltCli.__attrs_post_init__(self)


@attr.s(kw_only=True, slots=True)
class SaltKey(PkgMixin, key.SaltKey):
    """
    Subclassed just to tweak the binary paths if needed.
    """

    def __attrs_post_init__(self):
        self.script_name = "salt-key"
        key.SaltKey.__attrs_post_init__(self)


@attr.s(kw_only=True, slots=True)
class ApiRequest:
    port: int = attr.ib(repr=False)
    account: TestAccount = attr.ib(repr=False)
    session: requests.Session = attr.ib(init=False, repr=False)
    api_uri: str = attr.ib(init=False)
    auth_data: dict[str, str] = attr.ib(init=False)

    @session.default
    def _default_session(self):
        return requests.Session()

    @api_uri.default
    def _default_api_uri(self):
        return f"http://localhost:{self.port}"

    @auth_data.default
    def _default_auth_data(self):
        return {
            "username": self.account.username,
            "password": self.account.password,
            "eauth": "auto",
            "out": "json",
        }

    def post(self, url, data):
        post_data = dict(**self.auth_data, **data)
        resp = self.session.post(f"{self.api_uri}/run", data=post_data).json()
        minion = next(iter(resp["return"][0]))
        return resp["return"][0][minion]

    def __enter__(self):
        self.session.__enter__()
        return self

    def __exit__(self, *args):
        self.session.__exit__(*args)