Coverage for src/types/yaml.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from copy import deepcopy 

5from io import TextIOWrapper 

6from re import fullmatch as regex_fullmatch, sub as regex_sub 

7from sys import maxsize 

8from typing import Any, Dict, List, Optional 

9 

10# Modules libraries 

11from yaml import dump as yaml_dump, Dumper as yaml_Dumper, load_all as yaml_load_all 

12from yaml import SafeLoader as yaml_SafeLoader, YAMLError as yaml_Error 

13from yaml.nodes import SequenceNode as yaml_SequenceNode 

14 

15# YAML class 

16class YAML: 

17 

18 # Data type 

19 Data = Dict[str, Any] 

20 

21 # Error type 

22 Error = yaml_Error 

23 

24 # Input type, pylint: disable=too-few-public-methods 

25 class Input: 

26 

27 # Constants 

28 TYPE_ARRAY: str = 'array' 

29 TYPE_BOOLEAN: str = 'boolean' 

30 TYPE_NUMBER: str = 'number' 

31 TYPE_STRING: str = 'string' 

32 

33 # Members 

34 default: Optional[float | int | str] = None 

35 description: Optional[str] = None 

36 options: Optional[List[str]] = None 

37 regex: Optional[str] = None 

38 value: Optional[float | int | str] = None 

39 value_type: str = TYPE_STRING 

40 

41 # Set description 

42 def set_description(self, description: str) -> None: 

43 self.description = description 

44 

45 # Set type 

46 def set_type(self, value: str) -> None: 

47 if value not in [ 

48 # YAML.Input.TYPE_ARRAY, 

49 YAML.Input.TYPE_BOOLEAN, 

50 YAML.Input.TYPE_NUMBER, 

51 YAML.Input.TYPE_STRING, 

52 ]: 

53 raise TypeError(f'\'{value}\': Input type unknown value') 

54 self.value_type = value 

55 

56 # Set options 

57 def set_options(self, options: List[Any]) -> None: 

58 if not options: 

59 raise ValueError('Options cannot be empty for input') 

60 if self.value_type not in [YAML.Input.TYPE_NUMBER, YAML.Input.TYPE_STRING]: 

61 raise TypeError( 

62 f'\'{self.value_type}\': Options can only be used with string and number inputs' 

63 ) 

64 for option in options: 

65 if self.value_type == YAML.Input.TYPE_NUMBER and not isinstance( 

66 option, (float, int)): 

67 raise TypeError( 

68 f'\'{option}\': Options can only be numbers for inputs') 

69 if self.value_type in YAML.Input.TYPE_STRING and not isinstance( 

70 option, str): 

71 raise TypeError( 

72 f'\'{option}\': Options can only be strings for inputs') 

73 if self.regex and not regex_fullmatch(self.regex, option): 

74 raise ValueError(f'\'{option}\': Option value does not match' 

75 f' required RegEx pattern \'{self.regex}\'') 

76 self.options = options[:] 

77 

78 # Set regex 

79 def set_regex( 

80 self, 

81 regex: str, 

82 ) -> None: 

83 if self.value_type not in [YAML.Input.TYPE_STRING]: 

84 raise TypeError( 

85 f'\'{self.value_type}\': RegEx validation can only be used with string inputs' 

86 ) 

87 self.regex = regex 

88 

89 # Set default 

90 def set_default( 

91 self, 

92 default: bool | float | int | str, 

93 ) -> None: 

94 self.default = self.parse_value(default) 

95 

96 # Validate input default 

97 if self.options and self.default not in self.options: 

98 raise ValueError( 

99 f'\'{self.default}\': Default value not allowed in options' 

100 f': {", ".join(self.options)})') 

101 if self.regex and not regex_fullmatch(self.regex, str(self.default)): 

102 raise ValueError(f'\'{self.default}\': Default value does not match' 

103 f' required RegEx pattern \'{self.regex}\'') 

104 

105 # Set value 

106 def set_value( 

107 self, 

108 value: bool | float | int | str, 

109 ) -> None: 

