Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/flask_restx/postman.py: 33%

1# -*- coding: utf-8 -*- 

2from time import time 

3from uuid import uuid5, NAMESPACE_URL 

4 

5from urllib.parse import urlencode 

6 

7 

8def clean(data): 

9 """Remove all keys where value is None""" 

10 return dict((k, v) for k, v in data.items() if v is not None) 

11 

12 

13DEFAULT_VARS = { 

14 "string": "", 

15 "integer": 0, 

16 "number": 0, 

17} 

18 

19 

20class Request(object): 

21 """Wraps a Swagger operation into a Postman Request""" 

22 

23 def __init__(self, collection, path, params, method, operation): 

24 self.collection = collection 

25 self.path = path 

26 self.params = params 

27 self.method = method.upper() 

28 self.operation = operation 

29 

30 @property 

31 def id(self): 

32 seed = str(" ".join((self.method, self.url))) 

33 return str(uuid5(self.collection.uuid, seed)) 

34 

35 @property 

36 def url(self): 

37 return self.collection.api.base_url.rstrip("/") + self.path 

38 

39 @property 

40 def headers(self): 

41 headers = {} 

42 # Handle content-type 

43 if self.method != "GET": 

44 consumes = self.collection.api.__schema__.get("consumes", []) 

45 consumes = self.operation.get("consumes", consumes) 

46 if len(consumes): 

47 headers["Content-Type"] = consumes[-1] 

48 

49 # Add all parameters headers 

50 for param in self.operation.get("parameters", []): 

51 if param["in"] == "header": 

52 headers[param["name"]] = param.get("default", "") 

53 

54 # Add security headers if needed (global then local) 

55 for security in self.collection.api.__schema__.get("security", []): 

56 for key, header in self.collection.apikeys.items(): 

57 if key in security: 

58 headers[header] = "" 

59 for security in self.operation.get("security", []): 

60 for key, header in self.collection.apikeys.items(): 

61 if key in security: 

62 headers[header] = "" 

63 

64 lines = [":".join(line) for line in headers.items()] 

65 return "\n".join(lines) 

66 

67 @property 

68 def folder(self): 

69 if "tags" not in self.operation or len(self.operation["tags"]) == 0: 

70 return 

71 tag = self.operation["tags"][0] 

72 for folder in self.collection.folders: 

73 if folder.tag == tag: 

74 return folder.id 

75 

76 def as_dict(self, urlvars=False): 

77 url, variables = self.process_url(urlvars) 

78 return clean( 

79 { 

80 "id": self.id, 

81 "method": self.method, 

82 "name": self.operation["operationId"], 

83 "description": self.operation.get("summary"), 

84 "url": url, 

85 "headers": self.headers, 

86 "collectionId": self.collection.id, 

87 "folder": self.folder, 

88 "pathVariables": variables, 

89 "time": int(time()), 

90 } 

91 ) 

92 

93 def process_url(self, urlvars=False): 

94 url = self.url 

95 path_vars = {} 

96 url_vars = {} 

97 params = dict((p["name"], p) for p in self.params) 

98 params.update( 

99 dict((p["name"], p) for p in self.operation.get("parameters", [])) 

100 ) 

101 if not params: 

102 return url, None 

103 for name, param in params.items(): 

104 if param["in"] == "path": 

105 url = url.replace("{%s}" % name, ":%s" % name) 

106 path_vars[name] = DEFAULT_VARS.get(param["type"], "") 

107 elif param["in"] == "query" and urlvars: 

108 default = DEFAULT_VARS.get(param["type"], "") 

109 url_vars[name] = param.get("default", default) 

110 if url_vars: 

111 url = "?".join((url, urlencode(url_vars))) 

112 return url, path_vars 

113 

114 

115class Folder(object): 

116 def __init__(self, collection, tag): 

117 self.collection = collection 

118 self.tag = tag["name"] 

119 self.description = tag["description"] 

120 

121 @property 

122 def id(self): 

123 return str(uuid5(self.collection.uuid, str(self.tag))) 

124 

125 @property 

126 def order(self): 

127 return [r.id for r in self.collection.requests if r.folder == self.id] 

128 

129 def as_dict(self): 

130 return clean( 

131 { 

132 "id": self.id, 

133 "name": self.tag, 

134 "description": self.description, 

135 "order": self.order, 

136 "collectionId": self.collection.id, 

137 } 

138 ) 

139 

140 

141class PostmanCollectionV1(object): 

142 """Postman Collection (V1 format) serializer""" 

143 

144 def __init__(self, api, swagger=False): 

145 self.api = api 

146 self.swagger = swagger 

147 

148 @property 

149 def uuid(self): 

150 return uuid5(NAMESPACE_URL, self.api.base_url) 

151 

152 @property 

153 def id(self): 

154 return str(self.uuid) 

155 

156 @property 

157 def requests(self): 

158 if self.swagger: 

159 # First request is Swagger specifications 

160 yield Request( 

161 self, 

162 "/swagger.json", 

163 {}, 

164 "get", 

165 { 

166 "operationId": "Swagger specifications", 

167 "summary": "The API Swagger specifications as JSON", 

168 }, 

169 ) 

170 # Then iter over API paths and methods 

171 for path, operations in self.api.__schema__["paths"].items(): 

172 path_params = operations.get("parameters", []) 

173 

174 for method, operation in operations.items(): 

175 if method != "parameters": 

176 yield Request(self, path, path_params, method, operation) 

177 

178 @property 

179 def folders(self): 

180 for tag in self.api.__schema__["tags"]: 

181 yield Folder(self, tag) 

182 

183 @property 

184 def apikeys(self): 

185 return dict( 

186 (name, secdef["name"]) 

187 for name, secdef in self.api.__schema__.get("securityDefinitions").items() 

188 if secdef.get("in") == "header" and secdef.get("type") == "apiKey" 

189 ) 

190 

191 def as_dict(self, urlvars=False): 

192 return clean( 

193 { 

194 "id": self.id, 

195 "name": " ".join((self.api.title, self.api.version)), 

196 "description": self.api.description, 

197 "order": [r.id for r in self.requests if not r.folder], 

198 "requests": [r.as_dict(urlvars=urlvars) for r in self.requests], 

199 "folders": [f.as_dict() for f in self.folders], 

200 "timestamp": int(time()), 

201 } 

202 )