+ def nsd_create_pkg_base(self, token, pkg_name):
+ result = {'error': True, 'data': ''}
+ headers = {"Content-Type": "application/gzip", "accept": "application/json",
+ 'Authorization': 'Bearer {}'.format(token['id'])}
+
+ _url = "{0}/nsd/v1/ns_descriptors_content/".format(self._base_path)
+
+ try:
+ self._create_base_pkg('nsd', pkg_name)
+ r = requests.post(_url, data=open('/tmp/' + pkg_name + '.tar.gz', 'rb'), verify=False, headers=headers)
+ except Exception as e:
+ log.exception(e)
+ result['data'] = str(e)
+ return result
+ if r.status_code == requests.codes.created:
+ result['data'] = r.json()
+ result['error'] = False
+ if r.status_code == requests.codes.conflict:
+ result['data'] = "Invalid ID."
+ return result
+
+ def vnfd_create_pkg_base(self, token, pkg_name):
+ result = {'error': True, 'data': ''}
+ headers = {"Content-Type": "application/gzip", "accept": "application/json",
+ 'Authorization': 'Bearer {}'.format(token['id'])}
+
+ _url = "{0}/vnfpkgm/v1/vnf_packages_content".format(self._base_path)
+
+ try:
+ self._create_base_pkg('vnfd', pkg_name)
+ r = requests.post(_url, data=open('/tmp/' + pkg_name + '.tar.gz', 'rb'), verify=False, headers=headers)
+ except Exception as e:
+ log.exception(e)
+ result['data'] = str(e)
+ return result
+ if r.status_code == requests.codes.created:
+ result['data'] = r.json()
+ result['error'] = False
+ if r.status_code == requests.codes.conflict:
+ result['data'] = "Invalid ID."
+ return result
+
+ def nsd_clone(self, token, id):
+ result = {'error': True, 'data': ''}
+ headers = {"Content-Type": "application/gzip", "accept": "application/json",
+ 'Authorization': 'Bearer {}'.format(token['id'])}
+
+ # get the package onboarded
+ tar_pkg = self.get_nsd_pkg(token, id)
+ tarf = tarfile.open(fileobj=tar_pkg)
+ tarf = self._descriptor_clone(tarf, 'nsd')
+ headers['Content-File-MD5'] = self.md5(open('/tmp/' + tarf.getnames()[0] + "_clone.tar.gz", 'rb'))
+
+ _url = "{0}/nsd/v1/ns_descriptors_content/".format(self._base_path)
+
+ try:
+ r = requests.post(_url, data=open('/tmp/' + tarf.getnames()[0] + "_clone.tar.gz", 'rb'), verify=False,
+ headers=headers)
+ except Exception as e:
+ log.exception(e)
+ result['data'] = str(e)
+ return result
+ if r.status_code == requests.codes.created:
+ result['error'] = False
+ if r.status_code == requests.codes.conflict:
+ result['data'] = "Invalid ID."
+
+ return result
+
+ def vnfd_clone(self, token, id):
+ result = {'error': True, 'data': ''}
+ headers = {"Content-Type": "application/gzip", "accept": "application/json",
+ 'Authorization': 'Bearer {}'.format(token['id'])}
+
+ # get the package onboarded
+ tar_pkg = self.get_vnfd_pkg(token, id)
+ tarf = tarfile.open(fileobj=tar_pkg)
+
+ tarf = self._descriptor_clone(tarf, 'vnfd')
+ headers['Content-File-MD5'] = self.md5(open('/tmp/' + tarf.getnames()[0] + "_clone.tar.gz", 'rb'))
+
+ _url = "{0}/vnfpkgm/v1/vnf_packages_content".format(self._base_path)
+
+ try:
+ r = requests.post(_url, data=open('/tmp/' + tarf.getnames()[0] + "_clone.tar.gz", 'rb'), verify=False,
+ headers=headers)
+ except Exception as e:
+ log.exception(e)
+ result['data'] = str(e)
+ return result
+ if r.status_code == requests.codes.created:
+ result['error'] = False
+ if r.status_code == requests.codes.conflict:
+ result['data'] = "Invalid ID."
+
+ return result
+