110 self.value = self.parse_value(value) 

111 

112 # Validate input value 

113 if self.options and self.value not in self.options: 

114 raise ValueError( 

115 f'\'{self.value}\': Value not allowed in options: {", ".join(self.options)})' 

116 ) 

117 if self.regex and not regex_fullmatch(self.regex, str(self.value)): 

118 raise ValueError(f'\'{self.value}\': Value does not match' 

119 f' required RegEx pattern \'{self.regex}\'') 

120 

121 # Parse value, pylint: disable=too-many-branches,too-many-return-statements 

122 def parse_value( 

123 self, 

124 value: bool | float | int | str, 

125 ) -> bool | float | int | str: 

126 

127 # Set boolean value 

128 if self.value_type == YAML.Input.TYPE_BOOLEAN: 

129 if isinstance(value, bool): 

130 return str(value).lower() 

131 if isinstance(value, int) and value in [1]: 

132 return 'true' 

133 if isinstance(value, int) and value in [0]: 

134 return 'false' 

135 if isinstance(value, str) and value.lower() in ['true', '1']: 

136 return 'true' 

137 if isinstance(value, str) and value.lower() in ['false', '0']: 

138 return 'false' 

139 raise ValueError(f'\'{value}\': Value is not a boolean') 

140 

141 # Set number value 

142 if self.value_type == YAML.Input.TYPE_NUMBER: 

143 if isinstance(value, bool): 

144 raise ValueError(f'\'{value}\': Value is not a number') 

145 if isinstance(value, (float, int)): 

146 return value 

147 if isinstance(value, str): 

148 try: 

149 if '.' in value: 

150 return float(value) 

151 return int(value) 

152 except ValueError: 

153 raise ValueError(f"'{value}': Value is not a number") # pylint: disable=raise-missing-from 

154 

155 # Set string value 

156 if self.value_type == YAML.Input.TYPE_STRING: 

157 if isinstance(value, (float, int, list)): 

158 return str(value) 

159 if isinstance(value, str): 

160 return value 

161 

162 # Invalid value type 

163 raise TypeError(f'\'{value}\': Value type \'{self.value_type}\' supported' 

164 ) # pragma: no cover 

165 

166 # Reference class 

167 class Reference: 

168 

169 # Constants 

170 NESTED_LIMIT: int = 10 

171 TAG: str = '!reference' 

172 

173 # Constructor 

174 def __init__(self, values: List[str]): 

175 self._values = values 

176 

177 # Values 

178 @property 

179 def values(self) -> List[str]: 

180 return self._values 

181 

182 # Add constructor 

183 @staticmethod 

184 def add_constructor( 

185 loader: yaml_SafeLoader, 

186 node: yaml_SequenceNode, 

187 ) -> 'YAML.Reference': 

188 return YAML.Reference(loader.construct_sequence(node)) 

189 

190 # Resolve, pylint: disable=too-many-branches 

191 @staticmethod 

192 def resolve(data: Any, node: Any) -> bool: 

193 

194 # Variables 

195 changed: bool = False 

196 

197 # Dictionnaries 

198 if isinstance(node, dict): 

199 for key in node.keys(): 

200 if isinstance(node[key], YAML.Reference): 

201 references = list(node[key].values) 

202 referenced_node = data 

203 for reference in references: 

204 if reference in referenced_node: 

205 referenced_node = referenced_node[reference] 

206 else: # pragma: no cover 

207 referenced_node = None 

208 break 

209 changed = True 

210 if referenced_node: 

211 node[key] = deepcopy(referenced_node) 

212 else: # pragma: no cover 

213 del node[key] 

214 elif YAML.Reference.resolve(data, node[key]): 

215 changed = True 

216 

217 # Lists 

218 elif isinstance(node, list): 

219 for index in reversed(range(len(node))): 

220 if isinstance(node[index], YAML.Reference): 

221 references = list(node[index].values) 

222 referenced_node = data 

223 for reference in references: 

