gerrit.onap Code Review - integration.git/blob - test/mocks/prov-mns-provider/src/ProvMnSProvider.py

1 from http.server import HTTPServer, BaseHTTPRequestHandler

2 import re

3 import json

4 import base64

5 from urllib.parse import urlparse, parse_qs

6

7 with open("DefinedNRMFunction.json",'r') as f:

8 jsonFile = json.loads(f.read())

9 SupportingFunctionList = jsonFile["NRMFunction"]

10

11 with open("UserInfo.json",'r') as f:

12 UserFile = json.loads(f.read())

13

14 with open("ConfigInfo.json",'r') as f:

15 ConfigFile = json.loads(f.read())

16

17 with open("preSetMOI.json",'r') as f:

18 Cretaed_MOIs = json.loads(f.read())

19 Cretaed_MOIs_list = Cretaed_MOIs['preSetMOI']

20

21 ipAddress = ConfigFile["ipAddress"]

22 portNumber = ConfigFile["portNumber"]

23 prefix = ConfigFile["prefix"]

24

25 username = UserFile['userName']

26 password = UserFile['password']

27 Auth_str = username+":"+password

28 print(Auth_str)

29 base64string = base64.b64encode(bytes(Auth_str,'utf-8'))

30 authheader = "Basic %s" % base64string.decode('utf-8')

31 print(authheader)

32

33 class ServerHTTP(BaseHTTPRequestHandler):

34 def do_GET(self):

35 path = self.path

36 print("\n**************************** NEW GET REQUEST ********************************")

37 request = urlparse(path)

38 print("the PATH of the received GET request:" + request.path)

39 pathlist = request.path.split('/')

40 prefix_check = True

41 try:

42 if "/" + pathlist[1] + "/"+ pathlist[2] != prefix:

43 prefix_check = False

44 className = pathlist[3]

45 idName = pathlist[4]

46 except IndexError:

47 prefix_check = False

48 response = {}

49 query_params = parse_qs(request.query)

50 if self.headers['Authorization'] == authheader and prefix_check is True:

51 if className in SupportingFunctionList:

52 try:

53 print("the value of the scope : "+ str(query_params['scope']))

54 print("the value of the filter : "+ str(query_params['filter']))

55 print("the value of the fields : "+ str(query_params['fields']))

56 except:

57 print("the request body doesn't follow the standard format")

58 response['error'] = "the request body doesn't follow the standard format"

59 print("Fail to get MOI object: "+'/' +className+'/'+idName)

60 self.send_response(406)

61 else:

62 find_moi = False

63 for MOI in Cretaed_MOIs_list:

64 if (idName == MOI['id'] and className == MOI['class']):

65 find_moi = True

66 try:

67 attributes = {}

68 for field in query_params['fields']:

69 attributes[field] = MOI['attributes'][field]

70 except:

71 print("the createed MOI doesn't contain the required attribute")

72 response['error'] = "the createed MOI doesn't contain the required attribute"

73 print("Fail to get MOI object: "+'/' +className+'/'+idName)

74 self.send_response(406)

75 else:

76 print("Successfully get MOI object: "+ className+'_'+idName)

77 response = {"data":[{"href":"/"+className+"/"+idName,"class":className,"id":idName,"attributes":attributes}]}

78 self.send_response(200)

79 if (find_moi is False):

80 response['error'] = {"errorInfo":"MOI does not exist"}

81 print("Fail to get MOI object: "+'/' +className+'/'+idName)

82 self.send_response(406)

83 else:

84 response['error'] = {"errorInfo":"MOI class not support"}

85 print("Fail to get MOI object: "+'/' +className+'/'+idName)

86 self.send_response(406)

87 else:

88 if prefix_check is True:

89 self.send_response(401)

90 response['error'] = {"errorInfo":"not Authorized"}

91 else:

92 self.send_response(404)

93 response['error'] = {"errorInfo":"wrong prefix"}

94 self.send_header("Content-type","application/json")

95 self.end_headers()

96 buf = json.dumps(response)

97 self.wfile.write(bytes(buf,'utf-8'))

98

99 def do_PATCH(self):

100 path = self.path

101 print("\n**************************** NEW PATCH REQUEST ********************************")

102 request = urlparse(path)

103 print("the PATH of the received GET request:" + request.path)

104 pathlist = request.path.split('/')

105 prefix_check = True

106 try:

107 if "/" + pathlist[1] + "/"+ pathlist[2] != prefix:

108 prefix_check = False

109 className = pathlist[3]

110 idName = pathlist[4]

111 except IndexError:

112 prefix_check = False

113 response = {}

114 query_params = parse_qs(request.query)

115 if self.headers['Authorization'] == authheader and prefix_check is True:

116 if className in SupportingFunctionList:

117 datas = self.rfile.read(int(self.headers['content-length']))

118 json_str = datas.decode('utf-8')

119 json_str = re.sub('\'','\"', json_str)

120 json_dict = json.loads(json_str)

121 try:

122 print("the value of the scope : "+ str(query_params['scope']))

123 print("the value of the filter : "+ str(query_params['filter']))

124 print("the modified attribute values : "+json.dumps(json_dict['data']))

125 except:

126 print("the request body doesn't follow the standard format")

127 response['error'] = "the request body doesn't follow the standard format"

128 print("Fail to modify MOI object: "+'/' +className+'/'+idName)

129 self.send_response(406)

130 else:

131 find_moi = False

132 for MOI in Cretaed_MOIs_list:

133 if (idName == MOI['id'] and className == MOI['class']):

134 find_moi = True

135 wrong_attribute = False

136 for key, value in json_dict['data'].items():

137 if (key in MOI['attributes']):

138 MOI['attributes'][key] = value

139 else:

140 wrong_attribute = True

141 if (wrong_attribute is True):

