from driver.interfaces.serializers import HealReqSerializer, InstScaleHealRespSerializer, ScaleReqSerializer, \
NotifyReqSerializer, GrantRespSerializer, GrantReqSerializer, JobQueryRespSerializer, TerminateVnfRequestSerializer, \
- InstantiateVnfRequestSerializer, QueryVnfResponseSerializer
+ InstantiateVnfRequestSerializer, QueryVnfResponseSerializer, SubscribesRespSerializer, \
+ SubscribeReqSerializer, SubscribeRespSerializer, VnfPkgsSerializer, NfvoInfoReqSerializer
from driver.pub.config.config import VNF_FTP
from driver.pub.utils import restcall
from driver.pub.utils.restcall import req_by_msb
logger.debug("[%s] packageInfo=%s", fun_name(), packageInfo)
logger.debug("VNF_FTP=%s", VNF_FTP)
data = {
+ "vnfinstancename": "default",
"NFVOID": 1,
"VNFMID": vnfmid,
"vnfd_id": packageInfo.get("vnfdId"),
- "deployflavorid": "TODO",
+ "deployflavorid": "default",
"extension": {},
"inputs": []
}
for name, value in ignorcase_get(additionalParam, "inputs").items():
data["inputs"].append({"key_name": name, "value": value, "type": "TODO"})
+ inputs_json = load_json_file("inputs.json")
+ [data["inputs"].append(item) for item in inputs_json["inputs"]]
+
logger.debug("[%s] call_req data=%s", fun_name(), data)
ret = restcall.call_req(
class Subscribe(APIView):
+ @swagger_auto_schema(
+ responses={
+ status.HTTP_200_OK: SubscribesRespSerializer(),
+ status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"
+ }
+ )
def get(self, request):
logger.debug("====Subscribe get====")
resp_data = {
}
return Response(data=resp_data, status=status.HTTP_200_OK)
+ @swagger_auto_schema(
+ request_body=SubscribeReqSerializer(),
+ responses={
+ status.HTTP_201_CREATED: SubscribeRespSerializer(),
+ status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"
+ }
+ )
def post(self, request):
logger.debug("====Subscribe post====")
resp_data = {"subscribeid": "cdbddb00-452c-11e9-91e8-acc860114657"}
class SubscribeDetail(APIView):
+ @swagger_auto_schema(
+ responses={
+ status.HTTP_204_NO_CONTENT: "None",
+ status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"
+ }
+ )
def delete(self, request, subscribeId):
logger.debug("====SubscribeDetail delete %s====", subscribeId)
return Response(status=status.HTTP_204_NO_CONTENT)
class VnfPkgs(APIView):
+ @swagger_auto_schema(
+ responses={
+ status.HTTP_200_OK: VnfPkgsSerializer(),
+ status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"
+ }
+ )
def get(self, request):
logger.debug("====VnfPkgs get====")
resp_data = {
class VnfPkg(APIView):
+ @swagger_auto_schema(
+ responses={
+ status.HTTP_200_OK: "File stream for vnf pkg file",
+ status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"
+ }
+ )
def get(self, request, packageId, fileName):
logger.debug("====VnfPkg get====%s, %s", packageId, fileName)
file_range = request.META.get('RANGE')
start, end = parse_file_range(local_file_path, file_range)
file_iterator = read(local_file_path, start, end)
return StreamingHttpResponse(file_iterator, status=status.HTTP_200_OK)
+
+
+class NfvoInfo(APIView):
+ @swagger_auto_schema(
+ request_body=NfvoInfoReqSerializer(),
+ responses={
+ status.HTTP_200_OK: "Update successfully",
+ status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"
+ }
+ )
+ def put(self, request, vnfmid):
+ logger.debug("====NfvoInfo put====%s", vnfmid)
+ req_data = {
+ "nfvoid": request.data.get("nfvoid", "1"),
+ "vnfmid": vnfmid,
+ "nfvourl": request.data.get("nfvourl", "http://127.0.0.1:80")
+ }
+ ret = get_vnfminfo_from_nslcm(vnfmid)
+ if ret[0] != 0:
+ return Response(data={'error': ret[1]}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ vnfm_info = json.JSONDecoder().decode(ret[1])
+ logger.debug("[%s] vnfm_info=%s", fun_name(), vnfm_info)
+ ret = restcall.call_req(
+ base_url=ignorcase_get(vnfm_info, "url"),
+ user=ignorcase_get(vnfm_info, "userName"),
+ passwd=ignorcase_get(vnfm_info, "password"),
+ auth_type=restcall.rest_no_auth,
+ resource="v1/nfvo/info",
+ method='put',
+ content=json.dumps(req_data))
+ if ret[0] != 0:
+ return Response(data={'error': ret[1]}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+ logger.debug("update nfvo info successfully.")
+ return Response(data={}, status=status.HTTP_200_OK)
+
+
+class HealthCheckView(APIView):
+ @swagger_auto_schema(
+ responses={
+ status.HTTP_200_OK: 'Active'})
+ def get(self, request, format=None):
+ logger.debug("HealthCheck")
+ return Response({"status": "active"})