224 if reference in referenced_node: 

225 referenced_node = referenced_node[reference] 

226 else: # pragma: no cover 

227 referenced_node = None 

228 break 

229 changed = True 

230 if referenced_node: 

231 node[index:index + 1] = deepcopy(referenced_node) 

232 else: # pragma: no cover 

233 del node[index] 

234 elif YAML.Reference.resolve(data, node[index]): 

235 changed = True # pragma: no cover 

236 

237 # Standard types 

238 # else: 

239 # pass 

240 

241 # Result 

242 return changed 

243 

244 # Parse spec inputs 

245 @staticmethod 

246 def parse_spec_inputs( 

247 spec: Data, 

248 file_path: str, 

249 file_inputs: Dict[str, str], 

250 ) -> Dict[str, Input]: 

251 

252 # Variables 

253 inputs: Dict[str, YAML.Input] = {} 

254 

255 # Validate YAML spec syntax 

256 if 'spec' not in spec or 'inputs' not in spec['spec']: 

257 return inputs 

258 

259 # Parse YAML spec inputs 

260 for key, item in spec['spec']['inputs'].items(): 

261 

262 # Prepare input 

263 inputs[key] = YAML.Input() 

264 if isinstance(item, dict): 

265 if 'description' in item and isinstance(item['description'], str): 

266 inputs[key].set_description(item['description']) 

267 if 'type' in item and isinstance(item['type'], str): 

268 inputs[key].set_type(item['type']) 

269 if 'regex' in item and isinstance(item['regex'], str): 

270 inputs[key].set_regex(item['regex']) 

271 if 'options' in item and isinstance(item['options'], list): 

272 inputs[key].set_options(list(item['options'])) 

273 if 'default' in item and isinstance(item['default'], 

274 (bool, float, int, str)): 

275 inputs[key].set_default(item['default']) 

276 

277 # Prepare input value 

278 if isinstance(item, 

279 dict) and key in file_inputs and file_inputs[key] is not None: 

280 inputs[key].set_value(file_inputs[key]) 

281 elif inputs[key].default is not None: 

282 inputs[key].set_value(inputs[key].default) # type: ignore[arg-type] 

283 elif (options := inputs[key].options) is not None: 

284 inputs[key].set_value(options[0]) 

285 

286 # Validate input value 

287 if inputs[key].value is None: 

288 raise SyntaxError(f'Missing value for input \'{key}\' in \'{file_path}\'') 

289 

290 # Result 

291 return inputs 

292 

293 # Evaluate string 

294 @staticmethod 

295 def evaluate_spec_field(field: str, inputs: Dict[str, Input]) -> str: 

296 

297 # Variables 

298 result: str = field 

299 

300 # Evaluate inputs 

301 for variable, node in inputs.items(): 

302 assert node.value is not None 

303 result = regex_sub(fr'\$\[\[\s*inputs\.{variable}\s*\]\]', str(node.value), 

304 result) 

305 

306 # Result 

307 return result 

308 

309 # Evaluate spec 

310 @staticmethod 

311 def evaluate_spec(data: Data, inputs: Dict[str, Input]) -> Data: 

312 

313 # Recurse on keys 

314 if isinstance(data, dict): 

315 return { 

316 YAML.evaluate_spec_field(key, inputs): YAML.evaluate_spec(value, inputs) 

317 for key, value in data.items() 

318 } 

319 

320 # Recurse on items 

321 if isinstance(data, list): 

322 return [YAML.evaluate_spec(value, inputs) for value in data] 

323 

324 # Evaluate string items 

325 if isinstance(data, str): 

326 return YAML.evaluate_spec_field(data, inputs) 

327 

328 # Fallback 

329 return data # pragma: no cover 

330 

331 # Load, pylint: disable=too-many-branches 

332 @staticmethod 

333 def load( 

334 stream: TextIOWrapper, 

335 inputs: Dict[str, str], 

336 configure: bool, 

337 ) -> Optional[Data]: 

338 

339 # Variables 

340 data: Optional[YAML.Data] 

