tests/pytests/unit/modules/test_disk.py - master - githubmirror/salt

tests/pytests/unit/modules/test_disk.py
446 строк13 KB
Zain Asif
Fix byte conversion factors off by a power of 10 in disk._parse_numbers (#69680)
05 июл 2026, 09:54
05 июл 2026, 09:547b5028e
"""
:codeauthor: Jayesh Kariya <jayeshk@saltstack.com>
"""

import decimal

import pytest

import salt.modules.disk as disk
from salt.exceptions import CommandExecutionError, SaltInvocationError
from tests.support.mock import MagicMock, patch


@pytest.fixture
def stub_disk_usage():
    return {
        "/": {
            "filesystem": None,
            "1K-blocks": 10000,
            "used": 10000,
            "available": 10000,
            "capacity": 10000,
        },
        "/dev": {
            "filesystem": None,
            "1K-blocks": 10000,
            "used": 10000,
            "available": 10000,
            "capacity": 10000,
        },
        "/run": {
            "filesystem": None,
            "1K-blocks": 10000,
            "used": 10000,
            "available": 10000,
            "capacity": 10000,
        },
        "/run/lock": {
            "filesystem": None,
            "1K-blocks": 10000,
            "used": 10000,
            "available": 10000,
            "capacity": 10000,
        },
        "/run/shm": {
            "filesystem": None,
            "1K-blocks": 10000,
            "used": 10000,
            "available": 10000,
            "capacity": 10000,
        },
        "/run/user": {
            "filesystem": None,
            "1K-blocks": 10000,
            "used": 10000,
            "available": 10000,
            "capacity": 10000,
        },
        "/sys/fs/cgroup": {
            "filesystem": None,
            "1K-blocks": 10000,
            "used": 10000,
            "available": 10000,
            "capacity": 10000,
        },
    }


@pytest.fixture
def stub_disk_inodeusage():
    return {
        "/": {
            "inodes": 10000,
            "used": 10000,
            "free": 10000,
            "use": 10000,
            "filesystem": None,
        },
        "/dev": {
            "inodes": 10000,
            "used": 10000,
            "free": 10000,
            "use": 10000,
            "filesystem": None,
        },
        "/run": {
            "inodes": 10000,
            "used": 10000,
            "free": 10000,
            "use": 10000,
            "filesystem": None,
        },
        "/run/lock": {
            "inodes": 10000,
            "used": 10000,
            "free": 10000,
            "use": 10000,
            "filesystem": None,
        },
        "/run/shm": {
            "inodes": 10000,
            "used": 10000,
            "free": 10000,
            "use": 10000,
            "filesystem": None,
        },
        "/run/user": {
            "inodes": 10000,
            "used": 10000,
            "free": 10000,
            "use": 10000,
            "filesystem": None,
        },
        "/sys/fs/cgroup": {
            "inodes": 10000,
            "used": 10000,
            "free": 10000,
            "use": 10000,
            "filesystem": None,
        },
    }


@pytest.fixture
def stub_disk_percent():
    return {
        "/": 50,
        "/dev": 10,
        "/run": 10,
        "/run/lock": 10,
        "/run/shm": 10,
        "/run/user": 10,
        "/sys/fs/cgroup": 10,
    }


@pytest.fixture
def stub_disk_blkid():
    return {"/dev/sda": {"TYPE": "ext4", "UUID": None}}


@pytest.fixture
def configure_loader_modules():
    return {disk: {}}


def test_usage_dict(stub_disk_usage):
    with patch.dict(disk.__grains__, {"kernel": "Linux"}), patch(
        "salt.modules.disk.usage", MagicMock(return_value=stub_disk_usage)
    ):
        mock_cmd = MagicMock(return_value=1)
        with patch.dict(disk.__salt__, {"cmd.run": mock_cmd}):
            assert stub_disk_usage == disk.usage(args=None)


def test_usage_none():
    with patch.dict(disk.__grains__, {"kernel": "Linux"}), patch(
        "salt.modules.disk.usage", MagicMock(return_value="")
    ):
        mock_cmd = MagicMock(return_value=1)
        with patch.dict(disk.__salt__, {"cmd.run": mock_cmd}):
            assert "" == disk.usage(args=None)


def test_inodeusage(stub_disk_inodeusage):
    with patch.dict(disk.__grains__, {"kernel": "OpenBSD"}), patch(
        "salt.modules.disk.inodeusage", MagicMock(return_value=stub_disk_inodeusage)
    ):
        mock = MagicMock()
        with patch.dict(disk.__salt__, {"cmd.run": mock}):
            assert stub_disk_inodeusage == disk.inodeusage(args=None)


def test_percent(stub_disk_percent):
    with patch.dict(disk.__grains__, {"kernel": "Linux"}), patch(
        "salt.modules.disk.percent", MagicMock(return_value=stub_disk_percent)
    ):
        mock = MagicMock()
        with patch.dict(disk.__salt__, {"cmd.run": mock}):
            assert stub_disk_percent == disk.percent(args=None)


def test_percent_args():
    with patch.dict(disk.__grains__, {"kernel": "Linux"}), patch(
        "salt.modules.disk.percent", MagicMock(return_value="/")
    ):
        mock = MagicMock()
        with patch.dict(disk.__salt__, {"cmd.run": mock}):
            assert "/" == disk.percent("/")


def test_blkid(stub_disk_blkid):
    with patch.dict(
        disk.__salt__, {"cmd.run_stdout": MagicMock(return_value=1)}
    ), patch("salt.modules.disk.blkid", MagicMock(return_value=stub_disk_blkid)):
        assert stub_disk_blkid == disk.blkid()


@pytest.mark.skip_on_windows(reason="Skip on Windows")
@pytest.mark.skip_on_darwin(reason="Skip on Darwin")
@pytest.mark.skip_on_freebsd
def test_blkid_token():
    run_stdout_mock = MagicMock(return_value={"retcode": 1})
    with patch.dict(disk.__salt__, {"cmd.run_all": run_stdout_mock}):
        disk.blkid(token="TYPE=ext4")
        run_stdout_mock.assert_called_with(
            ["blkid", "-t", "TYPE=ext4"], python_shell=False
        )


def test_dump():
    mock = MagicMock(return_value={"retcode": 0, "stdout": ""})
    with patch.dict(disk.__salt__, {"cmd.run_all": mock}):
        disk.dump("/dev/sda")
        mock.assert_called_once_with(
            "blockdev --getro --getsz --getss --getpbsz --getiomin "
            "--getioopt --getalignoff --getmaxsect --getsize "
            "--getsize64 --getra --getfra /dev/sda",
            python_shell=False,
        )


def test_wipe():
    mock = MagicMock(return_value={"retcode": 0, "stdout": ""})
    with patch.dict(disk.__salt__, {"cmd.run_all": mock}):
        disk.wipe("/dev/sda")
        mock.assert_called_once_with("wipefs -a /dev/sda", python_shell=False)


def test_tune():
    mock = MagicMock(
        return_value={
            "retcode": 0,
            "stdout": "712971264\n512\n512\n512\n0\n0\n88\n712971264\n365041287168\n512\n512",
        }
    )
    with patch.dict(disk.__salt__, {"cmd.run_all": mock}):
        mock_dump = MagicMock(return_value={"retcode": 0, "stdout": ""})
        with patch("salt.modules.disk.dump", mock_dump):
            kwargs = {"read-ahead": 512, "filesystem-read-ahead": 1024}
            disk.tune("/dev/sda", **kwargs)

            mock.assert_called_with(
                "blockdev --setra 512 --setfra 1024 /dev/sda", python_shell=False
            )


@pytest.mark.parametrize(
    "kwarg_key,kwarg_value,expected_flag",
    [
        ("read-write", True, "setrw"),
        ("read-only", True, "setro"),
    ],
)
def test_tune_issue_68490(kwarg_key, kwarg_value, expected_flag):
    """
    Test for issue #68490 where blockdev was not called with correct flags
    for read-write (--setrw) and read-only (--setro)
    """
    mock = MagicMock(return_value={"retcode": 0, "stdout": ""})
    with patch.dict(disk.__salt__, {"cmd.run_all": mock}):
        mock_dump = MagicMock(return_value={"retcode": 0, "stdout": ""})
        with patch("salt.modules.disk.dump", mock_dump):
            kwargs = {kwarg_key: kwarg_value}
            disk.tune("/dev/sda", **kwargs)

            mock.assert_called_with(
                f"blockdev --{expected_flag} /dev/sda", python_shell=False
            )


def test_tune_invalid_option():
    """
    Test for passing invalid option to disk.tune
    """
    with pytest.raises(SaltInvocationError) as excinfo:
        disk.tune("/dev/sda", invalid_option=True)


@pytest.mark.parametrize(
    "kwarg_key, kwarg_value, raise_error",
    [
        ("read-only", True, False),
        ("read-only", False, True),
        ("read-write", True, False),
        ("read-write", False, True),
    ],
)
def test_tune_true_only_options(kwarg_key, kwarg_value, raise_error):
    """
    Test for options that should only be set to True for disk.tune
    """
    mock = MagicMock(return_value={"retcode": 0, "stdout": ""})
    with patch.dict(disk.__salt__, {"cmd.run_all": mock}):
        mock_dump = MagicMock(return_value={"retcode": 0, "stdout": ""})
        with patch("salt.modules.disk.dump", mock_dump):
            kwargs = {kwarg_key: kwarg_value}
            if raise_error:
                with pytest.raises(SaltInvocationError) as excinfo:
                    disk.tune("/dev/sda", **kwargs)
                return
            else:
                result = disk.tune("/dev/sda", **kwargs)
                assert result["retcode"] == 0


def test_tune_blockdev_error():
    """
    Test for blockdev command returning an error in disk.tune
    """
    mock = MagicMock(return_value={"retcode": 1, "stderr": "Some error"})
    with patch.dict(disk.__salt__, {"cmd.run_all": mock}):
        mock_dump = MagicMock(return_value={"retcode": 0, "stdout": ""})
        with patch("salt.modules.disk.dump", mock_dump):
            kwargs = {"read-ahead": 512}
            with pytest.raises(CommandExecutionError) as excinfo:
                disk.tune("/dev/sda", **kwargs)


def test_format():
    """
    unit tests for disk.format
    """
    device = "/dev/sdX1"
    mock = MagicMock(return_value=0)
    with patch.dict(disk.__salt__, {"cmd.retcode": mock}), patch(
        "salt.utils.path.which", MagicMock(return_value=True)
    ):
        assert disk.format_(device) is True


def test_fat_format():
    """
    unit tests for disk.format when using fat argument
    """
    device = "/dev/sdX1"
    expected = ["mkfs", "-t", "fat", "-F", 12, "/dev/sdX1"]
    mock = MagicMock(return_value=0)
    with patch.dict(disk.__salt__, {"cmd.retcode": mock}), patch(
        "salt.utils.path.which", MagicMock(return_value=True)
    ):
        assert disk.format_(device, fs_type="fat", fat=12) is True
        args, kwargs = mock.call_args_list[0]
        assert expected == args[0]


@pytest.mark.skip_if_binaries_missing("lsblk", "df", check_all=True)
def test_fstype():
    """
    unit tests for disk.fstype
    """
    device = "/dev/sdX1"
    fs_type = "ext4"
    mock = MagicMock(return_value=f"FSTYPE\n{fs_type}")
    with patch.dict(disk.__grains__, {"kernel": "Linux"}), patch.dict(
        disk.__salt__, {"cmd.run": mock}
    ), patch("salt.utils.path.which", MagicMock(return_value=True)):
        assert disk.fstype(device) == fs_type


def test_resize2fs():
    """
    unit tests for disk.resize2fs
    """
    device = "/dev/sdX1"
    mock = MagicMock()
    with patch.dict(disk.__salt__, {"cmd.run_all": mock}), patch(
        "salt.utils.path.which", MagicMock(return_value=True)
    ):
        disk.resize2fs(device)
        mock.assert_called_once_with(f"resize2fs {device}", python_shell=False)


@pytest.mark.skip_on_windows(reason="Skip on Windows")
@pytest.mark.skip_if_binaries_missing("mkfs")
def test_format_():
    """
    unit tests for disk.format_
    """
    device = "/dev/sdX1"
    mock = MagicMock(return_value=0)
    with patch.dict(disk.__salt__, {"cmd.retcode": mock}):
        disk.format_(device=device)
        mock.assert_any_call(["mkfs", "-t", "ext4", device], ignore_retcode=True)


@pytest.mark.skip_on_windows(reason="Skip on Windows")
@pytest.mark.skip_if_binaries_missing("mkfs")
def test_format__nodiscard_ext():
    """
    unit tests for disk.format_ with discard=False on an ext filesystem
    """
    device = "/dev/sdX1"
    mock = MagicMock(return_value=0)
    with patch.dict(disk.__salt__, {"cmd.retcode": mock}):
        disk.format_(device=device, fs_type="ext4", discard=False)
        mock.assert_any_call(
            ["mkfs", "-t", "ext4", "-E", "nodiscard", device], ignore_retcode=True
        )


@pytest.mark.skip_on_windows(reason="Skip on Windows")
@pytest.mark.skip_if_binaries_missing("mkfs")
def test_format__nodiscard_xfs():
    """
    unit tests for disk.format_ with discard=False on an xfs filesystem
    """
    device = "/dev/sdX1"
    mock = MagicMock(return_value=0)
    with patch.dict(disk.__salt__, {"cmd.retcode": mock}):
        disk.format_(device=device, fs_type="xfs", discard=False)
        mock.assert_any_call(["mkfs", "-t", "xfs", "-K", device], ignore_retcode=True)


@pytest.mark.parametrize(
    "text, expected",
    [
        ("1.0K", decimal.Decimal("1000.0")),
        ("32.8K", decimal.Decimal("32800.0")),
        ("1.0M", decimal.Decimal("1000000.0")),
        ("1.0G", decimal.Decimal("1000000000.0")),
        ("100", decimal.Decimal("100")),
    ],
)
def test_parse_numbers_issue_65490(text, expected):
    """
    Test for issue #65490 where postfixed values returned by
    _parse_numbers were off by a factor of 10 (e.g. "1.0K" was
    parsed as 10000.0 instead of 1000.0)
    """
    assert disk._parse_numbers(text) == expected


@pytest.mark.skip_on_windows(reason="Skip on Windows")
@pytest.mark.skip_if_binaries_missing("mkfs")
def test_format__fat():
    """
    unit tests for disk.format_ with FAT parameter
    """
    device = "/dev/sdX1"
    mock = MagicMock(return_value=0)
    with patch.dict(disk.__salt__, {"cmd.retcode": mock}):
        disk.format_(device=device, fs_type="fat", fat=12)
        mock.assert_any_call(
            ["mkfs", "-t", "fat", "-F", 12, device], ignore_retcode=True
        )