Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/modules/script.py: 0%

1import io 

2import typing as t 

3from viur.core.bones import * 

4from viur.core.prototypes.tree import Tree, TreeSkel, SkelType 

5from viur.core.modules.file import File 

6from viur.core import db, conf, current, skeleton, tasks, errors 

7from viur.core.decorators import exposed 

8from viur.core.i18n import translate 

9import zipfile 

10 

11 

12class BaseScriptAbstractSkel(TreeSkel): 

13 path = StringBone( 

14 descr="Path", 

15 readOnly=True, 

16 unique=UniqueValue(UniqueLockMethod.SameValue, True, "This path is already taken!") 

17 ) 

18 

19 @classmethod 

20 def fromClient(cls, skel, data, *args, **kwargs): 

21 # Set script name when provided, so that the path can be regenerated 

22 if name := data.get("name"): 

23 skel["name"] = name 

24 conf.main_app.script.update_path(skel) 

25 

26 ret = super().fromClient(skel, data, *args, **kwargs) 

27 

28 if not ret: 

29 # in case the path failed because the unique value is already taken, rewrite the error for name field 

30 for error in skel.errors: 

31 if error.severity == ReadFromClientErrorSeverity.Invalid and error.fieldPath == ["path"]: 

32 error.fieldPath = ["name"] 

33 break 

34 

35 return ret 

36 

37 

38class ScriptNodeSkel(BaseScriptAbstractSkel): 

39 kindName = "viur-script-node" 

40 

41 rootNode = BooleanBone( 

42 descr="Is root node?", 

43 defaultValue=False, 

44 ) 

45 

46 plugin = BooleanBone( 

47 descr="Is plugin?", 

48 defaultValue=False 

49 ) 

50 

51 name = StringBone( 

52 descr="Folder", 

53 required=True, 

54 vfunc=lambda value: None if File.is_valid_filename(value) else "Foldername is invalid" 

55 ) 

56 

57 

58class ScriptLeafSkel(BaseScriptAbstractSkel): 

59 kindName = "viur-script-leaf" 

60 

61 name = StringBone( 

62 descr="Filename", 

63 required=True, 

64 vfunc=lambda value: 

65 None if File.is_valid_filename(value) and value.endswith(".py") and value.removesuffix(".py") 

66 else "Filename is invalid or doesn't have a '.py'-suffix", 

67 ) 

68 

69 script = PythonBone( 

70 descr="Code", 

71 ) 

72 

73 access = SelectBone( 

74 descr="Required access rights to run this Script", 

75 values=lambda: { 

76 right: translate(f"viur.core.modules.user.accessright.{right}", defaultText=right) 

77 for right in sorted(conf.user.access_rights) 

78 }, 

79 multiple=True, 

80 ) 

81 

82 

83class Script(Tree): 

84 """ 

85 Script is a system module used to serve a filesystem for scripts used by ViUR Scriptor and ViUR CLI. 

86 """ 

87 

88 leafSkelCls = ScriptLeafSkel 

89 nodeSkelCls = ScriptNodeSkel 

90 

91 roles = { 

92 "admin": "*", 

93 } 

94 

95 def adminInfo(self): 

96 return conf.script_admin_info or {} 

97 

98 def getAvailableRootNodes(self): 

99 if not current.user.get(): 

100 return [] 

101 

102 return [{ 

103 "name": "Scripts", 

104 "key": self.rootnodeSkel(ensure=True)["key"], 

105 }] 

106 

107 @exposed 

108 def view(self, skelType: SkelType, key: db.KeyType, *args, **kwargs) -> t.Any: 

109 try: 

110 return super().view(skelType, key, *args, **kwargs) 

111 except errors.NotFound: 

112 # When key is not found, try to interpret key as path 

113 if skel := self.viewSkel(skelType).all().mergeExternalFilter({"path": key}).getSkel(): 

114 return super().view(skelType, skel["key"], *args, **kwargs) 

