Coverage for yasmon/processor.py: 100%

1from yasmon.callbacks import AbstractCallback 

2from yasmon.callbacks import CallbackDict 

3from yasmon.callbacks import ShellCallback, LoggerCallback, MailCallback 

4from yasmon.callbacks import CallbackSyntaxError 

5from yasmon.callbacks import CallbackNotImplementedError 

6from yasmon.tasks import TaskList, WatchfilesTask, TaskNotImplementedError 

7from yasmon.loggers import LoggerSyntaxError, AbstractLogger 

8from yasmon.loggers import StdErrLogger, FileLogger, JournalLogger 

9 

10from loguru import logger 

11import yaml 

12 

13 

14class YAMLSyntaxError(Exception): 

15 """ 

16 Raised when on general YAML syntax errors. 

17 """ 

18 

19 def __init__(self, message="yaml syntax error"): 

20 self.message = message 

21 super().__init__(self.message) 

22 

23 

24class YAMLProcessor: 

25 def __init__(self): 

26 self.data = None 

27 

28 def load_file(self, filename: str): 

29 try: 

30 fh = open(filename, "r") 

31 logger.info(f'using config file {filename}') 

32 except FileNotFoundError as err: 

33 logger.error(f"YAML file {filename} not found") 

34 raise err 

35 except PermissionError as err: 

36 logger.error( 

37 f"{err.__class__.__name__} while opening {filename} {err}") 

38 raise err 

39 except OSError as err: 

40 logger.error( 

41 f"{err.__class__.__name__} while opening {filename} {err}") 

42 raise err 

43 except Exception as err: 

44 logger.error( 

45 f"{err.__class__.__name__} while opening {filename} {err}") 

46 raise err 

47 else: 

48 try: 

49 self.data = yaml.safe_load(fh) 

50 except yaml.YAMLError as err: 

51 raise YAMLSyntaxError(err) 

52 finally: 

53 fh.close() 

54 

55 if self.data is None: 

56 raise YAMLSyntaxError('config file is empty') 

57 

58 def add_loggers(self) -> list[AbstractLogger]: 

59 """ 

60 Add defined loggers. 

61 

62 :return: list of added loggers or empty list 

63 :rtype: list[AbstractLogger] 

64 """ 

65 logger.info('processing loggers...') 

66 loggers: list[AbstractLogger] = [] 

67 

68 for key in self.data: 

69 data = self.data[key] 

70 yaml_data = yaml.dump(data) 

71 try: 

72 match key: 

73 case 'log_stderr': 

74 instance = StdErrLogger.from_yaml(yaml_data) 

75 loggers.append(instance) 

76 case 'log_file': 

77 instance = FileLogger.from_yaml(yaml_data) 

78 loggers.append(instance) 

79 case 'log_journal': 

80 instance = JournalLogger.from_yaml(yaml_data) 

81 loggers.append(instance) 

82 except LoggerSyntaxError: 

83 raise 

84 

85 return loggers 

86 

87 def load_document(self, document: str): 

88 try: 

89 self.data = yaml.safe_load(document) 

90 logger.info(f'config:\n{document}') 

91 except yaml.YAMLError as err: 

92 raise YAMLSyntaxError(err) 

93 

94 if self.data is None: 

95 raise YAMLSyntaxError('config document is empty') 

96 

97 def get_tasks(self, callbacks: CallbackDict): 

98 logger.debug('processing tasks...') 

99 if 'tasks' not in self.data: 

100 raise YAMLSyntaxError('tasks not defined') 

101 

102 if not self.data['tasks']: 

103 logger.warning('no tasks defined') 

104 return TaskList() 

105 

106 tasks = self.data['tasks'] 

107 

108 if type(tasks) is not dict: 

109 raise YAMLSyntaxError('tasks must be a dictionary') 

110 

111 taskslist = TaskList() 

112 for task in tasks: 

113 taskdata = tasks[task] 

114 if type(taskdata) is not dict: 

115 raise YAMLSyntaxError(f'{task} task data must be a dictionary') 

116 

117 taskdata_yaml = yaml.dump(taskdata) 

118 

119 if 'callbacks' not in taskdata: 

120 raise YAMLSyntaxError(f'{task} task data must include' 

121 ' callbacks list') 

122 

123 task_callbacks: list[AbstractCallback] = [ 

124 callbacks[c] for c in taskdata["callbacks"] 

125 if c in taskdata["callbacks"] 

126 ] 

127 

128 match taskdata['type']: 

129 case 'watchfiles': 

130 taskslist.append(WatchfilesTask.from_yaml(task, 

131 taskdata_yaml, 

132 task_callbacks)) 

133 case _: 

134 raise TaskNotImplementedError( 

135 f'task type {taskdata["type"]}' 

136 ' not implement') 

137 

138 logger.debug('done processing tasks') 

139 return taskslist 

140 

141 def get_callbacks(self): 

142 logger.debug('processing callbacks...') 

143 if 'callbacks' not in self.data: 

144 raise YAMLSyntaxError('callbacks not defined') 

145 

146 if not self.data['callbacks']: 

147 logger.warning('no callbacks defined') 

148 return 

149 

150 callbacks = self.data['callbacks'] 

151 

152 if type(self.data['callbacks']) is not dict: 

153 raise YAMLSyntaxError('callbacks must be a dictionary') 

154 

155 callbacksdict = CallbackDict() 

156 for callback in callbacks: 

157 callbackdata = self.data['callbacks'].get(callback) 

158 if type(callbackdata) is not dict: 

159 raise YAMLSyntaxError(f'{callback} callback data must' 

160 ' be a dictionary') 

161 

162 try: 

163 match callbackdata['type']: 

164 case 'shell': 

165 callbacksdict[callback] = ShellCallback.from_yaml( 

166 callback, yaml.dump(callbackdata)) 

167 case 'logger': 

168 callbacksdict[callback] = LoggerCallback.from_yaml( 

169 callback, yaml.dump(callbackdata)) 

170 case 'mail': 

171 callbacksdict[callback] = MailCallback.from_yaml( 

172 callback, yaml.dump(callbackdata)) 

173 case _: 

174 raise CallbackNotImplementedError( 

175 f'callback type {callbackdata["type"]} ' 

176 'not implement') 

177 except CallbackSyntaxError as err: 

178 logger.error(f'error while processing callbacks: {err}. ' 

179 'Exiting!') 

180 raise err 

181 

182 logger.debug('done processing callbacks') 

183 return callbacksdict