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