Coverage for src / dotbot / plugins / shell.py: 97%

1from typing import Any, Dict 

2 

3from dotbot.plugin import Plugin 

4from dotbot.util.common import shell_command 

5 

6 

7class Shell(Plugin): 

8 """ 

9 Run arbitrary shell commands. 

10 """ 

11 

12 supports_dry_run = True 

13 

14 _directive = "shell" 

15 _has_shown_override_message = False 

16 

17 def can_handle(self, directive: str) -> bool: 

18 return directive == self._directive 

19 

20 def handle(self, directive: str, data: Any) -> bool: 

21 if directive != self._directive: 

22 msg = f"Shell cannot handle directive {directive}" 

23 raise ValueError(msg) 

24 return self._process_commands(data) 

25 

26 def _process_commands(self, data: Any) -> bool: 

27 success = True 

28 defaults = self._context.defaults().get("shell", {}) 

29 options = self._get_option_overrides() 

30 for item in data: 

31 stdin = defaults.get("stdin", False) 

32 stdout = defaults.get("stdout", False) 

33 stderr = defaults.get("stderr", False) 

34 quiet = defaults.get("quiet", False) 

35 if isinstance(item, dict): 

36 cmd = item["command"] 

37 msg = item.get("description", None) 

38 stdin = item.get("stdin", stdin) 

39 stdout = item.get("stdout", stdout) 

40 stderr = item.get("stderr", stderr) 

41 quiet = item.get("quiet", quiet) 

42 elif isinstance(item, list): 

43 cmd = item[0] 

44 msg = item[1] if len(item) > 1 else None 

45 else: 

46 cmd = item 

47 msg = None 

48 prefix = "Would run command " if self._context.dry_run() else "" 

49 if quiet: 

50 if msg is not None: 

51 self._log.info(f"{prefix}{msg}") 

52 # if quiet and no msg, show nothing 

53 elif msg is None: 

54 self._log.action(f"{prefix}{cmd}") 

55 else: 

56 self._log.action(f"{prefix}{msg} [{cmd}]") 

57 if self._context.dry_run(): 

58 continue 

59 stdout = options.get("stdout", stdout) 

60 stderr = options.get("stderr", stderr) 

61 ret = shell_command( 

62 cmd, 

63 cwd=self._context.base_directory(), 

64 enable_stdin=stdin, 

65 enable_stdout=stdout, 

66 enable_stderr=stderr, 

67 ) 

68 if ret != 0: 

69 success = False 

70 self._log.warning(f"Command [{cmd}] failed") 

71 if success: 

72 self._log.info("All commands have been executed") 

73 else: 

74 self._log.error("Some commands were not successfully executed") 

75 return success 

76 

77 def _get_option_overrides(self) -> Dict[str, bool]: 

78 ret = {} 

79 options = self._context.options() 

80 if options.verbose > 1: 

81 ret["stderr"] = True 

82 ret["stdout"] = True 

83 if not self._has_shown_override_message: 

84 self._log.debug("Shell: Found cli option to force show stderr and stdout.") 

85 self._has_shown_override_message = True 

86 return ret