1"""
2Tests for loop state(s)
3"""
4
5import pytest
6import salt.states.loop
7from tests.support.mixins import LoaderModuleMockMixin
8from tests.support.mock import MagicMock, patch
9from tests.support.unit import TestCase
10
11
12class LoopTestCase(TestCase, LoaderModuleMockMixin):
13
14    mock = MagicMock(return_value=True)
15    func = "foo.bar"
16    m_args = ["foo", "bar", "baz"]
17    m_kwargs = {"hello": "world"}
18    condition = "m_ret is True"
19    period = 1
20    timeout = 3
21
22    def setup_loader_modules(self):
23        return {
24            salt.states.loop: {
25                "__opts__": {"test": False},
26                "__salt__": {self.func: self.mock},
27            }
28        }
29
30    def setUp(self):
31        self.mock.reset_mock()
32
33    def test_until(self):
34        ret = salt.states.loop.until(
35            name=self.func,
36            m_args=self.m_args,
37            m_kwargs=self.m_kwargs,
38            condition=self.condition,
39            period=self.period,
40            timeout=self.timeout,
41        )
42        assert ret["result"] is True
43        self.mock.assert_called_once_with(*self.m_args, **self.m_kwargs)
44
45    def test_until_without_args(self):
46        ret = salt.states.loop.until(
47            name=self.func,
48            m_kwargs=self.m_kwargs,
49            condition=self.condition,
50            period=self.period,
51            timeout=self.timeout,
52        )
53        assert ret["result"] is True
54        self.mock.assert_called_once_with(**self.m_kwargs)
55
56    def test_until_without_kwargs(self):
57        ret = salt.states.loop.until(
58            name=self.func,
59            m_args=self.m_args,
60            condition=self.condition,
61            period=self.period,
62            timeout=self.timeout,
63        )
64        assert ret["result"] is True
65        self.mock.assert_called_once_with(*self.m_args)
66
67    def test_until_without_args_or_kwargs(self):
68        ret = salt.states.loop.until(
69            name=self.func,
70            condition=self.condition,
71            period=self.period,
72            timeout=self.timeout,
73        )
74        assert ret["result"] is True
75        self.mock.assert_called_once_with()
76
77
78class LoopTestCaseNoEval(TestCase, LoaderModuleMockMixin):
79    """
80    Test cases for salt.states.loop
81    """
82
83    def setup_loader_modules(self):
84        opts = salt.config.DEFAULT_MINION_OPTS.copy()
85        utils = salt.loader.utils(opts)
86        return {salt.states.loop: {"__opts__": opts, "__utils__": utils}}
87
88    def test_test_mode(self):
89        """
90        Test response when test_mode is enabled.
91        """
92        with patch.dict(
93            salt.states.loop.__salt__, {"foo.foo": True}  # pylint: disable=no-member
94        ), patch.dict(
95            salt.states.loop.__opts__, {"test": True}  # pylint: disable=no-member
96        ):
97            self.assertDictEqual(
98                salt.states.loop.until(name="foo.foo", condition="m_ret"),
99                {
100                    "name": "foo.foo",
101                    "result": None,
102                    "changes": {},
103                    "comment": "The execution module foo.foo will be run",
104                },
105            )
106            self.assertDictEqual(
107                salt.states.loop.until_no_eval(name="foo.foo", expected=2),
108                {
109                    "name": "foo.foo",
110                    "result": None,
111                    "changes": {},
112                    "comment": 'Would have waited for "foo.foo" to produce "2".',
113                },
114            )
115
116    @pytest.mark.slow_test
117    def test_immediate_success(self):
118        """
119        Test for an immediate success.
120        """
121        with patch.dict(
122            salt.states.loop.__salt__,
123            {  # pylint: disable=no-member
124                "foo.foo": lambda: 2,
125                "foo.baz": lambda x, y: True,
126            },
127        ), patch.dict(
128            salt.states.loop.__utils__,
129            {"foo.baz": lambda x, y: True},  # pylint: disable=no-member
130        ):
131            self.assertDictEqual(
132                salt.states.loop.until(name="foo.foo", condition="m_ret"),
133                {
134                    "name": "foo.foo",
135                    "result": True,
136                    "changes": {},
137                    "comment": "Condition m_ret was met",
138                },
139            )
140            # Using default compare_operator 'eq'
141            self.assertDictEqual(
142                salt.states.loop.until_no_eval(name="foo.foo", expected=2),
143                {
144                    "name": "foo.foo",
145                    "result": True,
146                    "changes": {},
147                    "comment": "Call provided the expected results in 1 attempts",
148                },
149            )
150            # Using compare_operator 'gt'
151            self.assertDictEqual(
152                salt.states.loop.until_no_eval(
153                    name="foo.foo", expected=1, compare_operator="gt"  # Returns 2
154                ),
155                {
156                    "name": "foo.foo",
157                    "result": True,
158                    "changes": {},
159                    "comment": "Call provided the expected results in 1 attempts",
160                },
161            )
162            # Using compare_operator 'ne'
163            self.assertDictEqual(
164                salt.states.loop.until_no_eval(
165                    name="foo.foo", expected=3, compare_operator="ne"  # Returns 2
166                ),
167                {
168                    "name": "foo.foo",
169                    "result": True,
170                    "changes": {},
171                    "comment": "Call provided the expected results in 1 attempts",
172                },
173            )
174            # Using __utils__['foo.baz'] as compare_operator
175            self.assertDictEqual(
176                salt.states.loop.until_no_eval(
177                    name="foo.foo",
178                    expected="anything, mocked compare_operator returns True anyway",
179                    compare_operator="foo.baz",
180                ),
181                {
182                    "name": "foo.foo",
183                    "result": True,
184                    "changes": {},
185                    "comment": "Call provided the expected results in 1 attempts",
186                },
187            )
188            # Using __salt__['foo.baz]' as compare_operator
189            self.assertDictEqual(
190                salt.states.loop.until_no_eval(
191                    name="foo.foo",
192                    expected="anything, mocked compare_operator returns True anyway",
193                    compare_operator="foo.baz",
194                ),
195                {
196                    "name": "foo.foo",
197                    "result": True,
198                    "changes": {},
199                    "comment": "Call provided the expected results in 1 attempts",
200                },
201            )
202
203    def test_immediate_failure(self):
204        """
205        Test for an immediate failure.
206        Period and timeout will be set to 0.01 to assume one attempt.
207        """
208        with patch.dict(
209            salt.states.loop.__salt__, {"foo.bar": lambda: False}
210        ):  # pylint: disable=no-member
211            self.assertDictEqual(
212                salt.states.loop.until(
213                    name="foo.bar", condition="m_ret", period=0.01, timeout=0.01
214                ),
215                {
216                    "name": "foo.bar",
217                    "result": False,
218                    "changes": {},
219                    "comment": "Timed out while waiting for condition m_ret",
220                },
221            )
222
223            self.assertDictEqual(
224                salt.states.loop.until_no_eval(
225                    name="foo.bar", expected=True, period=0.01, timeout=0.01
226                ),
227                {
228                    "name": "foo.bar",
229                    "result": False,
230                    "changes": {},
231                    "comment": (
232                        "Call did not produce the expected result after 1 attempts"
233                    ),
234                },
235            )
236
237    def test_eval_exceptions(self):
238        """
239        Test a couple of eval exceptions.
240        """
241        with patch.dict(
242            salt.states.loop.__salt__, {"foo.bar": lambda: None}
243        ):  # pylint: disable=no-member
244            self.assertRaises(
245                SyntaxError,
246                salt.states.loop.until,
247                name="foo.bar",
248                condition='raise NameError("FOO")',
249            )
250            self.assertRaises(
251                NameError, salt.states.loop.until, name="foo.bar", condition="foo"
252            )
253
254    def test_no_eval_exceptions(self):
255        """
256        Test exception handling in until_no_eval.
257        """
258        with patch.dict(
259            salt.states.loop.__salt__,  # pylint: disable=no-member
260            {"foo.bar": MagicMock(side_effect=KeyError("FOO"))},
261        ):
262            self.assertDictEqual(
263                salt.states.loop.until_no_eval(name="foo.bar", expected=True),
264                {
265                    "name": "foo.bar",
266                    "result": False,
267                    "changes": {},
268                    "comment": (
269                        "Exception occurred while executing foo.bar: {}:{}".format(
270                            type(KeyError()), "'FOO'"
271                        )
272                    ),
273                },
274            )
275
276    def test_retried_success(self):
277        """
278        Test if the function does indeed return after a fixed amount of retries.
279
280        Note: Do not merge these two tests in one with-block, as the side_effect
281        iterator is shared.
282        """
283        with patch.dict(
284            salt.states.loop.__salt__,  # pylint: disable=no-member
285            {
286                "foo.bar": MagicMock(side_effect=range(1, 7))
287            },  # pylint: disable=incompatible-py3-code
288        ):
289            self.assertDictEqual(
290                salt.states.loop.until(
291                    name="foo.bar", condition="m_ret == 5", period=0, timeout=1
292                ),
293                {
294                    "name": "foo.bar",
295                    "result": True,
296                    "changes": {},
297                    "comment": "Condition m_ret == 5 was met",
298                },
299            )
300
301        with patch.dict(
302            salt.states.loop.__salt__,  # pylint: disable=no-member
303            {
304                "foo.bar": MagicMock(side_effect=range(1, 7))
305            },  # pylint: disable=incompatible-py3-code
306        ):
307            self.assertDictEqual(
308                salt.states.loop.until_no_eval(
309                    name="foo.bar", expected=5, period=0, timeout=1
310                ),
311                {
312                    "name": "foo.bar",
313                    "result": True,
314                    "changes": {},
315                    "comment": "Call provided the expected results in 5 attempts",
316                },
317            )
318
319    def test_retried_failure(self):
320        """
321        Test if the function fails after the designated timeout.
322        """
323        with patch.dict(
324            salt.states.loop.__salt__,  # pylint: disable=no-member
325            {
326                "foo.bar": MagicMock(side_effect=range(1, 7))
327            },  # pylint: disable=incompatible-py3-code
328        ):
329            self.assertDictEqual(
330                salt.states.loop.until(
331                    name="foo.bar", condition="m_ret == 5", period=0.01, timeout=0.03
332                ),
333                {
334                    "name": "foo.bar",
335                    "result": False,
336                    "changes": {},
337                    "comment": "Timed out while waiting for condition m_ret == 5",
338                },
339            )
340
341        # period and timeout below has been increased to keep windows machines from
342        # returning a lower number of attempts (because it's slower).
343        with patch.dict(
344            salt.states.loop.__salt__,  # pylint: disable=no-member
345            {
346                "foo.bar": MagicMock(side_effect=range(1, 7))
347            },  # pylint: disable=incompatible-py3-code
348        ):
349            self.assertDictEqual(
350                salt.states.loop.until_no_eval(
351                    name="foo.bar", expected=5, period=0.2, timeout=0.5
352                ),
353                {
354                    "name": "foo.bar",
355                    "result": False,
356                    "changes": {},
357                    "comment": (
358                        "Call did not produce the expected result after 3 attempts"
359                    ),
360                },
361            )
362