vnfdModel=vnfd_json,
localFilePath=local_file_name,
vnfPackageUri=csar_name
- ).save()
+ ).save()
JobUtil.add_job_status(self.job_id, 100, "CSAR(%s) distribute successfully." % self.csar_id)
sdcCsarId=csar_id,
localFilePath=local_file_name,
nsdModel=nsd_json
- ).save()
+ ).save()
return [0, "CSAR(%s) distributed successfully." % csar_id]
resp = self.client.post("/api/catalog/v1/vnfpackages", {
"csarId": "1",
"vimIds": ["1"]
- }, format='json')
+ }, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
def test_nf_pkg_distribute_when_csar_already_exist(self):
@mock.patch.object(sdc, 'download_artifacts')
@mock.patch.object(toscaparser, 'parse_vnfd')
def test_nf_pkg_distribute_when_vnfd_already_exist(self,
- mock_parse_vnfd, mock_download_artifacts, mock_call_req):
+ mock_parse_vnfd, mock_download_artifacts, mock_call_req):
mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
mock_download_artifacts.return_value = "/home/hss.csar"
mock_call_req.return_value = [0, json.JSONEncoder().encode([{
"uuid": "1",
"toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
- }]), '200']
+ }]), '200']
VnfPackageModel(vnfPackageId="2", vnfdId="zte-hss-1.0").save()
NfDistributeThread(csar_id="1",
vim_ids=["1"],
@mock.patch.object(sdc, 'download_artifacts')
@mock.patch.object(toscaparser, 'parse_vnfd')
def test_nf_pkg_distribute_successfully(self,
- mock_parse_vnfd, mock_download_artifacts, mock_call_req):
+ mock_parse_vnfd, mock_download_artifacts, mock_call_req):
mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
mock_download_artifacts.return_value = "/home/hss.csar"
mock_call_req.return_value = [0, json.JSONEncoder().encode([{
"uuid": "1",
"toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
- }]), '200']
+ }]), '200']
NfDistributeThread(csar_id="1",
vim_ids=["1"],
lab_vim_id="",
@mock.patch.object(restcall, 'call_req')
@mock.patch.object(sdc, 'download_artifacts')
@mock.patch.object(toscaparser, 'parse_nsd')
- def test_ns_pkg_distribute_when_nsd_already_exists(self,
- mock_parse_nsd, mock_download_artifacts, mock_call_req):
+ def test_ns_pkg_distribute_when_nsd_already_exists(self,
+ mock_parse_nsd, mock_download_artifacts, mock_call_req):
mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
mock_download_artifacts.return_value = "/home/vcpe.csar"
mock_call_req.return_value = [0, json.JSONEncoder().encode([{
"uuid": "1",
"toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar"
- }]), '200']
+ }]), '200']
NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
resp = self.client.post("/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
@mock.patch.object(restcall, 'call_req')
@mock.patch.object(sdc, 'download_artifacts')
@mock.patch.object(toscaparser, 'parse_nsd')
- def test_ns_pkg_distribute_when_nf_not_distributed(self,
- mock_parse_nsd, mock_download_artifacts, mock_call_req):
+ def test_ns_pkg_distribute_when_nf_not_distributed(self,
+ mock_parse_nsd, mock_download_artifacts, mock_call_req):
mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
mock_download_artifacts.return_value = "/home/vcpe.csar"
mock_call_req.return_value = [0, json.JSONEncoder().encode([{
"uuid": "1",
"toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar"
- }]), '200']
+ }]), '200']
resp = self.client.post("/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual("failed", resp.data["status"])
@mock.patch.object(restcall, 'call_req')
@mock.patch.object(sdc, 'download_artifacts')
@mock.patch.object(toscaparser, 'parse_nsd')
- def test_ns_pkg_distribute_when_successfully(self,
- mock_parse_nsd, mock_download_artifacts, mock_call_req):
+ def test_ns_pkg_distribute_when_successfully(self,
+ mock_parse_nsd, mock_download_artifacts, mock_call_req):
mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
mock_download_artifacts.return_value = "/home/vcpe.csar"
mock_call_req.return_value = [0, json.JSONEncoder().encode([{
"uuid": "1",
"toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar"
- }]), '200']
+ }]), '200']
VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
resp = self.client.post("/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
url(r'^api/catalog/v1/parsernsd$', views.ns_model_parser, name='nsmodelparser_rc'),
url(r'^api/catalog/v1/parservnfd$', views.vnf_model_parser, name='vnfmodelparser_rc'),
]
-
-
SDC_BASE_URL = "https://127.0.0.1:8443/api/sdc/v1"
SDC_USER = "SDC"
SDC_PASSWD = "SDC"
-
-
-
-
-
-
-
logger = logging.getLogger(__name__)
-def call_req(base_url, user, passwd, auth_type, resource, method,
- content='', additional_headers={}):
+def call_req(base_url, user, passwd, auth_type, resource, method, content='', additional_headers={}):
callid = str(uuid.uuid1())
logger.debug("[%s]call_req('%s','%s','%s',%s,'%s','%s','%s')" % (
callid, base_url, user, passwd, auth_type, resource, method, content))
if old_key.upper() == key.upper():
return args[old_key]
return def_val
-
downloadcache = ~/cache/pip
[flake8]
-ignore = E501,E231,E128,W291,W293,E711,E303,E701,E225,E303,E302,E126,E203,F841,E123,E125,W391
+ignore = E501,E231,E128,W291,W293,E711,E303,E701,E225,E303,E302,E126,E203,F841
exclude = ./venv-tox,./.tox
[testenv]