341 documents: List[YAML.Data] 

342 file_inputs: Dict[str, YAML.Input] 

343 

344 # Prepare loader class 

345 loader = yaml_SafeLoader 

346 loader.add_constructor(YAML.Reference.TAG, YAML.Reference.add_constructor) 

347 

348 # Load YAML documents 

349 documents = list(yaml_load_all(stream, Loader=loader)) 

350 

351 # Empty YAML data 

352 if not documents: 

353 data = None 

354 

355 # Acquire YAML configuration data 

356 elif len(documents) == 1: 

357 data = documents[0] 

358 

359 # Acquire YAML component data 

360 elif len(documents) == 2: 

361 

362 # Parse spec inputs 

363 file_inputs = YAML.parse_spec_inputs( 

364 documents[0], 

365 stream.name, 

366 file_inputs=inputs, 

367 ) 

368 

369 # Configure spec inputs 

370 if configure: 

371 from ..features.menus import MenusFeature # pylint: disable=cyclic-import,import-outside-toplevel 

372 configure_inputs: Dict[str, Dict[str, bool | List[str] | str]] = {} 

373 for key, node in file_inputs.items(): 

374 configure_inputs[key] = {} 

375 configure_inputs[key]['help'] = node.description \ 

376 if node.description else 'Configure variable value' 

377 if node.value_type == YAML.Input.TYPE_ARRAY: # pragma: no cover 

378 raise TypeError(f'\'{key}\': Input type unknown value') 

379 if node.value_type == YAML.Input.TYPE_BOOLEAN: 

380 configure_inputs[key]['type'] = 'boolean' 

381 elif node.options: 

382 configure_inputs[key]['type'] = 'choice' 

383 configure_inputs[key]['values'] = node.options 

384 elif node.value_type == YAML.Input.TYPE_NUMBER: 

385 configure_inputs[key]['type'] = 'number' 

386 elif node.value_type == YAML.Input.TYPE_STRING: 

387 configure_inputs[key]['type'] = 'input' 

388 configure_inputs[key]['set'] = key in inputs 

389 if key in inputs: 

390 configure_inputs[key]['default'] = str(node.value) 

391 elif str(node.default): 

392 configure_inputs[key]['default'] = str(node.default) 

393 else: 

394 configure_inputs[key]['default'] = '' 

395 configured_variables: Dict[str, str] = \ 

396 MenusFeature.configure(configure_inputs) 

397 for key, value in configured_variables.items(): 

398 file_inputs[key].set_value(value) 

399 

400 # Evaluate spec inputs 

401 data = YAML.evaluate_spec( 

402 documents[1], 

403 inputs=file_inputs, 

404 ) 

405 

406 # Unsupport YAML data 

407 else: 

408 raise SyntaxError( 

409 f'Unsupported documents count ({len(documents)}) in \'{stream.name}\'') 

410 

411 # Result 

412 return data 

413 

414 # Dump 

415 @staticmethod 

416 def dump(data: Data) -> str: 

417 

418 # Dumper, pylint: disable=too-many-ancestors 

419 class Dumper(yaml_Dumper): 

420 pass 

421 

422 # Representer 

423 Dumper.add_representer( 

424 str, 

425 lambda dumper, data: dumper.represent_scalar( 

426 'tag:yaml.org,2002:str', 

427 data, 

428 style='|' if '\n' in data else None, 

429 ), 

430 ) 

431 

432 # Dump YAML data 

433 return str( 

434 yaml_dump( 

435 data, 

436 Dumper=Dumper, 

437 indent=2, 

438 sort_keys=False, 

439 width=maxsize, 

440 )) 

441 

442 # Resolve 

443 @staticmethod 

444 def resolve(data: Dict[Any, Any]) -> None: 

445 

446 # Resolve references 

447 for _ in range(YAML.Reference.NESTED_LIMIT): 

448 if not YAML.Reference.resolve(data, data): 

449 break 

450 else: # pragma: no cover 

451 # error 

452 pass