115 

116 raise 

117 

118 def onEdit(self, skelType, skel): 

119 self.update_path(skel) 

120 super().onEdit(skelType, skel) 

121 

122 def onEdited(self, skelType, skel): 

123 old_path = skel["path"] 

124 self.update_path(skel) 

125 if skel["path"] != old_path: 

126 skel.patch({"path": skel["path"]}) 

127 

128 if skelType == "node": 

129 self.update_path_recursive("node", skel["path"], skel["key"]) 

130 self.update_path_recursive("leaf", skel["path"], skel["key"]) 

131 

132 super().onEdited(skelType, skel) 

133 

134 @tasks.CallDeferred 

135 def update_path_recursive(self, skel_type, path, parent_key, cursor=None): 

136 """ 

137 Recursively updates all items under a given parent key. 

138 """ 

139 query = self.editSkel(skel_type).all().filter("parententry", parent_key) 

140 query.setCursor(cursor) 

141 

142 for skel in query.fetch(99): 

143 new_path = path + "/" + skel["name"] 

144 

145 # only update when path changed 

146 if new_path != skel["path"]: 

147 skel["path"] = new_path # self.onEdit() is NOT required, as it resolves the path again. 

148 skel.write() 

149 self.onEdited(skel_type, skel) # triggers this recursion for nodes, again. 

150 

151 if cursor := query.getCursor(): 

152 self.update_path_recursive(skel_type, path, parent_key, cursor) 

153 

154 def update_path(self, skel): 

155 """ 

156 Updates the path-value of a either a folder or a script file, by resolving the repository's root node. 

157 """ 

158 path = [skel["name"]] 

159 

160 key = skel["parententry"] 

161 while key: 

162 parent_skel = self.viewSkel("node") 

163 if not parent_skel.read(key) or parent_skel["key"] == skel["parentrepo"]: 

164 break 

165 

166 if parent_skel["name"]: 

167 path.insert(0, parent_skel["name"]) 

168 key = parent_skel["parententry"] 

169 skel["path"] = "/".join(path) 

170 

171 @exposed 

172 def get_importable(self): 

173 

174 def get_files_recursively(_importable_key): 

175 res = [] 

176 importable_files_query = self.viewSkel("leaf").all().filter("parententry", _importable_key) 

177 if not (importable_files_query := self.listFilter(importable_files_query)): 

178 raise errors.Unauthorized() 

179 for script_entry in importable_files_query.iter(): 

180 if script_entry["script"]: 

181 res.append(script_entry) 

182 importable_files_query = self.viewSkel("node").all().filter("parententry", _importable_key) 

183 for folder_entry in importable_files_query.iter(): 

184 res.extend(get_files_recursively(folder_entry.key)) 

185 return res 

186 

187 # get importable key 

188 qry_importable = (self.viewSkel("node").all() 

189 .filter("parententry", self.rootnodeSkel(ensure=True)["key"]) 

190 .filter("name =", "importable")) 

191 if not (qry_importable := self.listFilter(qry_importable)): 

192 raise errors.Unauthorized() 

193 

194 importable_key = (entity := qry_importable.getEntry()) and entity.key 

195 if not importable_key: 

196 raise errors.NotFound("No importable folder defined") 

197 

198 importable_files = get_files_recursively(importable_key) 

199 if not importable_files: 

200 raise errors.NotFound("Importable folder is empty") 

201 

202 zip_buffer = io.BytesIO() 

203 with zipfile.ZipFile(zip_buffer, "a", zipfile.ZIP_DEFLATED, False) as zip_file: 

204 for file in importable_files: 

205 zip_file.writestr(file["path"], file["script"]) 

206 

207 current.request.get().response.headers["Content-Disposition"] = "attachment; filename=importable.zip" 

208 current.request.get().response.headers["Content-Type"] = "application/zip" 

209 return zip_buffer.getvalue()