+ def _get_cached_file_folder_name(self,folder):
+ cached_file_folder = '%s/%s' % (folder, self.path,)
+ print("Cached file name before escaping : %s" % cached_file_folder)
+ cached_file_folder = cached_file_folder.replace('<','<').replace('>','>').replace('?','?').replace('*','*').replace('\\','*').replace(':',':').replace('|','|')
+ print("Cached file name after escaping (used for cache storage) : %s" % cached_file_folder)
+ return cached_file_folder
+
+ def _get_cached_content_file_name(self,cached_file_folder):
+ return "%s/.file" % (cached_file_folder,)
+
+ def _get_cached_header_file_name(self,cached_file_folder):
+ return "%s/.header" % (cached_file_folder,)
+
+ def _execute_content_generated_cases(self,http_type):
+ print("Testing special cases, cache files will be sent to :" +TMP_ROOT)
+ cached_file_folder = self._get_cached_file_folder_name(TMP_ROOT)
+ cached_file_content = self._get_cached_content_file_name(cached_file_folder)
+ cached_file_header = self._get_cached_header_file_name(cached_file_folder)
+ _file_available = os.path.exists(cached_file_content)
+
+ if self.path.startswith("/dcae-service-types?asdcResourceId=") and http_type == "GET":
+ if not _file_available:
+ print "self.path start with /dcae-service-types?asdcResourceId=, generating response json..."
+ uuidGenerated = str(uuid.uuid4())
+ typeId = "typeId-" + uuidGenerated
+ typeName = "typeName-" + uuidGenerated
+ print "typeId generated: " + typeName + " and typeName: "+ typeId
+ jsonGenerated = "{\"totalCount\":1, \"items\":[{\"typeId\":\"" + typeId + "\", \"typeName\":\"" + typeName +"\"}]}"
+ print "jsonGenerated: " + jsonGenerated
+
+ os.makedirs(cached_file_folder, 0777)
+ with open(cached_file_header, 'w') as f:
+ f.write("{\"Content-Length\": \"" + str(len(jsonGenerated)) + "\", \"Content-Type\": \"application/json\"}")
+ with open(cached_file_content, 'w') as f:
+ f.write(jsonGenerated)
+ return True
+ elif self.path.startswith("/dcae-operationstatus/install") and http_type == "GET":
+ if not _file_available:
+ print "self.path start with /dcae-operationstatus/install, generating response json..."
+ jsonGenerated = "{\"operationType\": \"install\", \"status\": \"succeeded\"}"
+ print "jsonGenerated: " + jsonGenerated
+
+ try:
+ os.makedirs(cached_file_folder, 0777)
+ except OSError as e:
+ if e.errno != errno.EEXIST:
+ raise
+ print(cached_file_folder+" already exists")
+
+ with open(cached_file_header, 'w') as f:
+ f.write("{\"Content-Length\": \"" + str(len(jsonGenerated)) + "\", \"Content-Type\": \"application/json\"}")
+ with open(cached_file_content, 'w') as f:
+ f.write(jsonGenerated)
+ return True
+ elif self.path.startswith("/dcae-operationstatus/uninstall") and http_type == "GET":
+ if not _file_available:
+ print "self.path start with /dcae-operationstatus/uninstall, generating response json..."
+ jsonGenerated = "{\"operationType\": \"uninstall\", \"status\": \"succeeded\"}"
+ print "jsonGenerated: " + jsonGenerated
+
+ try:
+ os.makedirs(cached_file_folder, 0777)
+ except OSError as e:
+ if e.errno != errno.EEXIST:
+ raise
+ print(cached_file_folder+" already exists")
+
+ with open(cached_file_header, 'w') as f:
+ f.write("{\"Content-Length\": \"" + str(len(jsonGenerated)) + "\", \"Content-Type\": \"application/json\"}")
+ with open(cached_file_content, 'w') as f:
+ f.write(jsonGenerated)
+ return True
+ elif self.path.startswith("/sdc/v1/catalog/services/") and http_type == "POST":
+ if not _file_available:
+ print "self.path start with /sdc/v1/catalog/services/, generating response json..."
+ jsondata = json.loads(self.data_string)
+ jsonGenerated = "{\"artifactName\":\"" + jsondata['artifactName'] + "\",\"artifactType\":\"" + jsondata['artifactType'] + "\",\"artifactURL\":\"" + self.path + "\",\"artifactDescription\":\"" + jsondata['description'] + "\",\"artifactChecksum\":\"ZjJlMjVmMWE2M2M1OTM2MDZlODlmNTVmZmYzNjViYzM=\",\"artifactUUID\":\"" + str(uuid.uuid4()) + "\",\"artifactVersion\":\"1\"}"
+ print "jsonGenerated: " + jsonGenerated
+
+ os.makedirs(cached_file_folder, 0777)
+ with open(cached_file_header, 'w') as f:
+ f.write("{\"Content-Length\": \"" + str(len(jsonGenerated)) + "\", \"Content-Type\": \"application/json\"}")
+ with open(cached_file_content, 'w') as f:
+ f.write(jsonGenerated)
+ return True
+ elif self.path.startswith("/dcae-deployments/") and http_type == "PUT":
+ print "self.path start with /dcae-deployments/ DEPLOY, generating response json..."
+ #jsondata = json.loads(self.data_string)
+ jsonGenerated = "{\"operationType\":\"install\",\"status\":\"processing\",\"links\":{\"status\":\"http:\/\/" + PROXY_ADDRESS + "\/dcae-operationstatus/install\"}}"
+ print "jsonGenerated: " + jsonGenerated
+ if not os.path.exists(cached_file_folder):
+ os.makedirs(cached_file_folder, 0777)
+ with open(cached_file_header, 'w+') as f:
+ f.write("{\"Content-Length\": \"" + str(len(jsonGenerated)) + "\", \"Content-Type\": \"application/json\"}")
+ with open(cached_file_content, 'w+') as f:
+ f.write(jsonGenerated)
+ return True
+ elif self.path.startswith("/dcae-deployments/") and http_type == "DELETE":
+ print "self.path start with /dcae-deployments/ UNDEPLOY, generating response json..."
+ #jsondata = json.loads(self.data_string)
+ jsonGenerated = "{\"operationType\":\"uninstall\",\"status\":\"processing\",\"links\":{\"status\":\"http:\/\/" + PROXY_ADDRESS + "\/dcae-operationstatus/uninstall\"}}"
+ print "jsonGenerated: " + jsonGenerated
+ if not os.path.exists(cached_file_folder):
+ os.makedirs(cached_file_folder, 0777)
+ with open(cached_file_header, 'w+') as f:
+ f.write("{\"Content-Length\": \"" + str(len(jsonGenerated)) + "\", \"Content-Type\": \"application/json\"}")
+ with open(cached_file_content, 'w+') as f:
+ f.write(jsonGenerated)
+ return True
+ elif (self.path.startswith("/pdp/api/") and (http_type == "PUT" or http_type == "DELETE")) or (self.path.startswith("/pdp/api/policyEngineImport") and http_type == "POST"):
+ print "self.path start with /pdp/api/, copying body to response ..."
+ if not os.path.exists(cached_file_folder):
+ os.makedirs(cached_file_folder, 0777)
+ with open(cached_file_header, 'w+') as f:
+ f.write("{\"Content-Length\": \"" + str(len(self.data_string)) + "\", \"Content-Type\": \""+str(self.headers['Content-Type'])+"\"}")
+ with open(cached_file_content, 'w+') as f:
+ f.write(self.data_string)
+ return True
+ elif self.path.startswith("/policy/api/v1/policytypes/") and http_type == "POST":
+ print "self.path start with POST new policy API /pdp/api/, copying body to response ..."
+ if not os.path.exists(cached_file_folder):
+ os.makedirs(cached_file_folder, 0777)
+ with open(cached_file_header, 'w+') as f:
+ f.write("{\"Content-Length\": \"" + str(len(self.data_string)) + "\", \"Content-Type\": \""+str(self.headers['Content-Type'])+"\"}")
+ with open(cached_file_content, 'w+') as f:
+ f.write(self.data_string)
+ return True
+ elif self.path.startswith("/policy/api/v1/policytypes/") and http_type == "DELETE":
+ print "self.path start with DELETE new policy API /policy/api/v1/policytypes/ ..."
+ if not os.path.exists(cached_file_folder):
+ os.makedirs(cached_file_folder, 0777)
+
+ with open(cached_file_header, 'w+') as f:
+ f.write("{\"Content-Length\": \"" + str(len("")) + "\", \"Content-Type\": \""+str("")+"\"}")
+ with open(cached_file_content, 'w+') as f:
+ f.write(self.data_string)
+ return True
+ elif self.path.startswith("/policy/pap/v1/pdps/policies") and http_type == "POST":
+ print "self.path start with POST new policy API /policy/pap/v1/pdps/ ..."
+ if not os.path.exists(cached_file_folder):
+ os.makedirs(cached_file_folder, 0777)
+ with open(cached_file_header, 'w+') as f:
+ f.write("{\"Content-Length\": \"" + str(len("")) + "\", \"Content-Type\": \""+str("")+"\"}")
+ with open(cached_file_content, 'w+') as f:
+ f.write(self.data_string)
+ return True
+ elif self.path.startswith("/policy/api/v1/policytypes/") and http_type == "GET":
+ print "self.path start with /policy/api/v1/policytypes/, generating response json..."
+ jsonGenerated = "{\"policyTypeId\": \"onap.policies.controlloop.operational\",\"policyTypeVersion\": \"1.0.0\",\"policyId\": \"OPERATIONAL_z711F_v1_0_ResourceInstanceName1_tca\"}"
+ print "jsonGenerated: " + jsonGenerated
+ if not os.path.exists(cached_file_folder):
+ os.makedirs(cached_file_folder, 0777)
+
+ with open(cached_file_header, 'w') as f:
+ f.write("{\"Content-Length\": \"" + str(len(jsonGenerated)) + "\", \"Content-Type\": \"application/json\"}")
+ with open(cached_file_content, 'w') as f:
+ f.write(jsonGenerated)
+ return True
+ else:
+ return False
+
+