response = requests.get(self.callback_uri, timeout=2)
if response.status_code != status.HTTP_204_NO_CONTENT:
raise GenericparserException("callbackUri %s returns %s status "
- "code." % (self.callback_uri,
- response.status_code))
+ "code." % (self.callback_uri,
+ response.status_code))
except Exception:
raise GenericparserException("callbackUri %s didn't return 204 status"
- "code." % self.callback_uri)
+ "code." % self.callback_uri)
def fill_resp_data(self, subscription):
subscription_filter = dict()
if const.BASIC in self.authentication.get("authType", '') and \
"paramsBasic" in self.authentication.keys() and \
not is_filter_type_equal(PARAMSBASICKEYS,
- self.authentication.
- get("paramsBasic").keys()):
+ self.authentication.get("paramsBasic").keys()):
raise NsdmBadRequestException('userName and password needed '
'for ' + const.BASIC)
if const.OAUTH2_CLIENT_CREDENTIALS in \
"paramsOauth2ClientCredentials" in \
self.authentication.keys() and \
not is_filter_type_equal(PARAMSOAUTH2CLIENTCREDENTIALSKEYS,
- self.authentication.
- get("paramsOauth2ClientCredentials")
- .keys()):
+ self.authentication.get("paramsOauth2ClientCredentials").keys()):
raise NsdmBadRequestException('clientId, clientPassword and '
'tokenEndpoint required '
'for ' + const.OAUTH2_CLIENT_CREDENTIALS)
links = {
"self": {
"href":
- const.NSDM_SUBSCRIPTION_ROOT_URI + self.subscription_id
+ const.NSDM_SUBSCRIPTION_ROOT_URI + self.subscription_id
}
}
subscription_save_db = {
resp_data = json.loads(response.content)
self.assertEqual({"status": "active"}, resp_data)
- def test_catalog_health_check(self):
- response = self.client.get("/api/genericparser/v1/health_check")
- self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
- resp_data = json.loads(response.content)
- self.assertEqual({"status": "active"}, resp_data)
-
def test_parser_health_check(self):
response = self.client.get("/api/parser/v1/health_check")
self.assertEqual(status.HTTP_200_OK, response.status_code, response.content)
"version": 1,
"name": "VCPE_NS",
"desginer": "ZTE",
- "invariant_id": "vcpe_ns_sff_1"
- }
- }
- }
+ "invariant_id": "vcpe_ns_sff_1"}}
+ }
def tearDown(self):
pass
def test_ns_pkg_distribute_when_ns_exists(self):
NSPackageModel(nsPackageId="1", nsdId="2").save()
resp = self.client.post(
- "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
+ "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual("failed", resp.data["status"])
self.assertEqual(
def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req):
mock_call_req.return_value = [0, "[]", '200']
resp = self.client.post(
- "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
+ "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual("failed", resp.data["status"])
self.assertEqual(
}]), '200']
NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
resp = self.client.post(
- "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
+ "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual("failed", resp.data["status"])
self.assertEqual(
"distributionStatus": "DISTRIBUTED",
}]), '200']
resp = self.client.post(
- "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
+ "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual("failed", resp.data["status"])
self.assertEqual(
VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
resp = self.client.post(
- "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
+ "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual("success", resp.data["status"])
self.assertEqual(
"lastUpdaterUserId": "jh0003"
}]
resp = self.client.post(
- "/api/genericparser/v1/nspackages", {"csarId": "1"}, format='json')
+ "/api/parser/v1/nspackages", {"csarId": "1"}, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual("failed", resp.data["status"])
self.assertEqual(
def test_ns_pkg_normal_delete(self):
NSPackageModel(nsPackageId="8", nsdId="2").save()
- resp = self.client.delete("/api/genericparser/v1/nspackages/8")
+ resp = self.client.delete("/api/parser/v1/nspackages/8")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual("success", resp.data["status"])
self.assertEqual(
nsdVersion="3",
nsPackageUri="14.csar",
nsdModel="").save()
- resp = self.client.get("/api/genericparser/v1/nspackages")
+ resp = self.client.get("/api/parser/v1/nspackages")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
expect_data = [{"csarId": "13",
"packageInfo": {"csarName": "13.csar",
nsdVersion="4",
nsPackageUri="14.csar",
nsdModel="").save()
- resp = self.client.get("/api/genericparser/v1/nspackages/14")
+ resp = self.client.get("/api/parser/v1/nspackages/14")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
expect_data = {
"csarId": "14",
self.assertEqual(expect_data, resp.data)
def test_ns_pkg_get_one_not_found(self):
- resp = self.client.get("/api/genericparser/v1/nspackages/22")
+ resp = self.client.get("/api/parser/v1/nspackages/22")
self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertEqual(
{"error": "Ns package[22] not Found."},
mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
req_data = {"csarId": "18", "inputs": []}
resp = self.client.post(
- "/api/genericparser/v1/parsernsd",
+ "/api/parser/v1/parsernsd",
req_data,
format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
def test_nsd_parse_when_csar_not_exist(self):
req_data = {"csarId": "1", "inputs": []}
resp = self.client.post(
- "/api/genericparser/v1/parsernsd",
+ "/api/parser/v1/parsernsd",
req_data,
format='json')
self.assertEqual(
import mock
import os
-
from django.test import TestCase
from rest_framework import status
from rest_framework.test import APIClient
PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
req_data = {"csarId": "8", "inputs": []}
- resp = self.client.post("/api/genericparser/v1/parserpnfd", req_data, format='json')
+ resp = self.client.post("/api/parser/v1/parserpnfd", req_data, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual({"model": '{"c": "d"}'}, resp.data)
@mock.patch.object(NfDistributeThread, 'run')
def test_nf_pkg_distribute_normal(self, mock_run):
- resp = self.client.post("/api/genericparser/v1/vnfpackages", {
+ resp = self.client.post("/api/parser/v1/vnfpackages", {
"csarId": "1",
"vimIds": ["1"]
}, format='json')
@mock.patch.object(NfPkgDeleteThread, 'run')
def test_nf_pkg_delete_normal(self, mock_run):
- resp = self.client.delete("/api/genericparser/v1/vnfpackages/1")
+ resp = self.client.delete("/api/parser/v1/vnfpackages/1")
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
def test_nf_pkg_normal_delete(self):
vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
VnfPackageModel(vnfPackageId="4", vnfdId="4", vnfVendor='4', vnfdVersion='4',
vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
- resp = self.client.get("/api/genericparser/v1/vnfpackages")
+ resp = self.client.get("/api/parser/v1/vnfpackages")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
expect_data = [
{
VnfPackageModel(vnfPackageId="4", vnfdId="4", vnfVendor='4', vnfdVersion='4',
vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
- resp = self.client.get("/api/genericparser/v1/vnfpackages/4")
+ resp = self.client.get("/api/parser/v1/vnfpackages/4")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
expect_data = {
"imageInfo": [],
VnfPackageModel(vnfPackageId="4", vnfdId="4", vnfVendor='4', vnfdVersion='4',
vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
- resp = self.client.get("/api/genericparser/v1/vnfpackages/2")
+ resp = self.client.get("/api/parser/v1/vnfpackages/2")
self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
VnfPackageModel(vnfPackageId="8", vnfdId="10").save()
mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
req_data = {"csarId": "8", "inputs": []}
- resp = self.client.post("/api/genericparser/v1/parservnfd", req_data, format='json')
+ resp = self.client.post("/api/parser/v1/parservnfd", req_data, format='json')
self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
self.assertEqual({"model": '{"c": "d"}'}, resp.data)
def test_vnfd_parse_when_csar_not_exist(self):
req_data = {"csarId": "1", "inputs": []}
- resp = self.client.post("/api/genericparser/v1/parservnfd", req_data, format='json')
+ resp = self.client.post("/api/parser/v1/parservnfd", req_data, format='json')
self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})
--- /dev/null
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.9 on 2019-04-22 14:42
+from __future__ import unicode_literals
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('database', '0001_initial'),
+ ]
+
+ operations = [
+ migrations.AlterModelTable(
+ name='jobmodel',
+ table='GENERICPARSER_JOB',
+ ),
+ migrations.AlterModelTable(
+ name='jobstatusmodel',
+ table='GENERICPARSER_JOB_STATUS',
+ ),
+ migrations.AlterModelTable(
+ name='nsdmsubscriptionmodel',
+ table='GENERICPARSER_NSDM_SUBSCRIPTION',
+ ),
+ migrations.AlterModelTable(
+ name='nspackagemodel',
+ table='GENERICPARSER_NSPACKAGE',
+ ),
+ migrations.AlterModelTable(
+ name='pnfpackagemodel',
+ table='GENERICPARSER_PNFPACKAGE',
+ ),
+ migrations.AlterModelTable(
+ name='servicepackagemodel',
+ table='GENERICPARSER_SERVICEPACKAGE',
+ ),
+ migrations.AlterModelTable(
+ name='softwareimagemodel',
+ table='GENERICPARSER_SOFTWAREIMAGEMODEL',
+ ),
+ migrations.AlterModelTable(
+ name='vnfpackagemodel',
+ table='GENERICPARSER_VNFPACKAGE',
+ ),
+ ]
from genericparser.pub.config import config as pub_config
from logging import config as log_config
from onaplogging import monkey
+
monkey.patch_all()
# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
'django.contrib.auth.middleware.SessionAuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
- 'genericparser.middleware.LogContextMiddleware',
+ 'genericparser.middleware.LogContextMiddleware',
]
ROOT_URLCONF = 'genericparser.urls'
import mock
from genericparser.pub.utils import idutil
+
idutil.get_auto_id = mock.Mock()
idutil.get_auto_id.return_value = 1
+++ /dev/null
-test
\ No newline at end of file
[testenv]
deps = -r{toxinidir}/requirements.txt
-commands = coverage run --branch manage.py test catalog
+commands = coverage run --branch manage.py test genericparser
[testenv:pep8]
deps = flake8