Coverage for src / template / __init__.py: 90%

1import copy 

2import json 

3import os 

4import sys 

5 

6import yaml 

7 

8from .. import constants as c 

9from . import schema 

10from .methods import directory, git 

11 

12 

13class Template: 

14 def __init__(self, **kwargs): 

15 self.path = kwargs.get("template_path") 

16 self.filename = kwargs.get("template_filename", c.TEMPLATE_FILENAME) 

17 self.branch = kwargs.get("branch") 

18 self.workpath = kwargs.get("workpath") 

19 self.includes = None 

20 self.data = None 

21 

22 def __repr__(self): 

23 # no cover: start 

24 return json.dumps( 

25 { 

26 "path": self.path, 

27 "filename": self.filename, 

28 "branch": self.branch, 

29 "workpath": self.workpath, 

30 }, 

31 ) 

32 # no cover: stop 

33 

34 @property 

35 def actions(self): 

36 if self.data is None: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 return [] 

38 # pylint: disable=unsubscriptable-object 

39 return self.data["actions"] 

40 

41 @property 

42 def questions(self): 

43 if self.data is None: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true

44 return {} 

45 # pylint: disable=unsubscriptable-object 

46 return self.data["questions"] 

47 

48 @property 

49 def children(self): 

50 """ 

51 Return only the includes for this template 

52 """ 

53 if not self.data: 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 return 

55 

56 # By reversing the list, we'll be building the list 

57 # of questions in the right order 

58 items = reversed(self.data.get("includes", [])) 

59 for item in items: 

60 path = item["include"] 

61 filename = item.get("filename") 

62 branch = item.get("branch") 

63 workpath = item.get("workpath") 

64 

65 yield Template(template_path=path, template_filename=filename, branch=branch, workpath=workpath) 

66 

67 

68def find(tpl): 

69 for func in [directory.find, git.find]: 69 ↛ 73line 69 didn't jump to line 73 because the loop on line 69 didn't complete

70 tpl.workpath = func(tpl) 

71 if tpl.workpath: 

72 return tpl 

73 sys.exit(f'error: no "{tpl.filename}" file found in "{tpl.path}"') 

74 

75 

76def _load(tpl): 

77 path = os.path.join(tpl.workpath, tpl.filename) 

78 try: 

79 with open(path, encoding="UTF-8") as fd: 

80 data = yaml.safe_load(fd) or {} 

81 except Exception as err: 

82 sys.exit(f"error: failed to open '{tpl.workpath}': {err}") 

83 return data 

84 

85 

86def _reduce(data): 

87 # Reduce to relevant sections. 

88 # This allows a user to add whatever he wants to a scaffold.yml file 

89 retval = { 

90 "actions": data.get("actions", []), 

91 "answers": data.get("answers", {}), 

92 "includes": data.get("includes", []), 

93 "jinja2": data.get("jinja2", {}), 

94 "questions": data.get("questions", []), 

95 } 

96 return retval 

97 

98 

99def _validate(data): 

100 # TODO: merge with answers into utils 

101 yaml_schema = yaml.safe_load(schema.SCHEMA) 

102 validator = schema.LocalValidator(yaml_schema) 

103 

104 if not validator.validate(data): 

105 locations = str(json.dumps(validator.errors, indent=2)) 

106 raise SystemExit(f"error: YAML schema validation error. Location:\n{locations}") from None 

107 

108 return validator.normalized(data) 

109 

110 

111def load(tpl, root=None): 

112 if root is None: 

113 root = tpl 

114 

115 data = _load(tpl) 

116 data = _reduce(data) 

117 data = _validate(data) 

118 tpl.data = data 

119 

120 # Keep a list of all encountered children 

121 if root.includes is None: 

122 root.includes = [] 

123 root.includes += [tpl] 

124 

125 # In order to load the questions, we need to recurse into 

126 # all files and build the questions list by priority, the 

127 # inclusion order 

128 return _recurse(tpl, root) 

129 

130 

131def _recurse(tpl, root): 

132 for other in tpl.children: 

133 other = find(other) 

134 other = load(other, root) 

135 tpl = _merge(tpl, other) 

136 return tpl 

137 

138 

139def _merge(tpl, other): 

140 # Merge questions. 

141 # Since the "name" must be unique, we use it as a key 

142 # other has priority on self 

143 result = {i["name"]: i for i in other.data["questions"]} 

144 for question in tpl.data["questions"]: 

145 result[question["name"]] = question 

146 tpl.data["questions"] = [v for k, v in result.items()] 

147 

148 # Sort by order if present 

149 tpl.data["questions"] = sorted(tpl.data["questions"], key=lambda item: item.get("order", 0)) 

150 

151 return tpl