142 print("the createed MOI doesn't contain the required attribute")

143 response['error'] = "the createed MOI doesn't contain the required attribute"

144 print("Fail to get modify object: "+'/' +className+'/'+idName)

145 self.send_response(406)

146 else:

147 print("Successfully modify MOI object: "+ className+'_'+idName)

148 response = {"data":[MOI]}

149 self.send_response(200)

150 if (find_moi is False):

151 response['error'] = {"errorInfo":"MOI does not exist"}

152 print("Fail to get MOI object: "+'/' +className+'/'+idName)

153 self.send_response(406)

154 else:

155 response['error'] = {"errorInfo":"MOI class not support"}

156 print("Fail to modify MOI object: "+'/' +className+'/'+idName)

157 self.send_response(406)

158 else:

159 if prefix_check is True:

160 self.send_response(401)

161 response['error'] = {"errorInfo":"not Authorized"}

162 else:

163 self.send_response(404)

164 response['error'] = {"errorInfo":"wrong prefix"}

165 self.send_header("Content-type","application/json")

166 self.end_headers()

167 buf = json.dumps(response)

168 self.wfile.write(bytes(buf,'utf-8'))

169

170 def do_DELETE(self):

171 path = self.path

172 print("\n**************************** NEW DELETE REQUEST ********************************")

173 request = urlparse(path)

174 print("the PATH of the received DELETE request:" + request.path)

175 pathlist = request.path.split('/')

176 prefix_check = True

177 try:

178 if "/" + pathlist[1] + "/"+ pathlist[2] != prefix:

179 prefix_check = False

180 className = pathlist[3]

181 idName = pathlist[4]

182 except IndexError:

183 prefix_check = False

184 response = {}

185 query_params = parse_qs(request.query)

186 if self.headers['Authorization'] == authheader and prefix_check is True:

187 if className in SupportingFunctionList:

188 try:

189 print("the value of the scope : "+ str(query_params['scope']))

190 print("the value of the filter : "+ str(query_params['filter']))

191 except:

192 print("the request body doesn't follow the standard format")

193 response['error'] = "the request body doesn't follow the standard format"

194 print("Fail to delete MOI object: "+'/' +className+'/'+idName)

195 self.send_response(406)

196 else:

197 find_moi = False

198 for MOI in Cretaed_MOIs_list:

199 if (idName == MOI['id'] and className == MOI['class']):

200 find_moi = True

201 Cretaed_MOIs_list.remove(MOI)

202 print("Successfully delete MOI object: "+ className+'_'+idName)

203 response = {"data":["/"+className+"/"+idName]}

204 self.send_response(200)

205 if (find_moi is False):

206 response['error'] = {"errorInfo":"MOI does not exist"}

207 print("Fail to delete MOI object: "+'/' +className+'/'+idName)

208 self.send_response(406)

209 else:

210 response['error'] = {"errorInfo":"MOI class not support"}

211 print("Fail to delete MOI object: "+'/' +className+'/'+idName)

212 self.send_response(406)

213 else:

214 if prefix_check is True:

215 self.send_response(401)

216 response['error'] = {"errorInfo":"not Authorized"}

217 else:

218 self.send_response(404)

219 response['error'] = {"errorInfo":"wrong prefix"}

220 self.send_header("Content-type","application/json")

221 self.end_headers()

222 buf = json.dumps(response)

223 self.wfile.write(bytes(buf,'utf-8'))

224

225 def do_PUT(self):

226 path = self.path

227 print("\n**************************** NEW PUT REQUEST ********************************")

228 print("the PATH of the received PUT request:" + path)

229 pathlist = path.split('/')

230 prefix_check = True

231 try:

232 if "/" + pathlist[1] + "/"+ pathlist[2] != prefix:

233 prefix_check = False

234 className = pathlist[3]

235 idName = pathlist[4]

236 except IndexError:

237 prefix_check = False

238 response = {}

239 if self.headers['Authorization'] == authheader and prefix_check is True:

240 if className in SupportingFunctionList:

241 datas = self.rfile.read(int(self.headers['content-length']))

242 json_str = datas.decode('utf-8')

243 json_str = re.sub('\'','\"', json_str)

244 json_dict = json.loads(json_str)

245 try:

246 print("the class of the New MOI : "+json_dict['data']['class'])

247 print("the ID of the New MOI : "+json_dict['data']['id'])

248 print("the href of the New MOI : "+json_dict['data']['href'])

249 print("the attributes of the New MOI : "+json.dumps(json_dict['data']['attributes']))

250 except:

251 print("the request body doesn't follow the standard format")

252 response['error'] = "the request body doesn't follow the standard format"

253 print("Fail to create MOI object: "+'/' +className+'/'+idName)

254 self.send_response(406)

255 else:

256 print("Successfully create MOI object: "+ className+'/'+idName)

257 Cretaed_MOIs_list.append(json_dict['data'])

258 response = json_dict

259 self.send_response(201)

260 self.send_header("Location",path)

261 else:

262 response['error'] = {"errorInfo":"MOI class not support"}

263 print("Fail to create MOI object: "+'/' +className+'/'+idName)

264 self.send_response(406)

265 else:

266 if prefix_check is True:

267 self.send_response(401)

268 response['error'] = {"errorInfo":"not Authorized"}

269 else:

270 self.send_response(404)

271 response['error'] = {"errorInfo":"wrong prefix"}

272 self.send_header("Content-type","application/json")

273 self.end_headers()

274 buf = json.dumps(response)

275 self.wfile.write(bytes(buf,'utf-8'))

276

277 def start_server(port):

278 http_server = HTTPServer((ipAddress, int(port)), ServerHTTP)

279 http_server.serve_forever()

280

281 if __name__ == "__main__":

282 start_server(int(portNumber))