Coverage for /usr/local/lib/python3.11/dist-packages/pyrocko/client/wadl.py: 98%

1# http://pyrocko.org - GPLv3 

2# 

3# The Pyrocko Developers, 21st Century 

4# ---|P------/S----------~Lg---------- 

5 

6''' 

7WADL data model. 

8''' 

9 

10import re 

11 

12from pyrocko import guts 

13from pyrocko.guts import make_typed_list_class, String, StringChoice, List, \ 

14 Int, Object, StringUnion, Bool, Defer 

15 

16from pyrocko.io.io_common import FileLoadError 

17 

18ResourceTypeList = make_typed_list_class(String) 

19 

20 

21guts_prefix = 'wadl' 

22guts_xmlns = 'http://wadl.dev.java.net/2009/02' 

23 

24re_rmsite = re.compile(r'https?://[^/]+') 

25re_multisl = re.compile(r'/+') 

26 

27 

28def clean_path(p): 

29 p = re_rmsite.sub('', p) 

30 p = re_multisl.sub('/', p) 

31 if not p.startswith('/'): 

32 p = '/' + p 

33 

34 return p 

35 

36 

37class HTTPMethods(StringChoice): 

38 choices = [ 

39 'GET', 

40 'POST', 

41 'PUT', 

42 'HEAD', 

43 'DELETE'] 

44 

45 

46UriList = make_typed_list_class(String) 

47 

48StatusCodeList = make_typed_list_class(Int) 

49 

50 

51class ParamStyle(StringChoice): 

52 choices = [ 

53 'plain', 

54 'query', 

55 'matrix', 

56 'header', 

57 'template'] 

58 

59 

60class Doc(Object): 

61 xmltagname = 'doc' 

62 title = String.T(optional=True, xmlstyle='attribute') 

63 

64 

65class Method2(StringUnion): 

66 members = [HTTPMethods, String] 

67 

68 

69class Include(Object): 

70 xmltagname = 'include' 

71 href = String.T(optional=True, xmlstyle='attribute') 

72 doc_list = List.T(Doc.T()) 

73 

74 

75class Option(Object): 

76 xmltagname = 'option' 

77 value = String.T(xmlstyle='attribute') 

78 media_type = String.T(optional=True, xmlstyle='attribute') 

79 doc_list = List.T(Doc.T()) 

80 

81 

82class Link(Object): 

83 xmltagname = 'link' 

84 resource_type = String.T( 

85 optional=True, xmlstyle='attribute', xmltagname='resource_type') 

86 rel = String.T(optional=True, xmlstyle='attribute') 

87 rev = String.T(optional=True, xmlstyle='attribute') 

88 doc_list = List.T(Doc.T()) 

89 

90 

91class Grammars(Object): 

92 xmltagname = 'grammars' 

93 doc_list = List.T(Doc.T()) 

94 include_list = List.T(Include.T()) 

95 

96 

97class DerefError(Exception): 

98 pass 

99 

100 

101class Element(Object): 

102 

103 def __init__(self, *args, **kwargs): 

104 Object.__init__(self, *args, **kwargs) 

105 self._hyper = None 

106 

107 def _update(self, hyper): 

108 self._hyper = hyper 

109 

110 id_ = getattr(self, 'id', None) 

111 

112 if id_: 

113 hyper[self.xmltagname][id_] = self 

114 

115 for child in self.get_children(): 

116 child._update(hyper) 

117 

118 def get_children(self): 

119 raise NotImplementedError() 

120 

121 def deref(self): 

122 if not self._hyper: 

123 raise Exception('Must call _update() before calling deref()') 

124 

125 obj = self 

126 seen = set() 

127 while True: 

128 seen.add(obj) 

129 href = getattr(obj, 'href', None) 

130 if href: 

131 try: 

132 obj = obj._hyper[obj.xmltagname][href.lstrip('#')] 

133 if obj in seen: 

134 raise DerefError('cyclic reference') 

135 

136 except KeyError: 

