tests/pytests/functional/modules/state/requisites/test_watch.py - master - githubmirror/salt

tests/pytests/functional/modules/state/requisites/test_watch.py
280 строк8 KB
Mitar
Allow watching states to opt-in to mod_watch when normal run had changes (#68960)
14 июл 2026, 23:17
14 июл 2026, 23:17c8f7e9a
import pytest

import salt.utils.platform

from . import normalize_ret

pytestmark = [
    pytest.mark.windows_whitelisted,
    pytest.mark.core_test,
]


def test_watch_in(state, state_tree):
    """
    test watch_in requisite when there is a success
    """
    sls_contents = """
    return_changes:
      test.succeed_with_changes:
        - watch_in:
          - test: watch_states

    watch_states:
      test.succeed_without_changes
    """
    changes = "test_|-return_changes_|-return_changes_|-succeed_with_changes"
    watch = "test_|-watch_states_|-watch_states_|-succeed_without_changes"
    with pytest.helpers.temp_file("requisite.sls", sls_contents, state_tree):
        ret = state.sls("requisite")
        assert ret[changes].full_return["__run_num__"] == 0
        assert ret[changes].changes["testing"]["new"] == "Something pretended to change"
        assert ret[watch].full_return["__run_num__"] == 2
        assert ret[watch].comment == "Watch statement fired."


def test_watch_in_failure(state, state_tree):
    """
    test watch_in requisite when there is a failure
    """
    sls_contents = """
    return_changes:
      test.fail_with_changes:
        - watch_in:
          - test: watch_states

    watch_states:
      test.succeed_without_changes
    """
    fail = "test_|-return_changes_|-return_changes_|-fail_with_changes"
    watch = "test_|-watch_states_|-watch_states_|-succeed_without_changes"
    with pytest.helpers.temp_file("requisite.sls", sls_contents, state_tree):
        ret = state.sls("requisite")
        assert ret[fail].result is False
        assert (
            ret[watch].comment
            == "One or more requisite failed: requisite.return_changes"
        )


def test_requisites_watch_any(state, state_tree):
    """
    Call sls file containing several require_in and require.

    Ensure that some of them are failing and that the order is right.
    """
    if salt.utils.platform.is_windows():
        cmd_true = "exit"
        cmd_false = "exit /B 1"
    else:
        cmd_true = "true"
        cmd_false = "false"
    sls_contents = """
    A:
      cmd.wait:
        - name: '{cmd_true}'
        - watch_any:
          - cmd: B
          - cmd: C
          - cmd: D

    B:
      cmd.run:
        - name: '{cmd_true}'

    C:
      cmd.run:
        - name: '{cmd_false}'

    D:
      cmd.run:
        - name: '{cmd_true}'

    E:
      cmd.wait:
        - name: '{cmd_true}'
        - watch_any:
          - cmd: F
          - cmd: G
          - cmd: H

    F:
      cmd.run:
        - name: '{cmd_true}'

    G:
      cmd.run:
        - name: '{cmd_false}'

    H:
      cmd.run:
        - name: '{cmd_false}'
    """.format(
        cmd_true=cmd_true, cmd_false=cmd_false
    )
    expected_result = {
        f"cmd_|-A_|-{cmd_true}_|-wait": {
            "__run_num__": 4,
            "comment": f'Command "{cmd_true}" run',
            "result": True,
            "changes": True,
        },
        f"cmd_|-B_|-{cmd_true}_|-run": {
            "__run_num__": 0,
            "comment": f'Command "{cmd_true}" run',
            "result": True,
            "changes": True,
        },
        f"cmd_|-C_|-{cmd_false}_|-run": {
            "__run_num__": 1,
            "comment": f'Command "{cmd_false}" run',
            "result": False,
            "changes": True,
        },
        f"cmd_|-D_|-{cmd_true}_|-run": {
            "__run_num__": 2,
            "comment": f'Command "{cmd_true}" run',
            "result": True,
            "changes": True,
        },
        f"cmd_|-E_|-{cmd_true}_|-wait": {
            "__run_num__": 9,
            "comment": f'Command "{cmd_true}" run',
            "result": True,
            "changes": True,
        },
        f"cmd_|-F_|-{cmd_true}_|-run": {
            "__run_num__": 5,
            "comment": f'Command "{cmd_true}" run',
            "result": True,
            "changes": True,
        },
        f"cmd_|-G_|-{cmd_false}_|-run": {
            "__run_num__": 6,
            "comment": f'Command "{cmd_false}" run',
            "result": False,
            "changes": True,
        },
        f"cmd_|-H_|-{cmd_false}_|-run": {
            "__run_num__": 7,
            "comment": f'Command "{cmd_false}" run',
            "result": False,
            "changes": True,
        },
    }
    with pytest.helpers.temp_file("requisite.sls", sls_contents, state_tree):
        ret = state.sls("requisite")
        result = normalize_ret(ret.raw)
        assert result == expected_result


def test_requisites_watch_any_fail(state, state_tree):
    """
    Call sls file containing several require_in and require.

    Ensure that some of them are failing and that the order is right.
    """
    sls_contents = """
    A:
      cmd.wait:
        - name: 'true'
        - watch_any:
          - cmd: B
          - cmd: C

    B:
      cmd.run:
        - name: 'false'

    C:
      cmd.run:
        - name: 'false'
    """
    with pytest.helpers.temp_file("requisite.sls", sls_contents, state_tree):
        ret = state.sls("requisite")
        assert "One or more requisite failed" in ret["cmd_|-A_|-true_|-wait"].comment


def test_issue_30820_requisite_in_match_by_name(state, state_tree):
    """
    This tests the case where a requisite_in matches by name instead of ID

    See https://github.com/saltstack/salt/issues/30820 for more info
    """
    sls_contents = """
    bar state:
      cmd.wait:
        - name: 'echo bar'

    echo foo:
      cmd.run:
        - watch_in:
          - cmd: 'echo bar'
    """
    bar_state = "cmd_|-bar state_|-echo bar_|-wait"
    with pytest.helpers.temp_file("requisite.sls", sls_contents, state_tree):
        ret = state.sls("requisite")
        assert bar_state in ret
        assert ret[bar_state].comment == 'Command "echo bar" run'


def test_watch_skips_mod_watch_when_normal_run_has_changes(state, state_tree):
    """
    By default, if the watcher's normal run already produces changes,
    ``mod_watch`` is not invoked. This preserves the long-standing Salt
    behavior that non-idempotent watchers (such as ``cmd.run``, whose
    ``mod_watch`` re-executes the command) do not run twice per state
    apply.
    """
    sls_contents = """
    trigger:
      test.succeed_with_changes

    watcher:
      test.succeed_with_changes:
        - watch:
          - test: trigger
    """
    watcher = "test_|-watcher_|-watcher_|-succeed_with_changes"
    with pytest.helpers.temp_file("requisite.sls", sls_contents, state_tree):
        ret = state.sls("requisite")
        # The normal run's own change is present.
        assert "testing" in ret[watcher].changes
        # mod_watch did not fire, so no "Requisites with changes" entry and
        # no "Watch statement fired." comment were appended.
        assert "Requisites with changes" not in ret[watcher].changes
        assert "Watch statement fired." not in ret[watcher].comment
        assert ret[watcher].result is True


def test_watch_fires_when_force_mod_watch_is_set(state, state_tree):
    """
    A watcher's normal return may set ``force_mod_watch: True`` to signal
    that its own changes do not subsume the work of ``mod_watch`` (for
    example, ``docker_container.running`` performing a network reconnect
    without recreating the container while a watched config file has
    changed). When the flag is set, ``mod_watch`` must fire in addition to
    the normal run and its result must be merged in so neither set of
    changes is lost.
    """
    sls_contents = """
    trigger:
      test.succeed_with_changes

    watcher:
      test.succeed_with_changes:
        - force_mod_watch: True
        - watch:
          - test: trigger
    """
    watcher = "test_|-watcher_|-watcher_|-succeed_with_changes"
    with pytest.helpers.temp_file("requisite.sls", sls_contents, state_tree):
        ret = state.sls("requisite")
        # The normal run's own change must still be present.
        assert "testing" in ret[watcher].changes
        # And mod_watch must also have fired, appending its own change entry
        # and comment to the result so the watched-state trigger is not lost.
        assert "Requisites with changes" in ret[watcher].changes
        assert ret[watcher].changes["Requisites with changes"] == ["test: trigger"]
        assert "Watch statement fired." in ret[watcher].comment
        assert ret[watcher].result is True