"Fossies" - the Fresh Open Source Software Archive 
Member "salt-3002.2/tests/unit/states/test_file.py" (18 Nov 2020, 130724 Bytes) of package /linux/misc/salt-3002.2.tar.gz:
As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style:
standard) with prefixed line numbers.
Alternatively you can here
view or
download the uninterpreted source code file.
See also the latest
Fossies "Diffs" side-by-side code changes report for "test_file.py":
3002.1_vs_3002.2.
1 import collections
2 import logging
3 import os
4 import plistlib
5 import pprint
6 import shutil
7 from datetime import datetime
8
9 import msgpack
10 import salt.modules.file as filemod
11 import salt.serializers.json as jsonserializer
12 import salt.serializers.msgpack as msgpackserializer
13 import salt.serializers.plist as plistserializer
14 import salt.serializers.python as pythonserializer
15 import salt.serializers.yaml as yamlserializer
16 import salt.states.file as filestate
17 import salt.utils.files
18 import salt.utils.json
19 import salt.utils.platform
20 import salt.utils.win_functions
21 import salt.utils.yaml
22 from salt.exceptions import CommandExecutionError
23 from salt.ext.six.moves import range
24 from tests.support.helpers import dedent, destructiveTest, slowTest, with_tempfile
25 from tests.support.mixins import LoaderModuleMockMixin
26 from tests.support.mock import MagicMock, Mock, call, mock_open, patch
27 from tests.support.runtests import RUNTIME_VARS
28 from tests.support.unit import TestCase, skipIf
29
30 try:
31 from dateutil.relativedelta import relativedelta
32
33 HAS_DATEUTIL = True
34 except ImportError:
35 HAS_DATEUTIL = False
36
37 NO_DATEUTIL_REASON = "python-dateutil is not installed"
38
39
40 log = logging.getLogger(__name__)
41
42
43 class TestFileState(TestCase, LoaderModuleMockMixin):
44 def setup_loader_modules(self):
45 return {
46 filestate: {
47 "__env__": "base",
48 "__salt__": {"file.manage_file": False},
49 "__serializers__": {
50 "yaml.serialize": yamlserializer.serialize,
51 "python.serialize": pythonserializer.serialize,
52 "json.serialize": jsonserializer.serialize,
53 "plist.serialize": plistserializer.serialize,
54 "msgpack.serialize": msgpackserializer.serialize,
55 },
56 "__opts__": {"test": False, "cachedir": ""},
57 "__instance_id__": "",
58 "__low__": {},
59 "__utils__": {},
60 }
61 }
62
63 def tearDown(self):
64 remove_dir = "/tmp/etc"
65 if salt.utils.platform.is_windows():
66 remove_dir = "c:\\tmp\\etc"
67 try:
68 salt.utils.files.rm_rf(remove_dir)
69 except OSError:
70 pass
71
72 def test_serialize(self):
73 def returner(contents, *args, **kwargs):
74 returner.returned = contents
75
76 returner.returned = None
77
78 with patch.dict(filestate.__salt__, {"file.manage_file": returner}):
79
80 dataset = {"foo": True, "bar": 42, "baz": [1, 2, 3], "qux": 2.0}
81
82 # If no serializer passed, result should be serialized as YAML
83 filestate.serialize("/tmp", dataset)
84 self.assertEqual(salt.utils.yaml.safe_load(returner.returned), dataset)
85
86 # If serializer and formatter passed, state should not proceed.
87 ret = filestate.serialize(
88 "/tmp", dataset, serializer="yaml", formatter="json"
89 )
90 assert ret["result"] is False
91 assert (
92 ret["comment"] == "Only one of serializer and formatter are allowed"
93 ), ret
94
95 # YAML
96 filestate.serialize("/tmp", dataset, serializer="yaml")
97 self.assertEqual(salt.utils.yaml.safe_load(returner.returned), dataset)
98 filestate.serialize("/tmp", dataset, formatter="yaml")
99 self.assertEqual(salt.utils.yaml.safe_load(returner.returned), dataset)
100
101 # JSON
102 filestate.serialize("/tmp", dataset, serializer="json")
103 self.assertEqual(salt.utils.json.loads(returner.returned), dataset)
104 filestate.serialize("/tmp", dataset, formatter="json")
105 self.assertEqual(salt.utils.json.loads(returner.returned), dataset)
106
107 # plist
108 filestate.serialize("/tmp", dataset, serializer="plist")
109 self.assertEqual(plistlib.loads(returner.returned), dataset)
110 filestate.serialize("/tmp", dataset, formatter="plist")
111 self.assertEqual(plistlib.loads(returner.returned), dataset)
112
113 # Python
114 filestate.serialize("/tmp", dataset, serializer="python")
115 self.assertEqual(returner.returned, pprint.pformat(dataset) + "\n")
116 filestate.serialize("/tmp", dataset, formatter="python")
117 self.assertEqual(returner.returned, pprint.pformat(dataset) + "\n")
118
119 # msgpack
120 filestate.serialize("/tmp", dataset, serializer="msgpack")
121 self.assertEqual(returner.returned, msgpack.packb(dataset))
122 filestate.serialize("/tmp", dataset, formatter="msgpack")
123 self.assertEqual(returner.returned, msgpack.packb(dataset))
124
125 mock_serializer = Mock(return_value="")
126 with patch.dict(
127 filestate.__serializers__, {"json.serialize": mock_serializer}
128 ):
129 # Test with "serializer" arg
130 filestate.serialize(
131 "/tmp", dataset, formatter="json", serializer_opts=[{"indent": 8}]
132 )
133 mock_serializer.assert_called_with(
134 dataset, indent=8, separators=(",", ": "), sort_keys=True
135 )
136 # Test with "formatter" arg
137 mock_serializer.reset_mock()
138 filestate.serialize(
139 "/tmp", dataset, formatter="json", serializer_opts=[{"indent": 8}]
140 )
141 mock_serializer.assert_called_with(
142 dataset, indent=8, separators=(",", ": "), sort_keys=True
143 )
144
145 def test_contents_and_contents_pillar(self):
146 def returner(contents, *args, **kwargs):
147 returner.returned = contents
148
149 returner.returned = None
150
151 manage_mode_mock = MagicMock()
152 with patch.dict(
153 filestate.__salt__,
154 {"file.manage_file": returner, "config.manage_mode": manage_mode_mock},
155 ):
156
157 ret = filestate.managed(
158 "/tmp/foo", contents="hi", contents_pillar="foo:bar"
159 )
160 self.assertEqual(False, ret["result"])
161
162 def test_contents_pillar_doesnt_add_more_newlines(self):
163 # make sure the newline
164 pillar_value = "i am the pillar value{}".format(os.linesep)
165
166 self.run_contents_pillar(pillar_value, expected=pillar_value)
167
168 def run_contents_pillar(self, pillar_value, expected):
169 returner = MagicMock(return_value=None)
170 path = "/tmp/foo"
171 pillar_path = "foo:bar"
172
173 # the values don't matter here
174 pillar_mock = MagicMock(return_value=pillar_value)
175 with patch.dict(
176 filestate.__salt__,
177 {
178 "file.manage_file": returner,
179 "config.manage_mode": MagicMock(),
180 "file.source_list": MagicMock(return_value=[None, None]),
181 "file.get_managed": MagicMock(return_value=[None, None, None]),
182 "pillar.get": pillar_mock,
183 },
184 ):
185
186 ret = filestate.managed(path, contents_pillar=pillar_path)
187
188 # make sure no errors are returned
189 self.assertEqual(None, ret)
190
191 # Make sure the contents value matches the expected value.
192 # returner.call_args[0] will be an args tuple containing all the args
193 # passed to the mocked returner for file.manage_file. Any changes to
194 # the arguments for file.manage_file may make this assertion fail.
195 # If the test is failing, check the position of the "contents" param
196 # in the manage_file() function in salt/modules/file.py, the fix is
197 # likely as simple as updating the 2nd index below.
198 self.assertEqual(expected, returner.call_args[0][-5])
199
200 def test_symlink(self):
201 """
202 Test to create a symlink.
203 """
204 name = os.sep + os.path.join("tmp", "testfile.txt")
205 target = salt.utils.files.mkstemp()
206 test_dir = os.sep + "tmp"
207 user = "salt"
208
209 if salt.utils.platform.is_windows():
210 group = "salt"
211 else:
212 group = "saltstack"
213
214 def return_val(kwargs):
215 val = {
216 "name": name,
217 "result": False,
218 "comment": "",
219 "changes": {},
220 }
221 val.update(kwargs)
222 return val
223
224 mock_t = MagicMock(return_value=True)
225 mock_f = MagicMock(return_value=False)
226 mock_empty = MagicMock(return_value="")
227 mock_uid = MagicMock(return_value="U1001")
228 mock_gid = MagicMock(return_value="g1001")
229 mock_target = MagicMock(return_value=target)
230 mock_user = MagicMock(return_value=user)
231 mock_grp = MagicMock(return_value=group)
232 mock_os_error = MagicMock(side_effect=OSError)
233
234 with patch.dict(filestate.__salt__, {"config.manage_mode": mock_t}):
235 comt = "Must provide name to file.symlink"
236 ret = return_val({"comment": comt, "name": ""})
237 self.assertDictEqual(filestate.symlink("", target), ret)
238
239 with patch.dict(
240 filestate.__salt__,
241 {
242 "config.manage_mode": mock_t,
243 "file.user_to_uid": mock_empty,
244 "file.group_to_gid": mock_empty,
245 "user.info": mock_empty,
246 "user.current": mock_user,
247 },
248 ):
249 if salt.utils.platform.is_windows():
250 comt = "User {} does not exist".format(user)
251 ret = return_val({"comment": comt, "name": name})
252 else:
253 comt = "User {} does not exist. Group {} does not exist.".format(
254 user, group
255 )
256 ret = return_val({"comment": comt, "name": name})
257 self.assertDictEqual(
258 filestate.symlink(name, target, user=user, group=group), ret
259 )
260
261 with patch.dict(
262 filestate.__salt__,
263 {
264 "config.manage_mode": mock_t,
265 "file.user_to_uid": mock_uid,
266 "file.group_to_gid": mock_gid,
267 "file.is_link": mock_f,
268 "user.info": mock_empty,
269 "user.current": mock_user,
270 },
271 ), patch.dict(filestate.__opts__, {"test": True}), patch.object(
272 os.path, "exists", mock_f
273 ):
274 if salt.utils.platform.is_windows():
275 comt = "User {} does not exist".format(user)
276 ret = return_val(
277 {"comment": comt, "result": False, "name": name, "changes": {}}
278 )
279 else:
280 comt = "Symlink {} to {} is set for creation".format(name, target)
281 ret = return_val(
282 {"comment": comt, "result": None, "changes": {"new": name}}
283 )
284 self.assertDictEqual(
285 filestate.symlink(name, target, user=user, group=group), ret
286 )
287
288 with patch.dict(
289 filestate.__salt__,
290 {
291 "config.manage_mode": mock_t,
292 "file.user_to_uid": mock_uid,
293 "file.group_to_gid": mock_gid,
294 "file.is_link": mock_f,
295 "user.info": mock_empty,
296 "user.current": mock_user,
297 },
298 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
299 os.path, "isdir", mock_f
300 ), patch.object(
301 os.path, "exists", mock_f
302 ):
303 if salt.utils.platform.is_windows():
304 comt = "User {} does not exist".format(user)
305 ret = return_val(
306 {"comment": comt, "result": False, "name": name, "changes": {}}
307 )
308 else:
309 comt = "Directory {} for symlink is not present".format(test_dir)
310 ret = return_val({"comment": comt, "result": False, "changes": {}})
311 self.assertDictEqual(
312 filestate.symlink(name, target, user=user, group=group), ret
313 )
314
315 with patch.dict(
316 filestate.__salt__,
317 {
318 "config.manage_mode": mock_t,
319 "file.user_to_uid": mock_uid,
320 "file.group_to_gid": mock_gid,
321 "file.is_link": mock_t,
322 "file.readlink": mock_target,
323 "user.info": mock_empty,
324 "user.current": mock_user,
325 },
326 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
327 os.path, "isdir", mock_t
328 ), patch.object(
329 salt.states.file, "_check_symlink_ownership", mock_t
330 ), patch(
331 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
332 ):
333 if salt.utils.platform.is_windows():
334 comt = "Symlink {} is present and owned by {}".format(name, user)
335 else:
336 comt = "Symlink {} is present and owned by {}:{}".format(
337 name, user, group
338 )
339 ret = return_val({"comment": comt, "result": True, "changes": {}})
340 self.assertDictEqual(
341 filestate.symlink(name, target, user=user, group=group), ret
342 )
343
344 with patch.dict(
345 filestate.__salt__,
346 {
347 "config.manage_mode": mock_t,
348 "file.user_to_uid": mock_uid,
349 "file.group_to_gid": mock_gid,
350 "file.is_link": mock_f,
351 "file.readlink": mock_target,
352 "user.info": mock_empty,
353 "user.current": mock_user,
354 },
355 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
356 os.path, "isdir", mock_t
357 ), patch.object(
358 os.path, "exists", mock_t
359 ), patch.object(
360 os.path, "lexists", mock_t
361 ), patch(
362 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
363 ):
364 comt = (
365 "Symlink & backup dest exists and Force not set. {} -> "
366 "{} - backup: {}".format(name, target, os.path.join(test_dir, "SALT"))
367 )
368 ret.update({"comment": comt, "result": False, "changes": {}})
369 self.assertDictEqual(
370 filestate.symlink(
371 name, target, user=user, group=group, backupname="SALT"
372 ),
373 ret,
374 )
375
376 with patch.dict(
377 filestate.__salt__,
378 {
379 "config.manage_mode": mock_t,
380 "file.user_to_uid": mock_uid,
381 "file.group_to_gid": mock_gid,
382 "file.is_link": mock_f,
383 "file.readlink": mock_target,
384 "user.info": mock_empty,
385 "user.current": mock_user,
386 },
387 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
388 os.path, "exists", mock_t
389 ), patch.object(
390 os.path, "isfile", mock_t
391 ), patch.object(
392 os.path, "isdir", mock_t
393 ), patch(
394 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
395 ):
396 comt = "Backupname must be an absolute path or a file name: {}".format(
397 "tmp/SALT"
398 )
399 ret.update({"comment": comt, "result": False, "changes": {}})
400 self.assertDictEqual(
401 filestate.symlink(
402 name, target, user=user, group=group, backupname="tmp/SALT"
403 ),
404 ret,
405 )
406
407 with patch.dict(
408 filestate.__salt__,
409 {
410 "config.manage_mode": mock_t,
411 "file.user_to_uid": mock_uid,
412 "file.group_to_gid": mock_gid,
413 "file.is_link": mock_f,
414 "file.readlink": mock_target,
415 "user.info": mock_empty,
416 "user.current": mock_user,
417 },
418 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
419 os.path, "isdir", mock_t
420 ), patch.object(
421 os.path, "exists", mock_t
422 ), patch.object(
423 os.path, "isfile", mock_t
424 ), patch(
425 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
426 ):
427 comt = "File exists where the symlink {} should be".format(name)
428 ret = return_val({"comment": comt, "changes": {}, "result": False})
429 self.assertDictEqual(
430 filestate.symlink(name, target, user=user, group=group), ret
431 )
432
433 with patch.dict(
434 filestate.__salt__,
435 {
436 "config.manage_mode": mock_t,
437 "file.user_to_uid": mock_uid,
438 "file.group_to_gid": mock_gid,
439 "file.is_link": mock_f,
440 "file.readlink": mock_target,
441 "file.symlink": mock_t,
442 "user.info": mock_t,
443 "file.lchown": mock_f,
444 },
445 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
446 os.path, "isdir", MagicMock(side_effect=[True, False])
447 ), patch.object(
448 os.path, "isdir", mock_t
449 ), patch.object(
450 os.path, "exists", mock_t
451 ), patch(
452 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
453 ):
454 comt = "Directory exists where the symlink {} should be".format(name)
455 ret = return_val({"comment": comt, "result": False, "changes": {}})
456 self.assertDictEqual(
457 filestate.symlink(name, target, user=user, group=group), ret
458 )
459
460 with patch.dict(
461 filestate.__salt__,
462 {
463 "config.manage_mode": mock_t,
464 "file.user_to_uid": mock_uid,
465 "file.group_to_gid": mock_gid,
466 "file.is_link": mock_f,
467 "file.readlink": mock_target,
468 "file.symlink": mock_os_error,
469 "user.info": mock_t,
470 "file.lchown": mock_f,
471 },
472 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
473 os.path, "isdir", MagicMock(side_effect=[True, False])
474 ), patch.object(
475 os.path, "isfile", mock_f
476 ), patch(
477 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
478 ):
479 comt = "Unable to create new symlink {} -> {}: ".format(name, target)
480 ret = return_val({"comment": comt, "result": False, "changes": {}})
481 self.assertDictEqual(
482 filestate.symlink(name, target, user=user, group=group), ret
483 )
484
485 with patch.dict(
486 filestate.__salt__,
487 {
488 "config.manage_mode": mock_t,
489 "file.user_to_uid": mock_uid,
490 "file.group_to_gid": mock_gid,
491 "file.is_link": mock_f,
492 "file.readlink": mock_target,
493 "file.symlink": mock_t,
494 "user.info": mock_t,
495 "file.lchown": mock_f,
496 "file.get_user": mock_user,
497 "file.get_group": mock_grp,
498 },
499 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
500 os.path, "isdir", MagicMock(side_effect=[True, False])
501 ), patch.object(
502 os.path, "isfile", mock_f
503 ), patch(
504 "salt.states.file._check_symlink_ownership", return_value=True
505 ), patch(
506 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
507 ):
508 comt = "Created new symlink {} -> {}".format(name, target)
509 ret = return_val(
510 {"comment": comt, "result": True, "changes": {"new": name}}
511 )
512 self.assertDictEqual(
513 filestate.symlink(name, target, user=user, group=group), ret
514 )
515
516 with patch.dict(
517 filestate.__salt__,
518 {
519 "config.manage_mode": mock_t,
520 "file.user_to_uid": mock_uid,
521 "file.group_to_gid": mock_gid,
522 "file.is_link": mock_f,
523 "file.readlink": mock_target,
524 "file.symlink": mock_t,
525 "user.info": mock_t,
526 "file.lchown": mock_f,
527 "file.get_user": mock_empty,
528 "file.get_group": mock_empty,
529 },
530 ), patch.dict(filestate.__opts__, {"test": False}), patch.object(
531 os.path, "isdir", MagicMock(side_effect=[True, False])
532 ), patch.object(
533 os.path, "isfile", mock_f
534 ), patch(
535 "salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
536 ), patch(
537 "salt.states.file._set_symlink_ownership", return_value=False
538 ), patch(
539 "salt.states.file._check_symlink_ownership", return_value=False
540 ):
541 comt = (
542 "Created new symlink {} -> {}, but was unable to set "
543 "ownership to {}:{}".format(name, target, user, group)
544 )
545 ret = return_val(
546 {"comment": comt, "result": False, "changes": {"new": name}}
547 )
548 self.assertDictEqual(
549 filestate.symlink(name, target, user=user, group=group), ret
550 )
551
552 @skipIf(salt.utils.platform.is_windows(), "Do not run on Windows")
553 def test_hardlink(self):
554 """
555 Test to create a hardlink.
556 """
557
558 name = os.path.join(os.sep, "tmp", "testfile.txt")
559 target = salt.utils.files.mkstemp()
560 test_dir = os.path.join(os.sep, "tmp")
561 user, group = "salt", "saltstack"
562
563 def return_val(**kwargs):
564 res = {
565 "name": name,
566 "result": False,
567 "comment": "",
568 "changes": {},
569 }
570 res.update(kwargs)
571 return res
572
573 mock_t = MagicMock(return_value=True)
574 mock_f = MagicMock(return_value=False)
575 mock_empty = MagicMock(return_value="")
576 mock_uid = MagicMock(return_value="U1001")
577 mock_gid = MagicMock(return_value="g1001")
578 mock_nothing = MagicMock(return_value={})
579 mock_stats = MagicMock(return_value={"inode": 1})
580 mock_execerror = MagicMock(side_effect=CommandExecutionError)
581
582 patches = {}
583 patches["file.user_to_uid"] = mock_empty
584 patches["file.group_to_gid"] = mock_empty
585 patches["user.info"] = mock_empty
586 patches["file.is_hardlink"] = mock_t
587 patches["file.stats"] = mock_empty
588
589 # Argument validation
590 with patch.dict(filestate.__salt__, patches):
591 expected = "Must provide name to file.hardlink"
592 ret = return_val(comment=expected, name="")
593 self.assertDictEqual(filestate.hardlink("", target), ret)
594
595 # User validation for dir_mode
596 with patch.dict(filestate.__salt__, patches), patch.dict(
597 filestate.__salt__, {"file.user_to_uid": mock_empty}
598 ), patch.dict(
599 filestate.__salt__, {"file.group_to_gid": mock_gid}
600 ), patch.object(
601 os.path, "isabs", mock_t
602 ):
603 expected = "User {} does not exist".format(user)
604 ret = return_val(comment=expected, name=name)
605 self.assertDictEqual(
606 filestate.hardlink(name, target, user=user, group=group), ret
607 )
608
609 # Group validation for dir_mode
610 with patch.dict(filestate.__salt__, patches), patch.dict(
611 filestate.__salt__, {"file.user_to_uid": mock_uid}
612 ), patch.dict(
613 filestate.__salt__, {"file.group_to_gid": mock_empty}
614 ), patch.object(
615 os.path, "isabs", mock_t
616 ):
617 expected = "Group {} does not exist".format(group)
618 ret = return_val(comment=expected, name=name)
619 self.assertDictEqual(
620 filestate.hardlink(name, target, user=user, group=group), ret
621 )
622
623 # Absolute path for name
624 nonabs = "./non-existent-path/to/non-existent-file"
625 with patch.dict(filestate.__salt__, patches), patch.dict(
626 filestate.__salt__, {"file.user_to_uid": mock_uid}
627 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
628 expected = "Specified file {} is not an absolute path".format(nonabs)
629 ret = return_val(comment=expected, name=nonabs)
630 self.assertDictEqual(
631 filestate.hardlink(nonabs, target, user=user, group=group), ret
632 )
633
634 # Absolute path for target
635 with patch.dict(filestate.__salt__, patches), patch.dict(
636 filestate.__salt__, {"file.user_to_uid": mock_uid}
637 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
638 expected = "Specified target {} is not an absolute path".format(nonabs)
639 ret = return_val(comment=expected, name=name)
640 self.assertDictEqual(
641 filestate.hardlink(name, nonabs, user=user, group=group), ret
642 )
643 # Test option -- nonexistent target
644 with patch.dict(filestate.__salt__, patches), patch.dict(
645 filestate.__salt__, {"file.user_to_uid": mock_uid}
646 ), patch.dict(
647 filestate.__salt__, {"file.group_to_gid": mock_gid}
648 ), patch.object(
649 os.path, "exists", mock_f
650 ), patch.dict(
651 filestate.__opts__, {"test": True}
652 ):
653 expected = "Target {} for hard link does not exist".format(target)
654 ret = return_val(comment=expected, name=name)
655 self.assertDictEqual(
656 filestate.hardlink(name, target, user=user, group=group), ret
657 )
658
659 # Test option -- target is a directory
660 with patch.dict(filestate.__salt__, patches), patch.dict(
661 filestate.__salt__, {"file.user_to_uid": mock_uid}
662 ), patch.dict(
663 filestate.__salt__, {"file.group_to_gid": mock_gid}
664 ), patch.object(
665 os.path, "exists", mock_t
666 ), patch.dict(
667 filestate.__opts__, {"test": True}
668 ):
669 expected = "Unable to hard link from directory {}".format(test_dir)
670 ret = return_val(comment=expected, name=name)
671 self.assertDictEqual(
672 filestate.hardlink(name, test_dir, user=user, group=group), ret
673 )
674
675 # Test option -- name is a directory
676 with patch.dict(filestate.__salt__, patches), patch.dict(
677 filestate.__salt__, {"file.user_to_uid": mock_uid}
678 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
679 filestate.__opts__, {"test": True}
680 ):
681 expected = "Unable to hard link to directory {}".format(test_dir)
682 ret = return_val(comment=expected, name=test_dir)
683 self.assertDictEqual(
684 filestate.hardlink(test_dir, target, user=user, group=group), ret
685 )
686
687 # Test option -- name does not exist
688 with patch.dict(filestate.__salt__, patches), patch.dict(
689 filestate.__salt__, {"file.user_to_uid": mock_uid}
690 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
691 filestate.__opts__, {"test": True}
692 ):
693 expected = "Hard link {} to {} is set for creation".format(name, target)
694 changes = dict(new=name)
695 ret = return_val(result=None, comment=expected, name=name, changes=changes)
696 self.assertDictEqual(
697 filestate.hardlink(name, target, user=user, group=group), ret
698 )
699
700 # Test option -- hardlink matches
701 with patch.dict(filestate.__salt__, patches), patch.dict(
702 filestate.__salt__, {"file.user_to_uid": mock_uid}
703 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
704 filestate.__salt__, {"file.is_hardlink": mock_t}
705 ), patch.dict(
706 filestate.__salt__, {"file.stats": mock_stats}
707 ), patch.object(
708 os.path, "exists", mock_t
709 ), patch.dict(
710 filestate.__opts__, {"test": True}
711 ):
712 expected = "The hard link {} is presently targetting {}".format(
713 name, target
714 )
715 ret = return_val(result=True, comment=expected, name=name)
716 self.assertDictEqual(
717 filestate.hardlink(name, target, user=user, group=group), ret
718 )
719
720 # Test option -- hardlink does not match
721 with patch.dict(filestate.__salt__, patches), patch.dict(
722 filestate.__salt__, {"file.user_to_uid": mock_uid}
723 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
724 filestate.__salt__, {"file.is_hardlink": mock_t}
725 ), patch.dict(
726 filestate.__salt__, {"file.stats": mock_nothing}
727 ), patch.object(
728 os.path, "exists", mock_t
729 ), patch.dict(
730 filestate.__opts__, {"test": True}
731 ):
732 expected = "Link {} target is set to be changed to {}".format(name, target)
733 changes = dict(change=name)
734 ret = return_val(result=None, comment=expected, name=name, changes=changes)
735 self.assertDictEqual(
736 filestate.hardlink(name, target, user=user, group=group), ret
737 )
738
739 # Test option -- force removal
740 with patch.dict(filestate.__salt__, patches), patch.dict(
741 filestate.__salt__, {"file.user_to_uid": mock_uid}
742 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
743 filestate.__salt__, {"file.is_hardlink": mock_f}
744 ), patch.object(
745 os.path, "exists", mock_t
746 ), patch.dict(
747 filestate.__opts__, {"test": True}
748 ):
749 expected = (
750 "The file or directory {} is set for removal to "
751 "make way for a new hard link targeting {}".format(name, target)
752 )
753 ret = return_val(result=None, comment=expected, name=name)
754 self.assertDictEqual(
755 filestate.hardlink(name, target, force=True, user=user, group=group),
756 ret,
757 )
758
759 # Test option -- without force removal
760 with patch.dict(filestate.__salt__, patches), patch.dict(
761 filestate.__salt__, {"file.user_to_uid": mock_uid}
762 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
763 filestate.__salt__, {"file.is_hardlink": mock_f}
764 ), patch.object(
765 os.path, "exists", mock_t
766 ), patch.dict(
767 filestate.__opts__, {"test": True}
768 ):
769 expected = (
770 "File or directory exists where the hard link {} "
771 "should be. Did you mean to use force?".format(name)
772 )
773 ret = return_val(result=False, comment=expected, name=name)
774 self.assertDictEqual(
775 filestate.hardlink(name, target, force=False, user=user, group=group),
776 ret,
777 )
778
779 # Target is a directory
780 with patch.dict(filestate.__salt__, patches), patch.dict(
781 filestate.__salt__, {"file.user_to_uid": mock_uid}
782 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
783 expected = "Unable to hard link from directory {}".format(test_dir)
784 ret = return_val(comment=expected, name=name)
785 self.assertDictEqual(
786 filestate.hardlink(name, test_dir, user=user, group=group), ret
787 )
788
789 # Name is a directory
790 with patch.dict(filestate.__salt__, patches), patch.dict(
791 filestate.__salt__, {"file.user_to_uid": mock_uid}
792 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
793 expected = "Unable to hard link to directory {}".format(test_dir)
794 ret = return_val(comment=expected, name=test_dir)
795 self.assertDictEqual(
796 filestate.hardlink(test_dir, target, user=user, group=group), ret
797 )
798
799 # Try overwrite file with link
800 with patch.dict(filestate.__salt__, patches), patch.dict(
801 filestate.__salt__, {"file.user_to_uid": mock_uid}
802 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
803 filestate.__salt__, {"file.is_hardlink": mock_f}
804 ), patch.object(
805 os.path, "isfile", mock_t
806 ):
807
808 expected = "File exists where the hard link {} should be".format(name)
809 ret = return_val(comment=expected, name=name)
810 self.assertDictEqual(
811 filestate.hardlink(name, target, user=user, group=group), ret
812 )
813
814 # Try overwrite link with same
815 with patch.dict(filestate.__salt__, patches), patch.dict(
816 filestate.__salt__, {"file.user_to_uid": mock_uid}
817 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
818 filestate.__salt__, {"file.is_hardlink": mock_t}
819 ), patch.dict(
820 filestate.__salt__, {"file.stats": mock_stats}
821 ), patch.object(
822 os.path, "isfile", mock_f
823 ):
824
825 expected = "Target of hard link {} is already pointing " "to {}".format(
826 name, target
827 )
828 ret = return_val(result=True, comment=expected, name=name)
829 self.assertDictEqual(
830 filestate.hardlink(name, target, user=user, group=group), ret
831 )
832
833 # Really overwrite link with same
834 with patch.dict(filestate.__salt__, patches), patch.dict(
835 filestate.__salt__, {"file.user_to_uid": mock_uid}
836 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
837 filestate.__salt__, {"file.is_hardlink": mock_t}
838 ), patch.dict(
839 filestate.__salt__, {"file.link": mock_t}
840 ), patch.dict(
841 filestate.__salt__, {"file.stats": mock_nothing}
842 ), patch.object(
843 os, "remove", mock_t
844 ), patch.object(
845 os.path, "isfile", mock_f
846 ):
847
848 expected = "Set target of hard link {} -> {}".format(name, target)
849 changes = dict(new=name)
850 ret = return_val(result=True, comment=expected, name=name, changes=changes)
851 self.assertDictEqual(
852 filestate.hardlink(name, target, user=user, group=group), ret
853 )
854
855 # Fail at overwriting link with same
856 with patch.dict(filestate.__salt__, patches), patch.dict(
857 filestate.__salt__, {"file.user_to_uid": mock_uid}
858 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
859 filestate.__salt__, {"file.is_hardlink": mock_t}
860 ), patch.dict(
861 filestate.__salt__, {"file.link": mock_execerror}
862 ), patch.dict(
863 filestate.__salt__, {"file.stats": mock_nothing}
864 ), patch.object(
865 os, "remove", mock_t
866 ), patch.object(
867 os.path, "isfile", mock_f
868 ):
869
870 expected = "Unable to set target of hard link {} -> " "{}: {}".format(
871 name, target, ""
872 )
873 ret = return_val(result=False, comment=expected, name=name)
874 self.assertDictEqual(
875 filestate.hardlink(name, target, user=user, group=group), ret
876 )
877
878 # Make new link
879 with patch.dict(filestate.__salt__, patches), patch.dict(
880 filestate.__salt__, {"file.user_to_uid": mock_uid}
881 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
882 filestate.__salt__, {"file.is_hardlink": mock_f}
883 ), patch.dict(
884 filestate.__salt__, {"file.link": mock_f}
885 ), patch.dict(
886 filestate.__salt__, {"file.stats": mock_nothing}
887 ), patch.object(
888 os, "remove", mock_t
889 ), patch.object(
890 os.path, "isfile", mock_f
891 ):
892
893 expected = "Created new hard link {} -> {}".format(name, target)
894 changes = dict(new=name)
895 ret = return_val(result=True, comment=expected, name=name, changes=changes)
896 self.assertDictEqual(
897 filestate.hardlink(name, target, user=user, group=group), ret
898 )
899
900 # Fail while making new link
901 with patch.dict(filestate.__salt__, patches), patch.dict(
902 filestate.__salt__, {"file.user_to_uid": mock_uid}
903 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
904 filestate.__salt__, {"file.is_hardlink": mock_f}
905 ), patch.dict(
906 filestate.__salt__, {"file.link": mock_execerror}
907 ), patch.dict(
908 filestate.__salt__, {"file.stats": mock_nothing}
909 ), patch.object(
910 os, "remove", mock_t
911 ), patch.object(
912 os.path, "isfile", mock_f
913 ):
914
915 expected = "Unable to create new hard link {} -> " "{}: {}".format(
916 name, target, ""
917 )
918 ret = return_val(result=False, comment=expected, name=name)
919 self.assertDictEqual(
920 filestate.hardlink(name, target, user=user, group=group), ret
921 )
922
923 # Force making new link over file
924 with patch.dict(filestate.__salt__, patches), patch.dict(
925 filestate.__salt__, {"file.user_to_uid": mock_uid}
926 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
927 filestate.__salt__, {"file.is_hardlink": mock_f}
928 ), patch.dict(
929 filestate.__salt__, {"file.link": mock_t}
930 ), patch.dict(
931 filestate.__salt__, {"file.stats": mock_nothing}
932 ), patch.object(
933 os, "remove", mock_t
934 ), patch.object(
935 os.path, "isfile", mock_t
936 ):
937
938 expected = "Created new hard link {} -> {}".format(name, target)
939 changes = dict(new=name)
940 changes["forced"] = "File for hard link was forcibly replaced"
941 ret = return_val(result=True, comment=expected, name=name, changes=changes)
942 self.assertDictEqual(
943 filestate.hardlink(name, target, user=user, force=True, group=group),
944 ret,
945 )
946
947 # Force making new link over file but error out
948 with patch.dict(filestate.__salt__, patches), patch.dict(
949 filestate.__salt__, {"file.user_to_uid": mock_uid}
950 ), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
951 filestate.__salt__, {"file.is_hardlink": mock_f}
952 ), patch.dict(
953 filestate.__salt__, {"file.link": mock_execerror}
954 ), patch.dict(
955 filestate.__salt__, {"file.stats": mock_nothing}
956 ), patch.object(
957 os, "remove", mock_t
958 ), patch.object(
959 os.path, "isfile", mock_t
960 ):
961
962 expected = "Unable to create new hard link {} -> " "{}: {}".format(
963 name, target, ""
964 )
965 changes = dict(forced="File for hard link was forcibly replaced")
966 ret = return_val(result=False, comment=expected, name=name, changes=changes)
967 self.assertDictEqual(
968 filestate.hardlink(name, target, user=user, force=True, group=group),
969 ret,
970 )
971
972 # 'absent' function tests: 1
973 def test_absent(self):
974 """
975 Test to make sure that the named file or directory is absent.
976 """
977 name = "/fake/file.conf"
978
979 ret = {"name": name, "result": False, "comment": "", "changes": {}}
980
981 mock_t = MagicMock(return_value=True)
982 mock_f = MagicMock(return_value=False)
983 mock_file = MagicMock(side_effect=[True, CommandExecutionError])
984 mock_tree = MagicMock(side_effect=[True, OSError])
985
986 comt = "Must provide name to file.absent"
987 ret.update({"comment": comt, "name": ""})
988
989 with patch.object(os.path, "islink", MagicMock(return_value=False)):
990 self.assertDictEqual(filestate.absent(""), ret)
991
992 with patch.object(os.path, "isabs", mock_f):
993 comt = "Specified file {} is not an absolute path".format(name)
994 ret.update({"comment": comt, "name": name})
995 self.assertDictEqual(filestate.absent(name), ret)
996
997 with patch.object(os.path, "isabs", mock_t):
998 comt = 'Refusing to make "/" absent'
999 ret.update({"comment": comt, "name": "/"})
1000 self.assertDictEqual(filestate.absent("/"), ret)
1001
1002 with patch.object(os.path, "isfile", mock_t):
1003 with patch.dict(filestate.__opts__, {"test": True}):
1004 comt = "File {} is set for removal".format(name)
1005 ret.update(
1006 {
1007 "comment": comt,
1008 "name": name,
1009 "result": None,
1010 "changes": {"removed": "/fake/file.conf"},
1011 }
1012 )
1013 self.assertDictEqual(filestate.absent(name), ret)
1014
1015 with patch.dict(filestate.__opts__, {"test": False}):
1016 with patch.dict(filestate.__salt__, {"file.remove": mock_file}):
1017 comt = "Removed file {}".format(name)
1018 ret.update(
1019 {
1020 "comment": comt,
1021 "result": True,
1022 "changes": {"removed": name},
1023 }
1024 )
1025 self.assertDictEqual(filestate.absent(name), ret)
1026
1027 comt = "Removed file {}".format(name)
1028 ret.update({"comment": "", "result": False, "changes": {}})
1029 self.assertDictEqual(filestate.absent(name), ret)
1030
1031 with patch.object(os.path, "isfile", mock_f):
1032 with patch.object(os.path, "isdir", mock_t):
1033 with patch.dict(filestate.__opts__, {"test": True}):
1034 comt = "Directory {} is set for removal".format(name)
1035 ret.update(
1036 {
1037 "comment": comt,
1038 "changes": {"removed": name},
1039 "result": None,
1040 }
1041 )
1042 self.assertDictEqual(filestate.absent(name), ret)
1043
1044 with patch.dict(filestate.__opts__, {"test": False}):
1045 with patch.dict(filestate.__salt__, {"file.remove": mock_tree}):
1046 comt = "Removed directory {}".format(name)
1047 ret.update(
1048 {
1049 "comment": comt,
1050 "result": True,
1051 "changes": {"removed": name},
1052 }
1053 )
1054 self.assertDictEqual(filestate.absent(name), ret)
1055
1056 comt = "Failed to remove directory {}".format(name)
1057 ret.update(
1058 {"comment": comt, "result": False, "changes": {}}
1059 )
1060 self.assertDictEqual(filestate.absent(name), ret)
1061
1062 with patch.object(os.path, "isdir", mock_f):
1063 with patch.dict(filestate.__opts__, {"test": True}):
1064 comt = "File {} is not present".format(name)
1065 ret.update({"comment": comt, "result": True})
1066 self.assertDictEqual(filestate.absent(name), ret)
1067
1068 # 'exists' function tests: 1
1069
1070 def test_exists(self):
1071 """
1072 Test to verify that the named file or directory is present or exists.
1073 """
1074 name = "/etc/grub.conf"
1075
1076 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1077
1078 mock_t = MagicMock(return_value=True)
1079 mock_f = MagicMock(return_value=False)
1080
1081 comt = "Must provide name to file.exists"
1082 ret.update({"comment": comt, "name": ""})
1083 self.assertDictEqual(filestate.exists(""), ret)
1084
1085 with patch.object(os.path, "exists", mock_f):
1086 comt = "Specified path {} does not exist".format(name)
1087 ret.update({"comment": comt, "name": name})
1088 self.assertDictEqual(filestate.exists(name), ret)
1089
1090 with patch.object(os.path, "exists", mock_t):
1091 comt = "Path {} exists".format(name)
1092 ret.update({"comment": comt, "result": True})
1093 self.assertDictEqual(filestate.exists(name), ret)
1094
1095 # 'missing' function tests: 1
1096
1097 def test_missing(self):
1098 """
1099 Test to verify that the named file or directory is missing.
1100 """
1101 name = "/etc/grub.conf"
1102
1103 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1104
1105 mock_t = MagicMock(return_value=True)
1106 mock_f = MagicMock(return_value=False)
1107
1108 comt = "Must provide name to file.missing"
1109 ret.update({"comment": comt, "name": "", "changes": {}})
1110 self.assertDictEqual(filestate.missing(""), ret)
1111
1112 with patch.object(os.path, "exists", mock_t):
1113 comt = "Specified path {} exists".format(name)
1114 ret.update({"comment": comt, "name": name})
1115 self.assertDictEqual(filestate.missing(name), ret)
1116
1117 with patch.object(os.path, "exists", mock_f):
1118 comt = "Path {} is missing".format(name)
1119 ret.update({"comment": comt, "result": True})
1120 self.assertDictEqual(filestate.missing(name), ret)
1121
1122 # 'managed' function tests: 1
1123
1124 def test_file_managed_should_fall_back_to_binary(self):
1125 expected_contents = b"\x8b"
1126 filename = "/tmp/blarg"
1127 mock_manage = MagicMock(return_value={"fnord": "fnords"})
1128 with patch(
1129 "salt.states.file._load_accumulators", MagicMock(return_value=([], []))
1130 ):
1131 with patch.dict(
1132 filestate.__salt__,
1133 {
1134 "file.get_managed": MagicMock(return_value=["", "", ""]),
1135 "file.source_list": MagicMock(return_value=["", ""]),
1136 "file.manage_file": mock_manage,
1137 "pillar.get": MagicMock(return_value=expected_contents),
1138 },
1139 ):
1140 ret = filestate.managed(
1141 filename, contents_pillar="fnord", encoding="utf-8"
1142 )
1143 actual_contents = mock_manage.call_args[0][14]
1144 self.assertEqual(actual_contents, expected_contents)
1145
1146 def test_managed(self):
1147 """
1148 Test to manage a given file, this function allows for a file to be
1149 downloaded from the salt master and potentially run through a templating
1150 system.
1151 """
1152 with patch(
1153 "salt.states.file._load_accumulators", MagicMock(return_value=([], []))
1154 ):
1155 name = "/etc/grub.conf"
1156 user = "salt"
1157 group = "saltstack"
1158
1159 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1160
1161 mock_t = MagicMock(return_value=True)
1162 mock_f = MagicMock(return_value=False)
1163 mock_cmd_fail = MagicMock(return_value={"retcode": 1})
1164 mock_uid = MagicMock(
1165 side_effect=[
1166 "",
1167 "U12",
1168 "U12",
1169 "U12",
1170 "U12",
1171 "U12",
1172 "U12",
1173 "U12",
1174 "U12",
1175 "U12",
1176 "U12",
1177 "U12",
1178 "U12",
1179 "U12",
1180 "U12",
1181 "U12",
1182 ]
1183 )
1184 mock_gid = MagicMock(
1185 side_effect=[
1186 "",
1187 "G12",
1188 "G12",
1189 "G12",
1190 "G12",
1191 "G12",
1192 "G12",
1193 "G12",
1194 "G12",
1195 "G12",
1196 "G12",
1197 "G12",
1198 "G12",
1199 "G12",
1200 "G12",
1201 "G12",
1202 ]
1203 )
1204 mock_if = MagicMock(
1205 side_effect=[True, False, False, False, False, False, False, False]
1206 )
1207 if salt.utils.platform.is_windows():
1208 mock_ret = MagicMock(return_value=ret)
1209 else:
1210 mock_ret = MagicMock(return_value=(ret, None))
1211 mock_dict = MagicMock(return_value={})
1212 mock_cp = MagicMock(side_effect=[Exception, True])
1213 mock_ex = MagicMock(
1214 side_effect=[Exception, {"changes": {name: name}}, True, Exception]
1215 )
1216 mock_mng = MagicMock(
1217 side_effect=[
1218 Exception,
1219 ("", "", ""),
1220 ("", "", ""),
1221 ("", "", True),
1222 ("", "", True),
1223 ("", "", ""),
1224 ("", "", ""),
1225 ]
1226 )
1227 mock_file = MagicMock(
1228 side_effect=[
1229 CommandExecutionError,
1230 ("", ""),
1231 ("", ""),
1232 ("", ""),
1233 ("", ""),
1234 ("", ""),
1235 ("", ""),
1236 ("", ""),
1237 ("", ""),
1238 ]
1239 )
1240 with patch.dict(
1241 filestate.__salt__,
1242 {
1243 "config.manage_mode": mock_t,
1244 "file.user_to_uid": mock_uid,
1245 "file.group_to_gid": mock_gid,
1246 "file.file_exists": mock_if,
1247 "file.check_perms": mock_ret,
1248 "file.check_managed_changes": mock_dict,
1249 "file.get_managed": mock_mng,
1250 "file.source_list": mock_file,
1251 "file.copy": mock_cp,
1252 "file.manage_file": mock_ex,
1253 "cmd.run_all": mock_cmd_fail,
1254 },
1255 ):
1256 comt = "Destination file name is required"
1257 ret.update({"comment": comt, "name": "", "changes": {}})
1258 self.assertDictEqual(filestate.managed(""), ret)
1259
1260 with patch.object(os.path, "isfile", mock_f):
1261 comt = (
1262 "File {} is not present and is not set for "
1263 "creation".format(name)
1264 )
1265 ret.update({"comment": comt, "name": name, "result": True})
1266 self.assertDictEqual(filestate.managed(name, create=False), ret)
1267
1268 # Group argument is ignored on Windows systems. Group is set to
1269 # user
1270 if salt.utils.platform.is_windows():
1271 comt = "User salt is not available Group salt" " is not available"
1272 else:
1273 comt = (
1274 "User salt is not available Group saltstack" " is not available"
1275 )
1276 ret.update({"comment": comt, "result": False})
1277 self.assertDictEqual(
1278 filestate.managed(name, user=user, group=group), ret
1279 )
1280
1281 with patch.object(os.path, "isabs", mock_f):
1282 comt = "Specified file {} is not an absolute path".format(name)
1283 ret.update({"comment": comt, "result": False})
1284 self.assertDictEqual(
1285 filestate.managed(name, user=user, group=group), ret
1286 )
1287
1288 with patch.object(os.path, "isabs", mock_t):
1289 with patch.object(os.path, "isdir", mock_t):
1290 comt = "Specified target {} is a directory".format(name)
1291 ret.update({"comment": comt})
1292 self.assertDictEqual(
1293 filestate.managed(name, user=user, group=group), ret
1294 )
1295
1296 with patch.object(os.path, "isdir", mock_f):
1297 comt = "Context must be formed as a dict"
1298 ret.update({"comment": comt})
1299 self.assertDictEqual(
1300 filestate.managed(
1301 name, user=user, group=group, context=True
1302 ),
1303 ret,
1304 )
1305
1306 comt = "Defaults must be formed as a dict"
1307 ret.update({"comment": comt})
1308 self.assertDictEqual(
1309 filestate.managed(
1310 name, user=user, group=group, defaults=True
1311 ),
1312 ret,
1313 )
1314
1315 comt = (
1316 "Only one of 'contents', 'contents_pillar', "
1317 "and 'contents_grains' is permitted"
1318 )
1319 ret.update({"comment": comt})
1320 self.assertDictEqual(
1321 filestate.managed(
1322 name,
1323 user=user,
1324 group=group,
1325 contents="A",
1326 contents_grains="B",
1327 contents_pillar="C",
1328 ),
1329 ret,
1330 )
1331
1332 with patch.object(os.path, "exists", mock_t):
1333 with patch.dict(filestate.__opts__, {"test": True}):
1334 comt = "File {} not updated".format(name)
1335 ret.update({"comment": comt})
1336 self.assertDictEqual(
1337 filestate.managed(
1338 name, user=user, group=group, replace=False
1339 ),
1340 ret,
1341 )
1342
1343 comt = "The file {} is in the correct state".format(
1344 name
1345 )
1346 ret.update({"comment": comt, "result": True})
1347 self.assertDictEqual(
1348 filestate.managed(
1349 name, user=user, contents="A", group=group
1350 ),
1351 ret,
1352 )
1353
1354 with patch.object(os.path, "exists", mock_f):
1355 with patch.dict(filestate.__opts__, {"test": False}):
1356 comt = "Unable to manage file: "
1357 ret.update({"comment": comt, "result": False})
1358 self.assertDictEqual(
1359 filestate.managed(
1360 name, user=user, group=group, contents="A"
1361 ),
1362 ret,
1363 )
1364
1365 comt = "Unable to manage file: "
1366 ret.update({"comment": comt, "result": False})
1367 self.assertDictEqual(
1368 filestate.managed(
1369 name, user=user, group=group, contents="A"
1370 ),
1371 ret,
1372 )
1373
1374 with patch.object(
1375 salt.utils.files, "mkstemp", return_value=name
1376 ):
1377 comt = "Unable to copy file {0} to {0}: ".format(
1378 name
1379 )
1380 ret.update({"comment": comt, "result": False})
1381 self.assertDictEqual(
1382 filestate.managed(
1383 name, user=user, group=group, check_cmd="A"
1384 ),
1385 ret,
1386 )
1387
1388 comt = "Unable to check_cmd file: "
1389 ret.update({"comment": comt, "result": False})
1390 self.assertDictEqual(
1391 filestate.managed(
1392 name, user=user, group=group, check_cmd="A"
1393 ),
1394 ret,
1395 )
1396
1397 comt = "check_cmd execution failed"
1398 ret.update(
1399 {
1400 "comment": comt,
1401 "result": False,
1402 "skip_watch": True,
1403 }
1404 )
1405 self.assertDictEqual(
1406 filestate.managed(
1407 name, user=user, group=group, check_cmd="A"
1408 ),
1409 ret,
1410 )
1411
1412 comt = "check_cmd execution failed"
1413 ret.update({"comment": True, "changes": {}})
1414 ret.pop("skip_watch", None)
1415 self.assertDictEqual(
1416 filestate.managed(name, user=user, group=group), ret
1417 )
1418
1419 self.assertTrue(
1420 filestate.managed(name, user=user, group=group)
1421 )
1422
1423 comt = "Unable to manage file: "
1424 ret.update({"comment": comt})
1425 self.assertDictEqual(
1426 filestate.managed(name, user=user, group=group), ret
1427 )
1428
1429 if salt.utils.platform.is_windows():
1430 mock_ret = MagicMock(return_value=ret)
1431 comt = "File {} not updated".format(name)
1432 else:
1433 perms = {"luser": user, "lmode": "0644", "lgroup": group}
1434 mock_ret = MagicMock(return_value=(ret, perms))
1435 comt = (
1436 "File {} will be updated with "
1437 "permissions 0400 from its current "
1438 "state of 0644".format(name)
1439 )
1440
1441 with patch.dict(
1442 filestate.__salt__, {"file.check_perms": mock_ret}
1443 ):
1444 with patch.object(os.path, "exists", mock_t):
1445 with patch.dict(filestate.__opts__, {"test": True}):
1446 ret.update({"comment": comt})
1447 if salt.utils.platform.is_windows():
1448 self.assertDictEqual(
1449 filestate.managed(
1450 name, user=user, group=group
1451 ),
1452 ret,
1453 )
1454 else:
1455 self.assertDictEqual(
1456 filestate.managed(
1457 name, user=user, group=group, mode=400
1458 ),
1459 ret,
1460 )
1461
1462 # 'directory' function tests: 1
1463
1464 def test_directory(self):
1465 """
1466 Test to ensure that a named directory is present and has the right perms
1467 """
1468 name = "/etc/testdir"
1469 user = "salt"
1470 group = "saltstack"
1471 if salt.utils.platform.is_windows():
1472 name = name.replace("/", "\\")
1473
1474 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1475
1476 check_perms_ret = {"name": name, "result": False, "comment": "", "changes": {}}
1477
1478 comt = "Must provide name to file.directory"
1479 ret.update({"comment": comt, "name": ""})
1480 self.assertDictEqual(filestate.directory(""), ret)
1481
1482 comt = "Cannot specify both max_depth and clean"
1483 ret.update({"comment": comt, "name": name})
1484 self.assertDictEqual(filestate.directory(name, clean=True, max_depth=2), ret)
1485
1486 mock_t = MagicMock(return_value=True)
1487 mock_f = MagicMock(return_value=False)
1488 if salt.utils.platform.is_windows():
1489 mock_perms = MagicMock(return_value=check_perms_ret)
1490 else:
1491 mock_perms = MagicMock(return_value=(check_perms_ret, ""))
1492 mock_uid = MagicMock(
1493 side_effect=[
1494 "",
1495 "U12",
1496 "U12",
1497 "U12",
1498 "U12",
1499 "U12",
1500 "U12",
1501 "U12",
1502 "U12",
1503 "U12",
1504 "U12",
1505 ]
1506 )
1507 mock_gid = MagicMock(
1508 side_effect=[
1509 "",
1510 "G12",
1511 "G12",
1512 "G12",
1513 "G12",
1514 "G12",
1515 "G12",
1516 "G12",
1517 "G12",
1518 "G12",
1519 "G12",
1520 ]
1521 )
1522 mock_check = MagicMock(
1523 return_value=(
1524 None,
1525 'The directory "{}" will be changed'.format(name),
1526 {name: {"directory": "new"}},
1527 )
1528 )
1529 mock_error = CommandExecutionError
1530 with patch.dict(
1531 filestate.__salt__,
1532 {
1533 "config.manage_mode": mock_t,
1534 "file.user_to_uid": mock_uid,
1535 "file.group_to_gid": mock_gid,
1536 "file.stats": mock_f,
1537 "file.check_perms": mock_perms,
1538 "file.mkdir": mock_t,
1539 },
1540 ), patch("salt.utils.win_dacl.get_sid", mock_error), patch(
1541 "os.path.isdir", mock_t
1542 ), patch(
1543 "salt.states.file._check_directory_win", mock_check
1544 ):
1545 if salt.utils.platform.is_windows():
1546 comt = ""
1547 else:
1548 comt = "User salt is not available Group saltstack" " is not available"
1549 ret.update({"comment": comt, "name": name})
1550 self.assertDictEqual(filestate.directory(name, user=user, group=group), ret)
1551
1552 with patch.object(os.path, "isabs", mock_f):
1553 comt = "Specified file {} is not an absolute path".format(name)
1554 ret.update({"comment": comt})
1555 self.assertDictEqual(
1556 filestate.directory(name, user=user, group=group), ret
1557 )
1558
1559 with patch.object(os.path, "isabs", mock_t):
1560 with patch.object(
1561 os.path,
1562 "isfile",
1563 MagicMock(side_effect=[True, True, False, True, True, True, False]),
1564 ):
1565 with patch.object(os.path, "lexists", mock_t):
1566 comt = "File exists where the backup target" " A should go"
1567 ret.update({"comment": comt})
1568 self.assertDictEqual(
1569 filestate.directory(
1570 name, user=user, group=group, backupname="A"
1571 ),
1572 ret,
1573 )
1574
1575 with patch.object(os.path, "isfile", mock_t):
1576 comt = "Specified location {} exists and is a file".format(name)
1577 ret.update({"comment": comt})
1578 self.assertDictEqual(
1579 filestate.directory(name, user=user, group=group), ret
1580 )
1581
1582 with patch.object(os.path, "islink", mock_t):
1583 comt = "Specified location {} exists and is a symlink".format(
1584 name
1585 )
1586 ret.update({"comment": comt})
1587 self.assertDictEqual(
1588 filestate.directory(name, user=user, group=group), ret
1589 )
1590
1591 with patch.object(os.path, "isdir", mock_f):
1592 with patch.dict(filestate.__opts__, {"test": True}):
1593 if salt.utils.platform.is_windows():
1594 comt = 'The directory "{}" will be changed' "".format(name)
1595 else:
1596 comt = (
1597 "The following files will be changed:\n{}:"
1598 " directory - new\n".format(name)
1599 )
1600 ret.update(
1601 {
1602 "comment": comt,
1603 "result": None,
1604 "changes": {name: {"directory": "new"}},
1605 }
1606 )
1607 self.assertDictEqual(
1608 filestate.directory(name, user=user, group=group), ret
1609 )
1610
1611 with patch.dict(filestate.__opts__, {"test": False}):
1612 with patch.object(os.path, "isdir", mock_f):
1613 comt = "No directory to create {} in".format(name)
1614 ret.update({"comment": comt, "result": False})
1615 self.assertDictEqual(
1616 filestate.directory(name, user=user, group=group), ret
1617 )
1618
1619 if salt.utils.platform.is_windows():
1620 isdir_side_effect = [False, True, False]
1621 else:
1622 isdir_side_effect = [True, False, True, False]
1623 with patch.object(
1624 os.path, "isdir", MagicMock(side_effect=isdir_side_effect)
1625 ):
1626 comt = "Failed to create directory {}".format(name)
1627 ret.update(
1628 {
1629 "comment": comt,
1630 "result": False,
1631 "changes": {name: "New Dir"},
1632 }
1633 )
1634 self.assertDictEqual(
1635 filestate.directory(name, user=user, group=group), ret
1636 )
1637
1638 check_perms_ret = {
1639 "name": name,
1640 "result": False,
1641 "comment": "",
1642 "changes": {},
1643 }
1644 if salt.utils.platform.is_windows():
1645 mock_perms = MagicMock(return_value=check_perms_ret)
1646 else:
1647 mock_perms = MagicMock(return_value=(check_perms_ret, ""))
1648
1649 recurse = ["silent"]
1650 ret = {
1651 "name": name,
1652 "result": False,
1653 "comment": "Directory /etc/testdir updated",
1654 "changes": {"recursion": "Changes silenced"},
1655 }
1656 if salt.utils.platform.is_windows():
1657 ret["comment"] = ret["comment"].replace("/", "\\")
1658 with patch.dict(
1659 filestate.__salt__, {"file.check_perms": mock_perms}
1660 ):
1661 with patch.object(os.path, "isdir", mock_t):
1662 self.assertDictEqual(
1663 filestate.directory(
1664 name, user=user, recurse=recurse, group=group
1665 ),
1666 ret,
1667 )
1668
1669 check_perms_ret = {
1670 "name": name,
1671 "result": False,
1672 "comment": "",
1673 "changes": {},
1674 }
1675 if salt.utils.platform.is_windows():
1676 mock_perms = MagicMock(return_value=check_perms_ret)
1677 else:
1678 mock_perms = MagicMock(return_value=(check_perms_ret, ""))
1679
1680 recurse = ["ignore_files", "ignore_dirs"]
1681 ret = {
1682 "name": name,
1683 "result": False,
1684 "comment": 'Must not specify "recurse" '
1685 'options "ignore_files" and '
1686 '"ignore_dirs" at the same '
1687 "time.",
1688 "changes": {},
1689 }
1690 with patch.dict(
1691 filestate.__salt__, {"file.check_perms": mock_perms}
1692 ):
1693 with patch.object(os.path, "isdir", mock_t):
1694 self.assertDictEqual(
1695 filestate.directory(
1696 name, user=user, recurse=recurse, group=group
1697 ),
1698 ret,
1699 )
1700
1701 comt = "Directory {} updated".format(name)
1702 ret = {
1703 "name": name,
1704 "result": True,
1705 "comment": comt,
1706 "changes": {
1707 "group": "group",
1708 "mode": "0777",
1709 "user": "user",
1710 },
1711 }
1712
1713 check_perms_ret = {
1714 "name": name,
1715 "result": True,
1716 "comment": "",
1717 "changes": {
1718 "group": "group",
1719 "mode": "0777",
1720 "user": "user",
1721 },
1722 }
1723
1724 if salt.utils.platform.is_windows():
1725 _mock_perms = MagicMock(return_value=check_perms_ret)
1726 else:
1727 _mock_perms = MagicMock(return_value=(check_perms_ret, ""))
1728 with patch.object(os.path, "isdir", mock_t):
1729 with patch.dict(
1730 filestate.__salt__, {"file.check_perms": _mock_perms}
1731 ):
1732 self.assertDictEqual(
1733 filestate.directory(name, user=user, group=group),
1734 ret,
1735 )
1736
1737 # 'recurse' function tests: 1
1738
1739 def test_recurse(self):
1740 """
1741 Test to recurse through a subdirectory on the master
1742 and copy said subdirectory over to the specified path.
1743 """
1744 name = "/opt/code/flask"
1745 source = "salt://code/flask"
1746 user = "salt"
1747 group = "saltstack"
1748 if salt.utils.platform.is_windows():
1749 name = name.replace("/", "\\")
1750
1751 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1752
1753 comt = (
1754 "'mode' is not allowed in 'file.recurse'."
1755 " Please use 'file_mode' and 'dir_mode'."
1756 )
1757 ret.update({"comment": comt})
1758 self.assertDictEqual(filestate.recurse(name, source, mode="W"), ret)
1759
1760 mock_t = MagicMock(return_value=True)
1761 mock_f = MagicMock(return_value=False)
1762 mock_uid = MagicMock(return_value="")
1763 mock_gid = MagicMock(return_value="")
1764 mock_l = MagicMock(return_value=[])
1765 mock_emt = MagicMock(side_effect=[[], ["code/flask"], ["code/flask"]])
1766 mock_lst = MagicMock(
1767 side_effect=[
1768 CommandExecutionError,
1769 (source, ""),
1770 (source, ""),
1771 (source, ""),
1772 ]
1773 )
1774 with patch.dict(
1775 filestate.__salt__,
1776 {
1777 "config.manage_mode": mock_t,
1778 "file.user_to_uid": mock_uid,
1779 "file.group_to_gid": mock_gid,
1780 "file.source_list": mock_lst,
1781 "cp.list_master_dirs": mock_emt,
1782 "cp.list_master": mock_l,
1783 },
1784 ):
1785
1786 # Group argument is ignored on Windows systems. Group is set to user
1787 if salt.utils.platform.is_windows():
1788 comt = "User salt is not available Group salt" " is not available"
1789 else:
1790 comt = "User salt is not available Group saltstack" " is not available"
1791 ret.update({"comment": comt})
1792 self.assertDictEqual(
1793 filestate.recurse(name, source, user=user, group=group), ret
1794 )
1795
1796 with patch.object(os.path, "isabs", mock_f):
1797 comt = "Specified file {} is not an absolute path".format(name)
1798 ret.update({"comment": comt})
1799 self.assertDictEqual(filestate.recurse(name, source), ret)
1800
1801 with patch.object(os.path, "isabs", mock_t):
1802 comt = "Invalid source '1' (must be a salt:// URI)"
1803 ret.update({"comment": comt})
1804 self.assertDictEqual(filestate.recurse(name, 1), ret)
1805
1806 comt = "Invalid source '//code/flask' (must be a salt:// URI)"
1807 ret.update({"comment": comt})
1808 self.assertDictEqual(filestate.recurse(name, "//code/flask"), ret)
1809
1810 comt = "Recurse failed: "
1811 ret.update({"comment": comt})
1812 self.assertDictEqual(filestate.recurse(name, source), ret)
1813
1814 comt = (
1815 "The directory 'code/flask' does not exist"
1816 " on the salt fileserver in saltenv 'base'"
1817 )
1818 ret.update({"comment": comt})
1819 self.assertDictEqual(filestate.recurse(name, source), ret)
1820
1821 with patch.object(os.path, "isdir", mock_f):
1822 with patch.object(os.path, "exists", mock_t):
1823 comt = "The path {} exists and is not a directory".format(name)
1824 ret.update({"comment": comt})
1825 self.assertDictEqual(filestate.recurse(name, source), ret)
1826
1827 with patch.object(os.path, "isdir", mock_t):
1828 comt = "The directory {} is in the correct state".format(name)
1829 ret.update({"comment": comt, "result": True})
1830 self.assertDictEqual(filestate.recurse(name, source), ret)
1831
1832 # 'replace' function tests: 1
1833
1834 def test_replace(self):
1835 """
1836 Test to maintain an edit in a file.
1837 """
1838 name = "/etc/grub.conf"
1839 pattern = "CentOS +"
1840 repl = "salt"
1841
1842 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1843
1844 comt = "Must provide name to file.replace"
1845 ret.update({"comment": comt, "name": "", "changes": {}})
1846 self.assertDictEqual(filestate.replace("", pattern, repl), ret)
1847
1848 mock_t = MagicMock(return_value=True)
1849 mock_f = MagicMock(return_value=False)
1850 with patch.object(os.path, "isabs", mock_f):
1851 comt = "Specified file {} is not an absolute path".format(name)
1852 ret.update({"comment": comt, "name": name})
1853 self.assertDictEqual(filestate.replace(name, pattern, repl), ret)
1854
1855 with patch.object(os.path, "isabs", mock_t):
1856 with patch.object(os.path, "exists", mock_t):
1857 with patch.dict(filestate.__salt__, {"file.replace": mock_f}):
1858 with patch.dict(filestate.__opts__, {"test": False}):
1859 comt = "No changes needed to be made"
1860 ret.update({"comment": comt, "name": name, "result": True})
1861 self.assertDictEqual(
1862 filestate.replace(name, pattern, repl), ret
1863 )
1864
1865 # 'blockreplace' function tests: 1
1866
1867 def test_blockreplace(self):
1868 """
1869 Test to maintain an edit in a file in a zone
1870 delimited by two line markers.
1871 """
1872 with patch(
1873 "salt.states.file._load_accumulators", MagicMock(return_value=([], []))
1874 ):
1875 name = "/etc/hosts"
1876
1877 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1878
1879 comt = "Must provide name to file.blockreplace"
1880 ret.update({"comment": comt, "name": ""})
1881 self.assertDictEqual(filestate.blockreplace(""), ret)
1882
1883 mock_t = MagicMock(return_value=True)
1884 mock_f = MagicMock(return_value=False)
1885 with patch.object(os.path, "isabs", mock_f):
1886 comt = "Specified file {} is not an absolute path".format(name)
1887 ret.update({"comment": comt, "name": name})
1888 self.assertDictEqual(filestate.blockreplace(name), ret)
1889
1890 with patch.object(os.path, "isabs", mock_t), patch.object(
1891 os.path, "exists", mock_t
1892 ):
1893 with patch.dict(filestate.__salt__, {"file.blockreplace": mock_t}):
1894 with patch.dict(filestate.__opts__, {"test": True}):
1895 comt = "Changes would be made"
1896 ret.update(
1897 {"comment": comt, "result": None, "changes": {"diff": True}}
1898 )
1899 self.assertDictEqual(filestate.blockreplace(name), ret)
1900
1901 # 'comment' function tests: 1
1902
1903 def test_comment(self):
1904 """
1905 Test to comment out specified lines in a file.
1906 """
1907 with patch.object(os.path, "exists", MagicMock(return_value=True)):
1908 name = "/etc/aliases" if salt.utils.platform.is_darwin() else "/etc/fstab"
1909 regex = "bind 127.0.0.1"
1910
1911 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1912
1913 comt = "Must provide name to file.comment"
1914 ret.update({"comment": comt, "name": ""})
1915 self.assertDictEqual(filestate.comment("", regex), ret)
1916
1917 mock_t = MagicMock(return_value=True)
1918 mock_f = MagicMock(return_value=False)
1919 with patch.object(os.path, "isabs", mock_f):
1920 comt = "Specified file {} is not an absolute path".format(name)
1921 ret.update({"comment": comt, "name": name})
1922 self.assertDictEqual(filestate.comment(name, regex), ret)
1923
1924 with patch.object(os.path, "isabs", mock_t):
1925 with patch.dict(
1926 filestate.__salt__,
1927 {"file.search": MagicMock(side_effect=[False, True, False, False])},
1928 ):
1929 comt = "Pattern already commented"
1930 ret.update({"comment": comt, "result": True})
1931 self.assertDictEqual(filestate.comment(name, regex), ret)
1932
1933 comt = "{}: Pattern not found".format(regex)
1934 ret.update({"comment": comt, "result": False})
1935 self.assertDictEqual(filestate.comment(name, regex), ret)
1936
1937 with patch.dict(
1938 filestate.__salt__,
1939 {
1940 "file.search": MagicMock(side_effect=[True, True, True]),
1941 "file.comment": mock_t,
1942 "file.comment_line": mock_t,
1943 },
1944 ):
1945 with patch.dict(filestate.__opts__, {"test": True}):
1946 comt = "File {} is set to be updated".format(name)
1947 ret.update(
1948 {
1949 "comment": comt,
1950 "result": None,
1951 "changes": {name: "updated"},
1952 }
1953 )
1954 self.assertDictEqual(filestate.comment(name, regex), ret)
1955
1956 with patch.dict(filestate.__opts__, {"test": False}):
1957 with patch.object(
1958 salt.utils.files, "fopen", MagicMock(mock_open())
1959 ):
1960 comt = "Commented lines successfully"
1961 ret.update({"comment": comt, "result": True, "changes": {}})
1962 self.assertDictEqual(filestate.comment(name, regex), ret)
1963
1964 # 'uncomment' function tests: 1
1965
1966 def test_uncomment(self):
1967 """
1968 Test to uncomment specified commented lines in a file
1969 """
1970 with patch.object(os.path, "exists", MagicMock(return_value=True)):
1971 name = "/etc/aliases" if salt.utils.platform.is_darwin() else "/etc/fstab"
1972 regex = "bind 127.0.0.1"
1973
1974 ret = {"name": name, "result": False, "comment": "", "changes": {}}
1975
1976 comt = "Must provide name to file.uncomment"
1977 ret.update({"comment": comt, "name": ""})
1978 self.assertDictEqual(filestate.uncomment("", regex), ret)
1979
1980 mock_t = MagicMock(return_value=True)
1981 mock_f = MagicMock(return_value=False)
1982 mock = MagicMock(side_effect=[False, True, False, False, True, True, True])
1983 with patch.object(os.path, "isabs", mock_f):
1984 comt = "Specified file {} is not an absolute path".format(name)
1985 ret.update({"comment": comt, "name": name})
1986 self.assertDictEqual(filestate.uncomment(name, regex), ret)
1987
1988 with patch.object(os.path, "isabs", mock_t):
1989 with patch.dict(
1990 filestate.__salt__,
1991 {
1992 "file.search": mock,
1993 "file.uncomment": mock_t,
1994 "file.comment_line": mock_t,
1995 },
1996 ):
1997 comt = "Pattern already uncommented"
1998 ret.update({"comment": comt, "result": True})
1999 self.assertDictEqual(filestate.uncomment(name, regex), ret)
2000
2001 comt = "{}: Pattern not found".format(regex)
2002 ret.update({"comment": comt, "result": False})
2003 self.assertDictEqual(filestate.uncomment(name, regex), ret)
2004
2005 with patch.dict(filestate.__opts__, {"test": True}):
2006 comt = "File {} is set to be updated".format(name)
2007 ret.update(
2008 {
2009 "comment": comt,
2010 "result": None,
2011 "changes": {name: "updated"},
2012 }
2013 )
2014 self.assertDictEqual(filestate.uncomment(name, regex), ret)
2015
2016 with patch.dict(filestate.__opts__, {"test": False}):
2017 with patch.object(
2018 salt.utils.files, "fopen", MagicMock(mock_open())
2019 ):
2020 comt = "Uncommented lines successfully"
2021 ret.update({"comment": comt, "result": True, "changes": {}})
2022 self.assertDictEqual(filestate.uncomment(name, regex), ret)
2023
2024 # 'prepend' function tests: 1
2025
2026 def test_prepend(self):
2027 """
2028 Test to ensure that some text appears at the beginning of a file.
2029 """
2030 name = "/tmp/etc/motd"
2031 if salt.utils.platform.is_windows():
2032 name = "c:\\tmp\\etc\\motd"
2033 assert not os.path.exists(os.path.split(name)[0])
2034 source = ["salt://motd/hr-messages.tmpl"]
2035 sources = ["salt://motd/devops-messages.tmpl"]
2036 text = ["Trust no one unless you have eaten much salt with him."]
2037
2038 ret = {"name": name, "result": False, "comment": "", "changes": {}}
2039
2040 comt = "Must provide name to file.prepend"
2041 ret.update({"comment": comt, "name": ""})
2042 self.assertDictEqual(filestate.prepend(""), ret)
2043
2044 comt = "source and sources are mutually exclusive"
2045 ret.update({"comment": comt, "name": name})
2046 self.assertDictEqual(
2047 filestate.prepend(name, source=source, sources=sources), ret
2048 )
2049
2050 mock_t = MagicMock(return_value=True)
2051 mock_f = MagicMock(return_value=False)
2052 with patch.dict(
2053 filestate.__salt__,
2054 {
2055 "file.directory_exists": mock_f,
2056 "file.makedirs": mock_t,
2057 "file.stats": mock_f,
2058 "cp.get_template": mock_f,
2059 "file.search": mock_f,
2060 "file.prepend": mock_t,
2061 },
2062 ):
2063 comt = (
2064 "The following files will be changed:\n/tmp/etc:" " directory - new\n"
2065 )
2066 changes = {"/tmp/etc": {"directory": "new"}}
2067 if salt.utils.platform.is_windows():
2068 comt = 'The directory "c:\\tmp\\etc" will be changed'
2069 changes = {"c:\\tmp\\etc": {"directory": "new"}}
2070 ret.update({"comment": comt, "name": name, "changes": changes})
2071 self.assertDictEqual(filestate.prepend(name, makedirs=True), ret)
2072
2073 with patch.object(os.path, "isabs", mock_f):
2074 comt = "Specified file {} is not an absolute path".format(name)
2075 ret.update({"comment": comt, "changes": {}})
2076 self.assertDictEqual(filestate.prepend(name), ret)
2077
2078 with patch.object(os.path, "isabs", mock_t):
2079 with patch.object(os.path, "exists", mock_t):
2080 comt = "Failed to load template file {}".format(source)
2081 ret.update({"comment": comt, "name": source, "data": []})
2082 self.assertDictEqual(filestate.prepend(name, source=source), ret)
2083
2084 ret.pop("data", None)
2085 ret.update({"name": name})
2086 with patch.object(
2087 salt.utils.files, "fopen", MagicMock(mock_open(read_data=""))
2088 ):
2089 with patch.dict(filestate.__utils__, {"files.is_text": mock_f}):
2090 with patch.dict(filestate.__opts__, {"test": True}):
2091 change = {"diff": "Replace binary file"}
2092 comt = "File {} is set to be updated".format(name)
2093 ret.update(
2094 {"comment": comt, "result": None, "changes": change}
2095 )
2096 self.assertDictEqual(
2097 filestate.prepend(name, text=text), ret
2098 )
2099
2100 with patch.dict(filestate.__opts__, {"test": False}):
2101 comt = "Prepended 1 lines"
2102 ret.update(
2103 {"comment": comt, "result": True, "changes": {}}
2104 )
2105 self.assertDictEqual(
2106 filestate.prepend(name, text=text), ret
2107 )
2108
2109 # 'touch' function tests: 1
2110
2111 def test_touch(self):
2112 """
2113 Test to replicate the 'nix "touch" command to create a new empty
2114 file or update the atime and mtime of an existing file.
2115 """
2116 name = "/var/log/httpd/logrotate.empty"
2117
2118 ret = {"name": name, "result": False, "comment": "", "changes": {}}
2119
2120 comt = "Must provide name to file.touch"
2121 ret.update({"comment": comt, "name": ""})
2122 self.assertDictEqual(filestate.touch(""), ret)
2123
2124 mock_t = MagicMock(return_value=True)
2125 mock_f = MagicMock(return_value=False)
2126 with patch.object(os.path, "isabs", mock_f):
2127 comt = "Specified file {} is not an absolute path".format(name)
2128 ret.update({"comment": comt, "name": name})
2129 self.assertDictEqual(filestate.touch(name), ret)
2130
2131 with patch.object(os.path, "isabs", mock_t):
2132 with patch.object(os.path, "exists", mock_f):
2133 with patch.dict(filestate.__opts__, {"test": True}):
2134 comt = "File {} is set to be created".format(name)
2135 ret.update(
2136 {"comment": comt, "result": None, "changes": {"new": name}}
2137 )
2138 self.assertDictEqual(filestate.touch(name), ret)
2139
2140 with patch.dict(filestate.__opts__, {"test": False}):
2141 with patch.object(os.path, "isdir", mock_f):
2142 comt = "Directory not present to touch file {}".format(name)
2143 ret.update({"comment": comt, "result": False, "changes": {}})
2144 self.assertDictEqual(filestate.touch(name), ret)
2145
2146 with patch.object(os.path, "isdir", mock_t):
2147 with patch.dict(filestate.__salt__, {"file.touch": mock_t}):
2148 comt = "Created empty file {}".format(name)
2149 ret.update(
2150 {"comment": comt, "result": True, "changes": {"new": name}}
2151 )
2152 self.assertDictEqual(filestate.touch(name), ret)
2153
2154 # 'copy' function tests: 1
2155
2156 def test_copy(self):
2157 """
2158 Test if the source file exists on the system, copy it to the named file.
2159 """
2160 name = "/tmp/salt"
2161 source = "/tmp/salt/salt"
2162 user = "salt"
2163 group = "saltstack"
2164
2165 ret = {"name": name, "result": False, "comment": "", "changes": {}}
2166
2167 comt = "Must provide name to file.copy"
2168 ret.update({"comment": comt, "name": ""})
2169 self.assertDictEqual(filestate.copy_("", source), ret)
2170
2171 mock_t = MagicMock(return_value=True)
2172 mock_f = MagicMock(return_value=False)
2173 mock_uid = MagicMock(side_effect=[""])
2174 mock_gid = MagicMock(side_effect=[""])
2175 mock_user = MagicMock(return_value=user)
2176 mock_grp = MagicMock(return_value=group)
2177 mock_io = MagicMock(side_effect=IOError)
2178 with patch.object(os.path, "isabs", mock_f):
2179 comt = "Specified file {} is not an absolute path".format(name)
2180 ret.update({"comment": comt, "name": name})
2181 self.assertDictEqual(filestate.copy_(name, source), ret)
2182
2183 with patch.object(os.path, "isabs", mock_t):
2184 with patch.object(os.path, "exists", mock_f):
2185 comt = 'Source file "{}" is not present'.format(source)
2186 ret.update({"comment": comt, "result": False})
2187 self.assertDictEqual(filestate.copy_(name, source), ret)
2188
2189 with patch.object(os.path, "exists", mock_t):
2190 with patch.dict(
2191 filestate.__salt__,
2192 {
2193 "file.user_to_uid": mock_uid,
2194 "file.group_to_gid": mock_gid,
2195 "file.get_user": mock_user,
2196 "file.get_group": mock_grp,
2197 "file.get_mode": mock_grp,
2198 "file.check_perms": mock_t,
2199 },
2200 ):
2201
2202 # Group argument is ignored on Windows systems. Group is set
2203 # to user
2204 if salt.utils.platform.is_windows():
2205 comt = (
2206 "User salt is not available Group salt" " is not available"
2207 )
2208 else:
2209 comt = (
2210 "User salt is not available Group saltstack"
2211 " is not available"
2212 )
2213 ret.update({"comment": comt, "result": False})
2214 self.assertDictEqual(
2215 filestate.copy_(name, source, user=user, group=group), ret
2216 )
2217
2218 comt1 = (
2219 'Failed to delete "{}" in preparation for'
2220 " forced move".format(name)
2221 )
2222 comt2 = (
2223 'The target file "{}" exists and will not be '
2224 "overwritten".format(name)
2225 )
2226 comt3 = 'File "{}" is set to be copied to "{}"'.format(source, name)
2227 with patch.object(os.path, "isdir", mock_f):
2228 with patch.object(os.path, "lexists", mock_t):
2229 with patch.dict(filestate.__opts__, {"test": False}):
2230 with patch.dict(
2231 filestate.__salt__, {"file.remove": mock_io}
2232 ):
2233 ret.update({"comment": comt1, "result": False})
2234 self.assertDictEqual(
2235 filestate.copy_(
2236 name, source, preserve=True, force=True
2237 ),
2238 ret,
2239 )
2240
2241 with patch.object(os.path, "isfile", mock_t):
2242 ret.update({"comment": comt2, "result": True})
2243 self.assertDictEqual(
2244 filestate.copy_(name, source, preserve=True),
2245 ret,
2246 )
2247
2248 with patch.object(os.path, "lexists", mock_f):
2249 with patch.dict(filestate.__opts__, {"test": True}):
2250 ret.update({"comment": comt3, "result": None})
2251 self.assertDictEqual(
2252 filestate.copy_(name, source, preserve=True), ret
2253 )
2254
2255 with patch.dict(filestate.__opts__, {"test": False}):
2256 comt = "The target directory /tmp is" " not present"
2257 ret.update({"comment": comt, "result": False})
2258 self.assertDictEqual(
2259 filestate.copy_(name, source, preserve=True), ret
2260 )
2261
2262 # 'rename' function tests: 1
2263
2264 def test_rename(self):
2265 """
2266 Test if the source file exists on the system,
2267 rename it to the named file.
2268 """
2269 name = "/tmp/salt"
2270 source = "/tmp/salt/salt"
2271
2272 ret = {"name": name, "result": False, "comment": "", "changes": {}}
2273
2274 comt = "Must provide name to file.rename"
2275 ret.update({"comment": comt, "name": ""})
2276 self.assertDictEqual(filestate.rename("", source), ret)
2277
2278 mock_t = MagicMock(return_value=True)
2279 mock_f = MagicMock(return_value=False)
2280
2281 mock_lex = MagicMock(side_effect=[False, True, True])
2282 with patch.object(os.path, "isabs", mock_f):
2283 comt = "Specified file {} is not an absolute path".format(name)
2284 ret.update({"comment": comt, "name": name})
2285 self.assertDictEqual(filestate.rename(name, source), ret)
2286
2287 mock_lex = MagicMock(return_value=False)
2288 with patch.object(os.path, "isabs", mock_t):
2289 with patch.object(os.path, "lexists", mock_lex):
2290 comt = 'Source file "{}" has already been moved out of ' "place".format(
2291 source
2292 )
2293 ret.update({"comment": comt, "result": True})
2294 self.assertDictEqual(filestate.rename(name, source), ret)
2295
2296 mock_lex = MagicMock(side_effect=[True, True, True])
2297 with patch.object(os.path, "isabs", mock_t):
2298 with patch.object(os.path, "lexists", mock_lex):
2299 comt = (
2300 'The target file "{}" exists and will not be '
2301 "overwritten".format(name)
2302 )
2303 ret.update({"comment": comt, "result": True})
2304 self.assertDictEqual(filestate.rename(name, source), ret)
2305
2306 mock_lex = MagicMock(side_effect=[True, True, True])
2307 mock_rem = MagicMock(side_effect=IOError)
2308 with patch.object(os.path, "isabs", mock_t):
2309 with patch.object(os.path, "lexists", mock_lex):
2310 with patch.dict(filestate.__opts__, {"test": False}):
2311 comt = (
2312 'Failed to delete "{}" in preparation for '
2313 "forced move".format(name)
2314 )
2315 with patch.dict(filestate.__salt__, {"file.remove": mock_rem}):
2316 ret.update({"name": name, "comment": comt, "result": False})
2317 self.assertDictEqual(
2318 filestate.rename(name, source, force=True), ret
2319 )
2320
2321 mock_lex = MagicMock(side_effect=[True, False, False])
2322 with patch.object(os.path, "isabs", mock_t):
2323 with patch.object(os.path, "lexists", mock_lex):
2324 with patch.dict(filestate.__opts__, {"test": True}):
2325 comt = 'File "{}" is set to be moved to "{}"'.format(source, name)
2326 ret.update({"name": name, "comment": comt, "result": None})
2327 self.assertDictEqual(filestate.rename(name, source), ret)
2328
2329 mock_lex = MagicMock(side_effect=[True, False, False])
2330 with patch.object(os.path, "isabs", mock_t):
2331 with patch.object(os.path, "lexists", mock_lex):
2332 with patch.object(os.path, "isdir", mock_f):
2333 with patch.dict(filestate.__opts__, {"test": False}):
2334 comt = "The target directory /tmp is not present"
2335 ret.update({"name": name, "comment": comt, "result": False})
2336 self.assertDictEqual(filestate.rename(name, source), ret)
2337
2338 mock_lex = MagicMock(side_effect=[True, False, False])
2339 with patch.object(os.path, "isabs", mock_t):
2340 with patch.object(os.path, "lexists", mock_lex):
2341 with patch.object(os.path, "isdir", mock_t):
2342 with patch.object(os.path, "islink", mock_f):
2343 with patch.dict(filestate.__opts__, {"test": False}):
2344 with patch.object(
2345 shutil, "move", MagicMock(side_effect=IOError)
2346 ):
2347 comt = 'Failed to move "{}" to "{}"'.format(
2348 source, name
2349 )
2350 ret.update(
2351 {"name": name, "comment": comt, "result": False}
2352 )
2353 self.assertDictEqual(
2354 filestate.rename(name, source), ret
2355 )
2356
2357 mock_lex = MagicMock(side_effect=[True, False, False])
2358 with patch.object(os.path, "isabs", mock_t):
2359 with patch.object(os.path, "lexists", mock_lex):
2360 with patch.object(os.path, "isdir", mock_t):
2361 with patch.object(os.path, "islink", mock_f):
2362 with patch.dict(filestate.__opts__, {"test": False}):
2363 with patch.object(shutil, "move", MagicMock()):
2364 comt = 'Moved "{}" to "{}"'.format(source, name)
2365 ret.update(
2366 {
2367 "name": name,
2368 "comment": comt,
2369 "result": True,
2370 "changes": {name: source},
2371 }
2372 )
2373 self.assertDictEqual(
2374 filestate.rename(name, source), ret
2375 )
2376
2377 # 'accumulated' function tests: 1
2378
2379 def test_accumulated(self):
2380 """
2381 Test to prepare accumulator which can be used in template in file.
2382 """
2383 with patch(
2384 "salt.states.file._load_accumulators", MagicMock(return_value=({}, {}))
2385 ), patch(
2386 "salt.states.file._persist_accummulators", MagicMock(return_value=True)
2387 ):
2388 name = "animals_doing_things"
2389 filename = "/tmp/animal_file.txt"
2390 text = " jumps over the lazy dog."
2391
2392 ret = {"name": name, "result": False, "comment": "", "changes": {}}
2393
2394 comt = "Must provide name to file.accumulated"
2395 ret.update({"comment": comt, "name": ""})
2396 self.assertDictEqual(filestate.accumulated("", filename, text), ret)
2397
2398 comt = "No text supplied for accumulator"
2399 ret.update({"comment": comt, "name": name})
2400 self.assertDictEqual(filestate.accumulated(name, filename, None), ret)
2401
2402 with patch.dict(
2403 filestate.__low__,
2404 {
2405 "require_in": "file",
2406 "watch_in": "salt",
2407 "__sls__": "SLS",
2408 "__id__": "ID",
2409 },
2410 ):
2411 comt = "Orphaned accumulator animals_doing_things in SLS:ID"
2412 ret.update({"comment": comt, "name": name})
2413 self.assertDictEqual(filestate.accumulated(name, filename, text), ret)
2414
2415 with patch.dict(
2416 filestate.__low__,
2417 {
2418 "require_in": [{"file": "A"}],
2419 "watch_in": [{"B": "C"}],
2420 "__sls__": "SLS",
2421 "__id__": "ID",
2422 },
2423 ):
2424 comt = "Accumulator {} for file {} " "was charged by text".format(
2425 name, filename
2426 )
2427 ret.update({"comment": comt, "name": name, "result": True})
2428 self.assertDictEqual(filestate.accumulated(name, filename, text), ret)
2429
2430 # 'serialize' function tests: 1
2431
2432 def test_serialize_into_managed_file(self):
2433 """
2434 Test to serializes dataset and store it into managed file.
2435 """
2436 name = "/etc/dummy/package.json"
2437
2438 ret = {"name": name, "result": False, "comment": "", "changes": {}}
2439
2440 comt = "Must provide name to file.serialize"
2441 ret.update({"comment": comt, "name": ""})
2442 self.assertDictEqual(filestate.serialize(""), ret)
2443
2444 mock_t = MagicMock(return_value=True)
2445 mock_f = MagicMock(return_value=False)
2446 with patch.object(os.path, "isfile", mock_f):
2447 comt = "File {} is not present and is not set for " "creation".format(
2448 name
2449 )
2450 ret.update({"comment": comt, "name": name, "result": True})
2451 self.assertDictEqual(filestate.serialize(name, create=False), ret)
2452
2453 comt = "Only one of 'dataset' and 'dataset_pillar' is permitted"
2454 ret.update({"comment": comt, "result": False})
2455 self.assertDictEqual(
2456 filestate.serialize(name, dataset=True, dataset_pillar=True), ret
2457 )
2458
2459 comt = "Neither 'dataset' nor 'dataset_pillar' was defined"
2460 ret.update({"comment": comt, "result": False})
2461 self.assertDictEqual(filestate.serialize(name), ret)
2462
2463 with patch.object(os.path, "isfile", mock_t):
2464 comt = "Python format is not supported for merging"
2465 ret.update({"comment": comt, "result": False})
2466 self.assertDictEqual(
2467 filestate.serialize(
2468 name, dataset=True, merge_if_exists=True, formatter="python"
2469 ),
2470 ret,
2471 )
2472
2473 comt = "A format is not supported"
2474 ret.update({"comment": comt, "result": False})
2475 self.assertDictEqual(
2476 filestate.serialize(name, dataset=True, formatter="A"), ret
2477 )
2478 mock_changes = MagicMock(return_value=True)
2479 mock_no_changes = MagicMock(return_value=False)
2480
2481 # __opts__['test']=True with changes
2482 with patch.dict(
2483 filestate.__salt__, {"file.check_managed_changes": mock_changes}
2484 ):
2485 with patch.dict(filestate.__opts__, {"test": True}):
2486 comt = "Dataset will be serialized and stored into {}".format(name)
2487 ret.update({"comment": comt, "result": None, "changes": True})
2488 self.assertDictEqual(
2489 filestate.serialize(name, dataset=True, formatter="python"), ret
2490 )
2491
2492 # __opts__['test']=True without changes
2493 with patch.dict(
2494 filestate.__salt__, {"file.check_managed_changes": mock_no_changes}
2495 ):
2496 with patch.dict(filestate.__opts__, {"test": True}):
2497 comt = "The file {} is in the correct state".format(name)
2498 ret.update({"comment": comt, "result": True, "changes": False})
2499 self.assertDictEqual(
2500 filestate.serialize(name, dataset=True, formatter="python"), ret
2501 )
2502
2503 mock = MagicMock(return_value=ret)
2504 with patch.dict(filestate.__opts__, {"test": False}):
2505 with patch.dict(filestate.__salt__, {"file.manage_file": mock}):
2506 comt = "Dataset will be serialized and stored into {}".format(name)
2507 ret.update({"comment": comt, "result": None})
2508 self.assertDictEqual(
2509 filestate.serialize(name, dataset=True, formatter="python"), ret
2510 )
2511
2512 # 'mknod' function tests: 1
2513
2514 def test_mknod(self):
2515 """
2516 Test to create a special file similar to the 'nix mknod command.
2517 """
2518 name = "/dev/AA"
2519 ntype = "a"
2520
2521 ret = {"name": name, "result": False, "comment": "", "changes": {}}
2522
2523 comt = "Must provide name to file.mknod"
2524 ret.update({"comment": comt, "name": ""})
2525 self.assertDictEqual(filestate.mknod("", ntype), ret)
2526
2527 comt = (
2528 "Node type unavailable: 'a'. Available node types are "
2529 "character ('c'), block ('b'), and pipe ('p')"
2530 )
2531 ret.update({"comment": comt, "name": name})
2532 self.assertDictEqual(filestate.mknod(name, ntype), ret)
2533
2534 # 'mod_run_check_cmd' function tests: 1
2535
2536 def test_mod_run_check_cmd(self):
2537 """
2538 Test to execute the check_cmd logic.
2539 """
2540 cmd = "A"
2541 filename = "B"
2542
2543 ret = {
2544 "comment": "check_cmd execution failed",
2545 "result": False,
2546 "skip_watch": True,
2547 }
2548
2549 mock = MagicMock(side_effect=[{"retcode": 1}, {"retcode": 0}])
2550 with patch.dict(filestate.__salt__, {"cmd.run_all": mock}):
2551 self.assertDictEqual(filestate.mod_run_check_cmd(cmd, filename), ret)
2552
2553 self.assertTrue(filestate.mod_run_check_cmd(cmd, filename))
2554
2555 @skipIf(not HAS_DATEUTIL, NO_DATEUTIL_REASON)
2556 @slowTest
2557 def test_retention_schedule(self):
2558 """
2559 Test to execute the retention_schedule logic.
2560
2561 This test takes advantage of knowing which files it is generating,
2562 which means it can easily generate list of which files it should keep.
2563 """
2564
2565 def generate_fake_files(
2566 format="example_name_%Y%m%dT%H%M%S.tar.bz2",
2567 starting=datetime(2016, 2, 8, 9),
2568 every=relativedelta(minutes=30),
2569 ending=datetime(2015, 12, 25),
2570 maxfiles=None,
2571 ):
2572 """
2573 For starting, make sure that it's over a week from the beginning of the month
2574 For every, pick only one of minutes, hours, days, weeks, months or years
2575 For ending, the further away it is from starting, the slower the tests run
2576 Full coverage requires over a year of separation, but that's painfully slow.
2577 """
2578
2579 if every.years:
2580 ts = datetime(starting.year, 1, 1)
2581 elif every.months:
2582 ts = datetime(starting.year, starting.month, 1)
2583 elif every.days:
2584 ts = datetime(starting.year, starting.month, starting.day)
2585 elif every.hours:
2586 ts = datetime(
2587 starting.year, starting.month, starting.day, starting.hour
2588 )
2589 elif every.minutes:
2590 ts = datetime(
2591 starting.year, starting.month, starting.day, starting.hour, 0
2592 )
2593 else:
2594 raise NotImplementedError("not sure what you're trying to do here")
2595
2596 fake_files = []
2597 count = 0
2598 while ending < ts:
2599 fake_files.append(ts.strftime(format=format))
2600 count += 1
2601 if maxfiles and maxfiles == "all" or maxfiles and count >= maxfiles:
2602 break
2603 ts -= every
2604 return fake_files
2605
2606 fake_name = "/some/dir/name"
2607 fake_retain = {
2608 "most_recent": 2,
2609 "first_of_hour": 4,
2610 "first_of_day": 7,
2611 "first_of_week": 6,
2612 "first_of_month": 6,
2613 "first_of_year": "all",
2614 }
2615 fake_strptime_format = "example_name_%Y%m%dT%H%M%S.tar.bz2"
2616 fake_matching_file_list = generate_fake_files()
2617 # Add some files which do not match fake_strptime_format
2618 fake_no_match_file_list = generate_fake_files(
2619 format="no_match_%Y%m%dT%H%M%S.tar.bz2", every=relativedelta(days=1)
2620 )
2621
2622 def lstat_side_effect(path):
2623 import re
2624 from time import mktime
2625
2626 x = re.match(r"^[^\d]*(\d{8}T\d{6})\.tar\.bz2$", path).group(1)
2627 ts = mktime(datetime.strptime(x, "%Y%m%dT%H%M%S").timetuple())
2628 return {
2629 "st_atime": 0.0,
2630 "st_ctime": 0.0,
2631 "st_gid": 0,
2632 "st_mode": 33188,
2633 "st_mtime": ts,
2634 "st_nlink": 1,
2635 "st_size": 0,
2636 "st_uid": 0,
2637 }
2638
2639 mock_t = MagicMock(return_value=True)
2640 mock_f = MagicMock(return_value=False)
2641 mock_lstat = MagicMock(side_effect=lstat_side_effect)
2642 mock_remove = MagicMock()
2643
2644 def run_checks(isdir=mock_t, strptime_format=None, test=False):
2645 expected_ret = {
2646 "name": fake_name,
2647 "changes": {"retained": [], "deleted": [], "ignored": []},
2648 "result": True,
2649 "comment": "Name provided to file.retention must be a directory",
2650 }
2651 if strptime_format:
2652 fake_file_list = sorted(
2653 fake_matching_file_list + fake_no_match_file_list
2654 )
2655 else:
2656 fake_file_list = sorted(fake_matching_file_list)
2657 mock_readdir = MagicMock(return_value=fake_file_list)
2658
2659 with patch.dict(filestate.__opts__, {"test": test}):
2660 with patch.object(os.path, "isdir", isdir):
2661 mock_readdir.reset_mock()
2662 with patch.dict(filestate.__salt__, {"file.readdir": mock_readdir}):
2663 with patch.dict(filestate.__salt__, {"file.lstat": mock_lstat}):
2664 mock_remove.reset_mock()
2665 with patch.dict(
2666 filestate.__salt__, {"file.remove": mock_remove}
2667 ):
2668 if strptime_format:
2669 actual_ret = filestate.retention_schedule(
2670 fake_name,
2671 fake_retain,
2672 strptime_format=fake_strptime_format,
2673 )
2674 else:
2675 actual_ret = filestate.retention_schedule(
2676 fake_name, fake_retain
2677 )
2678
2679 if not isdir():
2680 mock_readdir.assert_has_calls([])
2681 expected_ret["result"] = False
2682 else:
2683 mock_readdir.assert_called_once_with(fake_name)
2684 ignored_files = fake_no_match_file_list if strptime_format else []
2685 retained_files = set(
2686 generate_fake_files(maxfiles=fake_retain["most_recent"])
2687 )
2688 junk_list = [
2689 ("first_of_hour", relativedelta(hours=1)),
2690 ("first_of_day", relativedelta(days=1)),
2691 ("first_of_week", relativedelta(weeks=1)),
2692 ("first_of_month", relativedelta(months=1)),
2693 ("first_of_year", relativedelta(years=1)),
2694 ]
2695 for retainable, retain_interval in junk_list:
2696 new_retains = set(
2697 generate_fake_files(
2698 maxfiles=fake_retain[retainable], every=retain_interval
2699 )
2700 )
2701 # if we generate less than the number of files expected,
2702 # then the oldest file will also be retained
2703 # (correctly, since its the first in it's category)
2704 if (
2705 fake_retain[retainable] == "all"
2706 or len(new_retains) < fake_retain[retainable]
2707 ):
2708 new_retains.add(fake_file_list[0])
2709 retained_files |= new_retains
2710
2711 deleted_files = sorted(
2712 list(set(fake_file_list) - retained_files - set(ignored_files)),
2713 reverse=True,
2714 )
2715 retained_files = sorted(list(retained_files), reverse=True)
2716 expected_ret["changes"] = {
2717 "retained": retained_files,
2718 "deleted": deleted_files,
2719 "ignored": ignored_files,
2720 }
2721 if test:
2722 expected_ret["result"] = None
2723 expected_ret["comment"] = (
2724 "{} backups would have been removed from {}.\n"
2725 "".format(len(deleted_files), fake_name)
2726 )
2727 else:
2728 expected_ret["comment"] = (
2729 "{} backups were removed from {}.\n"
2730 "".format(len(deleted_files), fake_name)
2731 )
2732 mock_remove.assert_has_calls(
2733 [call(os.path.join(fake_name, x)) for x in deleted_files],
2734 any_order=True,
2735 )
2736
2737 self.assertDictEqual(actual_ret, expected_ret)
2738
2739 run_checks(isdir=mock_f)
2740 run_checks()
2741 run_checks(test=True)
2742 run_checks(strptime_format=fake_strptime_format)
2743 run_checks(strptime_format=fake_strptime_format, test=True)
2744
2745 @with_tempfile()
2746 def test_file_keyvalue_key_values(self, fpath):
2747 """
2748 test file.keyvalue when using key_values kwarg
2749 """
2750 content = dedent(
2751 """\
2752 #PermitRootLogin prohibit-password
2753 #StrictMode yes
2754 """
2755 )
2756
2757 with salt.utils.files.fopen(fpath, "w+") as fp_:
2758 fp_.write(content)
2759
2760 ret = filestate.keyvalue(
2761 name=fpath,
2762 key_values=collections.OrderedDict(PermitRootLogin="yes"),
2763 separator=" ",
2764 uncomment="#",
2765 key_ignore_case=True,
2766 )
2767
2768 with salt.utils.files.fopen(fpath, "r") as fp_:
2769 f_contents = fp_.read()
2770 self.assertIn("PermitRootLogin yes", f_contents)
2771 self.assertIn("#StrictMode yes", f_contents)
2772
2773 @with_tempfile()
2774 def test_file_keyvalue_empty(self, fpath):
2775 """
2776 test file.keyvalue when key_values is empty
2777 """
2778 content = dedent(
2779 """\
2780 #PermitRootLogin prohibit-password
2781 #StrictMode yes
2782 """
2783 )
2784
2785 with salt.utils.files.fopen(fpath, "w+") as fp_:
2786 fp_.write(content)
2787
2788 ret = filestate.keyvalue(
2789 name=fpath,
2790 key_values={},
2791 separator=" ",
2792 uncomment="#",
2793 key_ignore_case=True,
2794 )
2795
2796 self.assertEqual(
2797 ret["comment"],
2798 "file.keyvalue key and value not supplied and key_values is empty",
2799 )
2800 with salt.utils.files.fopen(fpath, "r") as fp_:
2801 f_contents = fp_.read()
2802 self.assertNotIn("PermitRootLogin yes", f_contents)
2803 self.assertIn("#StrictMode yes", f_contents)
2804
2805 @with_tempfile()
2806 def test_file_keyvalue_not_dict(self, fpath):
2807 """
2808 test file.keyvalue when key_values not a dict
2809 """
2810 content = dedent(
2811 """\
2812 #PermitRootLogin prohibit-password
2813 #StrictMode yes
2814 """
2815 )
2816
2817 with salt.utils.files.fopen(fpath, "w+") as fp_:
2818 fp_.write(content)
2819
2820 ret = filestate.keyvalue(
2821 name=fpath,
2822 key_values=["PermiteRootLogin", "yes"],
2823 separator=" ",
2824 uncomment="#",
2825 key_ignore_case=True,
2826 )
2827
2828 self.assertEqual(
2829 ret["comment"],
2830 "file.keyvalue key and value not supplied and key_values is not a dictionary",
2831 )
2832 with salt.utils.files.fopen(fpath, "r") as fp_:
2833 f_contents = fp_.read()
2834 self.assertNotIn("PermitRootLogin yes", f_contents)
2835 self.assertIn("#StrictMode yes", f_contents)
2836
2837
2838 class TestFindKeepFiles(TestCase):
2839 @skipIf(salt.utils.platform.is_windows(), "Do not run on Windows")
2840 def test__find_keep_files_unix(self):
2841 keep = filestate._find_keep_files(
2842 "/test/parent_folder", ["/test/parent_folder/meh.txt"]
2843 )
2844 expected = [
2845 "/",
2846 "/test",
2847 "/test/parent_folder",
2848 "/test/parent_folder/meh.txt",
2849 ]
2850 actual = sorted(list(keep))
2851 assert actual == expected, actual
2852
2853 @skipIf(not salt.utils.platform.is_windows(), "Only run on Windows")
2854 def test__find_keep_files_win32(self):
2855 """
2856 Test _find_keep_files. The `_find_keep_files` function is only called by
2857 _clean_dir, so case doesn't matter. Should return all lower case.
2858 """
2859 keep = filestate._find_keep_files(
2860 "c:\\test\\parent_folder",
2861 [
2862 "C:\\test\\parent_folder\\meh-1.txt",
2863 "C:\\Test\\Parent_folder\\Meh-2.txt",
2864 ],
2865 )
2866 expected = [
2867 "c:\\",
2868 "c:\\test",
2869 "c:\\test\\parent_folder",
2870 "c:\\test\\parent_folder\\meh-1.txt",
2871 "c:\\test\\parent_folder\\meh-2.txt",
2872 ]
2873 actual = sorted(list(keep))
2874 self.assertListEqual(actual, expected)
2875
2876
2877 class TestFileTidied(TestCase):
2878 def setUp(self):
2879 setattr(filestate, "__opts__", {})
2880 setattr(filestate, "__salt__", {})
2881
2882 def tearDown(self):
2883 delattr(filestate, "__opts__")
2884 delattr(filestate, "__salt__")
2885
2886 def test__tidied(self):
2887 name = os.sep + "test"
2888 if salt.utils.platform.is_windows():
2889 name = "c:" + name
2890 walker = [
2891 (os.path.join("test", "test1"), [], ["file1"]),
2892 (os.path.join("test", "test2", "test3"), [], []),
2893 (os.path.join("test", "test2"), ["test3"], ["file2"]),
2894 ("test", ["test1", "test2"], ["file3"]),
2895 ]
2896 today_delta = datetime.today() - datetime.utcfromtimestamp(0)
2897 remove = MagicMock(name="file.remove")
2898 with patch("os.walk", return_value=walker), patch(
2899 "os.path.islink", return_value=False
2900 ), patch("os.path.getatime", return_value=today_delta.total_seconds()), patch(
2901 "os.path.getsize", return_value=10
2902 ), patch.dict(
2903 filestate.__opts__, {"test": False}
2904 ), patch.dict(
2905 filestate.__salt__, {"file.remove": remove}
2906 ), patch(
2907 "os.path.isdir", return_value=True
2908 ):
2909 ret = filestate.tidied(name=name)
2910 exp = {
2911 "name": name,
2912 "changes": {
2913 "removed": [
2914 os.path.join("test", "test1", "file1"),
2915 os.path.join("test", "test2", "file2"),
2916 os.path.join("test", "file3"),
2917 ]
2918 },
2919 "result": True,
2920 "comment": "Removed 3 files or directories from directory {}".format(name),
2921 }
2922 self.assertDictEqual(exp, ret)
2923 assert remove.call_count == 3
2924
2925 remove.reset_mock()
2926 with patch("os.walk", return_value=walker), patch(
2927 "os.path.islink", return_value=False
2928 ), patch("os.path.getatime", return_value=today_delta.total_seconds()), patch(
2929 "os.path.getsize", return_value=10
2930 ), patch.dict(
2931 filestate.__opts__, {"test": False}
2932 ), patch.dict(
2933 filestate.__salt__, {"file.remove": remove}
2934 ), patch(
2935 "os.path.isdir", return_value=True
2936 ):
2937 ret = filestate.tidied(name=name, rmdirs=True)
2938 exp = {
2939 "name": name,
2940 "changes": {
2941 "removed": [
2942 os.path.join("test", "test1", "file1"),
2943 os.path.join("test", "test2", "file2"),
2944 os.path.join("test", "test2", "test3"),
2945 os.path.join("test", "file3"),
2946 os.path.join("test", "test1"),
2947 os.path.join("test", "test2"),
2948 ]
2949 },
2950 "result": True,
2951 "comment": "Removed 6 files or directories from directory {}".format(name),
2952 }
2953 self.assertDictEqual(exp, ret)
2954 assert remove.call_count == 6
2955
2956 def test__bad_input(self):
2957 exp = {
2958 "name": "test/",
2959 "changes": {},
2960 "result": False,
2961 "comment": "Specified file test/ is not an absolute path",
2962 }
2963 assert filestate.tidied(name="test/") == exp
2964 exp = {
2965 "name": "/bad-directory-name/",
2966 "changes": {},
2967 "result": False,
2968 "comment": "/bad-directory-name/ does not exist or is not a directory.",
2969 }
2970 assert filestate.tidied(name="/bad-directory-name/") == exp
2971
2972
2973 class TestFilePrivateFunctions(TestCase, LoaderModuleMockMixin):
2974 def setup_loader_modules(self):
2975 return {filestate: {"__salt__": {"file.stats": filemod.stats}}}
2976
2977 @destructiveTest
2978 @skipIf(salt.utils.platform.is_windows(), "File modes do not exist on windows")
2979 def test__check_directory(self):
2980 """
2981 Test the _check_directory function
2982 Make sure that recursive file permission checks return correctly
2983 """
2984 # set file permissions
2985 # Run _check_directory function
2986 # Verify that it returns correctly
2987 # Delete tmp directory structure
2988 root_tmp_dir = os.path.join(RUNTIME_VARS.TMP, "test__check_dir")
2989 expected_mode = 0o770
2990 changed_mode = 0o755
2991 depth = 3
2992 try:
2993
2994 def create_files(tmp_dir):
2995 for f in range(depth):
2996 path = os.path.join(tmp_dir, "file_{:03}.txt".format(f))
2997 with salt.utils.files.fopen(path, "w+"):
2998 os.chmod(path, expected_mode)
2999
3000 # Create tmp directory structure
3001 os.mkdir(root_tmp_dir)
3002 os.chmod(root_tmp_dir, expected_mode)
3003 create_files(root_tmp_dir)
3004
3005 for d in range(depth):
3006 dir_name = os.path.join(root_tmp_dir, "dir{:03}".format(d))
3007 os.mkdir(dir_name)
3008 os.chmod(dir_name, expected_mode)
3009 create_files(dir_name)
3010 for s in range(depth):
3011 sub_dir_name = os.path.join(dir_name, "dir{:03}".format(s))
3012 os.mkdir(sub_dir_name)
3013 os.chmod(sub_dir_name, expected_mode)
3014 create_files(sub_dir_name)
3015 # Symlinks on linux systems always have 0o777 permissions.
3016 # Ensure we are not treating them as modified files.
3017 target_dir = os.path.join(root_tmp_dir, "link_target_dir")
3018 target_file = os.path.join(target_dir, "link_target_file")
3019 link_dir = os.path.join(root_tmp_dir, "link_dir")
3020 link_to_dir = os.path.join(link_dir, "link_to_dir")
3021 link_to_file = os.path.join(link_dir, "link_to_file")
3022
3023 os.mkdir(target_dir)
3024 os.mkdir(link_dir)
3025 with salt.utils.files.fopen(target_file, "w+"):
3026 pass
3027 os.symlink(target_dir, link_to_dir)
3028 os.symlink(target_file, link_to_file)
3029 for path in (target_dir, target_file, link_dir, link_to_dir, link_to_file):
3030 try:
3031 os.chmod(path, expected_mode, follow_symlinks=False)
3032 except (NotImplementedError, SystemError):
3033 os.chmod(path, expected_mode)
3034
3035 # Set some bad permissions
3036 changed_files = {
3037 os.path.join(root_tmp_dir, "file_000.txt"),
3038 os.path.join(root_tmp_dir, "dir002", "file_000.txt"),
3039 os.path.join(root_tmp_dir, "dir000", "dir001", "file_002.txt"),
3040 os.path.join(root_tmp_dir, "dir001", "dir002"),
3041 os.path.join(root_tmp_dir, "dir002", "dir000"),
3042 os.path.join(root_tmp_dir, "dir001"),
3043 }
3044 for c in changed_files:
3045 os.chmod(c, changed_mode)
3046
3047 ret = filestate._check_directory(
3048 root_tmp_dir,
3049 dir_mode=oct(expected_mode),
3050 file_mode=oct(expected_mode),
3051 recurse=["mode"],
3052 )
3053 self.assertSetEqual(changed_files, set(ret[-1].keys()))
3054
3055 finally:
3056 # Cleanup
3057 shutil.rmtree(root_tmp_dir)
3058
3059
3060 @skipIf(not salt.utils.platform.is_linux(), "Selinux only supported on linux")
3061 class TestSelinux(TestCase, LoaderModuleMockMixin):
3062 def setup_loader_modules(self):
3063 return {
3064 filestate: {
3065 "__env__": "base",
3066 "__salt__": {"file.manage_file": False},
3067 "__opts__": {"test": False, "cachedir": ""},
3068 "__instance_id__": "",
3069 "__low__": {},
3070 "__utils__": {},
3071 }
3072 }
3073
3074 def test_selinux_change(self):
3075 file_name = "/tmp/some-test-file"
3076 check_perms_result = [
3077 {
3078 "comment": "The file {} is set to be changed".format(file_name),
3079 "changes": {
3080 "selinux": {
3081 "New": "User: unconfined_u Type: lost_found_t",
3082 "Old": "User: system_u Type: user_tmp_t",
3083 }
3084 },
3085 "name": file_name,
3086 "result": True,
3087 },
3088 {"luser": "root", "lmode": "0644", "lgroup": "root"},
3089 ]
3090
3091 with patch.object(os.path, "exists", MagicMock(return_value=True)):
3092 with patch.dict(
3093 filestate.__salt__,
3094 {
3095 "file.source_list": MagicMock(return_value=[file_name, None]),
3096 "file.check_perms": MagicMock(return_value=check_perms_result),
3097 },
3098 ):
3099 ret = filestate.managed(
3100 file_name,
3101 selinux={"seuser": "unconfined_u", "setype": "user_tmp_t"},
3102 )
3103 self.assertEqual(True, ret["result"])