137 raise DerefError(href) 

138 

139 else: 

140 return obj 

141 

142 

143class Param(Element): 

144 xmltagname = 'param' 

145 href = String.T(optional=True, xmlstyle='attribute') 

146 name = String.T(optional=True, xmlstyle='attribute') 

147 style = ParamStyle.T(optional=True, xmlstyle='attribute') 

148 id = String.T(optional=True, xmlstyle='attribute') 

149 type = String.T(default='xs:string', optional=True, xmlstyle='attribute') 

150 default = String.T(optional=True, xmlstyle='attribute') 

151 required = Bool.T(default='false', optional=True, xmlstyle='attribute') 

152 repeating = Bool.T(default='false', optional=True, xmlstyle='attribute') 

153 fixed = String.T(optional=True, xmlstyle='attribute') 

154 path = String.T(optional=True, xmlstyle='attribute') 

155 doc_list = List.T(Doc.T()) 

156 option_list = List.T(Option.T()) 

157 link = Link.T(optional=True) 

158 

159 def describe(self, indent): 

160 return indent + self.name 

161 

162 def get_children(self): 

163 return [] 

164 

165 

166class Representation(Element): 

167 xmltagname = 'representation' 

168 id = String.T(optional=True, xmlstyle='attribute') 

169 element = String.T(optional=True, xmlstyle='attribute') 

170 media_type = String.T(optional=True, xmlstyle='attribute') 

171 href = String.T(optional=True, xmlstyle='attribute') 

172 profile = UriList.T(optional=True, xmlstyle='attribute') 

173 doc_list = List.T(Doc.T()) 

174 param_list = List.T(Param.T()) 

175 

176 def get_children(self): 

177 return self.param_list 

178 

179 

180class Request(Element): 

181 xmltagname = 'request' 

182 doc_list = List.T(Doc.T()) 

183 param_list = List.T(Param.T()) 

184 representation_list = List.T(Representation.T()) 

185 

186 def iter_params(self): 

187 for param in self.param_list: 

188 param = param.deref() 

189 yield param 

190 

191 def describe(self, indent): 

192 lines = [] 

193 for param in self.iter_params(): 

194 lines.append(param.describe(indent)) 

195 

196 return lines 

197 

198 def get_children(self): 

199 return self.param_list + self.representation_list 

200 

201 

202class Response(Element): 

203 xmltagname = 'response' 

204 status = StatusCodeList.T(optional=True, xmlstyle='attribute') 

205 doc_list = List.T(Doc.T()) 

206 param_list = List.T(Param.T()) 

207 representation_list = List.T(Representation.T()) 

208 

209 def get_children(self): 

210 return self.param_list + self.representation_list 

211 

212 

213class Method(Element): 

214 xmltagname = 'method' 

215 id = String.T(optional=True, xmlstyle='attribute') 

216 name = String.T(optional=True, xmlstyle='attribute') 

217 href = String.T(optional=True, xmlstyle='attribute') 

218 doc_list = List.T(Doc.T()) 

219 request = Request.T(optional=True) 

220 response_list = List.T(Response.T()) 

221 

222 def describe(self, indent): 

223 lines = [indent + self.name] 

224 if self.request: 

225 lines.extend(self.request.describe(' ' + indent)) 

226 

227 return lines 

228 

229 def get_children(self): 

230 return ([self.request] if self.request else []) + self.response_list 

231 

232 

233class Resource(Element): 

234 xmltagname = 'resource' 

235 id = String.T(optional=True, xmlstyle='attribute') 

236 type = String.T(optional=True, xmlstyle='attribute') 

237 query_type = String.T( 

238 default='application/x-www-form-urlencoded', 

239 optional=True, 

240 xmlstyle='attribute') 

241 path = String.T(optional=True, xmlstyle='attribute') 

242 doc_list = List.T(Doc.T()) 

243 param_list = List.T(Param.T()) 

244 method_list = List.T(Method.T()) 

245 resource_list = List.T(Defer('Resource.T')) 

