Coverage for src / dotbot / dispatcher.py: 87%

1import os 

2from argparse import Namespace 

3from typing import Any, Dict, List, Optional, Type 

4 

5from dotbot.context import Context 

6from dotbot.messenger import Messenger 

7from dotbot.plugin import Plugin 

8from dotbot.util.module import load_plugins 

9 

10# Before b5499c7dc5b300462f3ce1c2a3d9b7a76233b39b, Dispatcher auto-loaded all 

11# plugins, but after that change, plugins are passed in explicitly (and loaded 

12# in cli.py). There are some plugins that rely on the old Dispatcher behavior, 

13# so this is a workaround for implementing similar functionality: when 

14# Dispatcher is constructed without an explicit list of plugins, _all_plugins is 

15# used instead. 

16_all_plugins: List[Type[Plugin]] = [] # filled in by cli.py 

17 

18 

19class Dispatcher: 

20 def __init__( 

21 self, 

22 base_directory: str, 

23 only: Optional[List[str]] = None, 

24 skip: Optional[List[str]] = None, 

25 exit_on_failure: bool = False, # noqa: FBT001, FBT002 part of established public API 

26 options: Optional[Namespace] = None, 

27 plugins: Optional[List[Type[Plugin]]] = None, 

28 ): 

29 # if the caller wants no plugins, the caller needs to explicitly pass in 

30 # plugins=[] 

31 self._log = Messenger() 

32 self._setup_context(base_directory, options, plugins) 

33 if plugins is None: 

34 plugins = _all_plugins 

35 self._plugins = [plugin(self._context) for plugin in plugins] 

36 self._only = only 

37 self._skip = skip 

38 self._exit = exit_on_failure 

39 self._dry_run: bool = options is not None and bool(options.dry_run) 

40 

41 def _setup_context( 

42 self, base_directory: str, options: Optional[Namespace], plugins: Optional[List[Type[Plugin]]] 

43 ) -> None: 

44 path = os.path.abspath(os.path.expanduser(base_directory)) 

45 if not os.path.exists(path): 

46 msg = "Nonexistent base directory" 

47 raise DispatchError(msg) 

48 self._context = Context(path, options, plugins) 

49 

50 def dispatch(self, tasks: List[Dict[str, Any]]) -> bool: 

51 success = True 

52 for task in tasks: 

53 for action in task: 

54 if ( 

55 (self._only is not None and action not in self._only) 

56 or (self._skip is not None and action in self._skip) 

57 ) and action != "defaults": 

58 self._log.info(f"Skipping action {action}") 

59 continue 

60 handled = False 

61 if action == "defaults": 

62 self._context.set_defaults(task[action]) # replace, not update 

63 handled = True 

64 # keep going, let other plugins handle this if they want 

65 if action == "plugins": 

66 for plugin_path in task[action]: 

67 try: 

68 # load the new plugins and add them to the list of plugins 

69 # this mutates self._context._plugins; we don't add a setter method 

70 # to Context because we don't want plugins to call it 

71 new_plugins = load_plugins([plugin_path], self._context._plugins) # noqa: SLF001 

72 for plugin_class in new_plugins: 

73 self._plugins.append(plugin_class(self._context)) 

74 except Exception as err: # noqa: BLE001 

75 self._log.warning(f"Failed to load plugin '{plugin_path}'") 

76 self._log.debug(str(err)) 

77 success = False 

78 if not success: 

79 self._log.error("Some plugins could not be loaded") 

80 if self._exit: 

81 self._log.error("Action plugins failed") 

82 return False 

83 handled = True 

84 # keep going, let other plugins handle this if they want 

85 for plugin in self._plugins: 

86 if plugin.can_handle(action): 

87 if self._dry_run and not plugin.supports_dry_run: 

88 self._log.action(f"Skipping dry-run-unaware plugin {plugin.__class__.__name__}") 

89 handled = True 

90 continue 

91 try: 

92 local_success = plugin.handle(action, task[action]) 

93 if not local_success and self._exit: 

94 # The action has failed, exit 

95 self._log.error(f"Action {action} failed") 

96 return False 

97 success &= local_success 

98 handled = True 

99 except Exception as err: # noqa: BLE001 

100 self._log.error(f"An error was encountered while executing action {action}") 

101 self._log.debug(str(err)) 

102 if self._exit: 

103 # There was an exception, exit 

104 return False 

105 if not handled: 

106 success = False 

107 self._log.error(f"Action {action} not handled") 

108 if self._exit: 

109 # Invalid action exit 

110 return False 

111 return success 

112 

113 

114class DispatchError(Exception): 

115 pass