246 

247 def iter_resources(self): 

248 yield self.path, self 

249 for res in self.resource_list: 

250 yield self.path + '/' + res.path, res 

251 

252 def iter_methods(self): 

253 for method in self.method_list: 

254 method = method.deref() 

255 yield method 

256 

257 def describe(self, indent): 

258 lines = [] 

259 for met in self.iter_methods(): 

260 lines.extend(met.describe(' ' + indent)) 

261 

262 return lines 

263 

264 def get_children(self): 

265 return self.param_list + self.method_list + self.resource_list 

266 

267 

268class Resources(Element): 

269 xmltagname = 'resources' 

270 base = String.T(optional=True, xmlstyle='attribute') 

271 doc_list = List.T(Doc.T()) 

272 resource_list = List.T(Resource.T()) 

273 

274 def iter_resources(self): 

275 for res in self.resource_list: 

276 for p, sr in res.iter_resources(): 

277 yield self.base + '/' + p, sr 

278 

279 def get_children(self): 

280 return self.resource_list 

281 

282 

283class ResourceType(Element): 

284 xmltagname = 'resource_type' 

285 id = String.T(optional=True, xmlstyle='attribute') 

286 doc_list = List.T(Doc.T()) 

287 param_list = List.T(Param.T()) 

288 method_list = List.T(Method.T()) 

289 resource_list = List.T(Resource.T()) 

290 

291 def get_children(self): 

292 return self.param_list + self.method_list + self.resource_list 

293 

294 

295class Application(Element): 

296 xmltagname = 'application' 

297 guessable_xmlns = [guts_xmlns] 

298 

299 doc_list = List.T(Doc.T()) 

300 grammars = Grammars.T(optional=True) 

301 resources_list = List.T(Resources.T()) 

302 resource_type_list = List.T(ResourceType.T(xmltagname='resource_type')) 

303 method_list = List.T(Method.T()) 

304 representation_list = List.T(Representation.T()) 

305 param_list = List.T(Param.T()) 

306 

307 def get_children(self): 

308 return self.resources_list + self.resource_type_list \ 

309 + self.method_list + self.representation_list \ 

310 + self.param_list 

311 

312 def update(self, force=False): 

313 if self._hyper is None or force: 

314 hyper = dict( 

315 resource_type={}, 

316 resource={}, 

317 method={}, 

318 representation={}, 

319 param={}) 

320 

321 self._update(hyper) 

322 

323 def iter_resources(self): 

324 self.update() 

325 for rs in self.resources_list: 

326 for p, res in rs.iter_resources(): 

327 yield clean_path(p), res 

328 

329 def iter_requests(self): 

330 for res_path, res in self.iter_resources(): 

331 for method in res.iter_methods(): 

332 if method.request: 

333 yield res_path, method.name, method.request 

334 

335 def supported_param_names(self, path, method='GET'): 

336 path = clean_path(path) 

337 for res_path, method_name, request in self.iter_requests(): 

338 if res_path == path and method_name == method: 

339 return [param.name for param in request.param_list] 

340 

341 def describe(self, indent=''): 

342 lines = [] 

343 for res_path, res in self.iter_resources(): 

344 lines.append(indent + res_path) 

345 lines.extend(res.describe(indent)) 

346 

347 return lines 

348 

349 def __str__(self): 

350 return '\n'.join(self.describe()) 

351 

352 

353def load_xml(*args, **kwargs): 

354 wadl = guts.load_xml(*args, **kwargs) 

355 if not isinstance(wadl, Application): 

356 FileLoadError('Not a WADL file.') 

357 

358 return wadl 

359 

360 

361if __name__ == '__main__': 

362 import os 

363 import sys 

364 import urllib.request 

365 

366 if os.path.exists(sys.argv[1]): 

367 wadl = load_xml(filename=sys.argv[1]) 

368 else: 

369 f = urllib.request.urlopen(sys.argv[1]) 

370 wadl = load_xml(stream=f) 

371 

372 print(wadl)