# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+
+
+import os
+from lcm.pub.utils import fileutil
+
+cur_path = os.path.dirname(os.path.abspath(__file__))
+CREATE_NS_DICT = fileutil.read_json_file(cur_path + '/data/create_ns.json')
+NS_INFO_AAI_DICT = fileutil.read_json_file(cur_path + '/data/ns_info_aai.json')
+VNFD_MODEL_DICT = fileutil.read_json_file(cur_path + '/data/vnfd_model.json')
+HEAL_NS_DICT = fileutil.read_json_file(cur_path + '/data/heal_ns.json')
+HEAL_VNF_DICT = fileutil.read_json_file(cur_path + '/data/heal_vnf.json')
+NSD_MODEL_DICT = fileutil.read_json_file(cur_path + '/data/nsd_model.json')
+NSD_MODEL_WITH_PNF_DICT = fileutil.read_json_file(cur_path + '/data/nsd_model_with_pnf.json')
+VNFM_LIST_IN_AAI_DICT = fileutil.read_json_file(cur_path + '/data/vnfm_list_in_aai.json')
+VNFM_IN_AAI_DICT = fileutil.read_json_file(cur_path + '/data/vnfm_in_aai.json')
+JOB_DICT = fileutil.read_json_file(cur_path + '/data/job.json')
+INSTANTIATE_NS_WITH_PNF_DICT = fileutil.read_json_file(cur_path + '/data/instantiate_ns_with_pnf.json')
+INSTANTIATE_NS_DICT = fileutil.read_json_file(cur_path + '/data/instantiate_ns.json')
+SCALING_MAP_DICT = fileutil.read_json_file(cur_path + '/data/scalemapping.json')
+SCALE_NS_DICT = fileutil.read_json_file(cur_path + '/data/scale_ns.json')
+OCCURRENCE_DICT = fileutil.read_json_file(cur_path + '/data/occurrence.json')
+NSLCMOP_WITH_EXCLUDE_DEFAULT_DICT = fileutil.read_json_file(cur_path + '/data/occurrence_exclude_default.json')
+SUBSCRIPTION_DICT = fileutil.read_json_file(cur_path + '/data/subscription.json')
# limitations under the License.
import json
import mock
-import os
import uuid
from django.test import TestCase
from lcm.pub.database.models import NSInstModel
from lcm.ns.biz.ns_create import CreateNSService
-from lcm.pub.utils import restcall, fileutil
+from lcm.pub.utils import restcall
from rest_framework import status
from rest_framework.test import APIClient
+from lcm.ns.tests import CREATE_NS_DICT
class TestNsInstantiate(TestCase):
def setUp(self):
self.client = APIClient()
- self.cur_path = os.path.dirname(os.path.abspath(__file__))
- self.create_ns_json = fileutil.read_json_file(self.cur_path + '/data/create_ns.json')
+ self.create_ns_dict = CREATE_NS_DICT
def tearDown(self):
NSInstModel.objects.all().delete()
r1_query_nspackage_from_catalog = [0, json.JSONEncoder().encode(nspackage_info), '201']
r2_create_ns_to_aai = [0, json.JSONEncoder().encode({}), '201']
mock_call_req.side_effect = [r1_query_nspackage_from_catalog, r2_create_ns_to_aai]
- self.create_ns_json["csarId"] = str(uuid.uuid4())
- response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_json, format='json')
+ self.create_ns_dict["csarId"] = str(uuid.uuid4())
+ response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_dict, format='json')
self.failUnlessEqual(status.HTTP_201_CREATED, response.status_code)
@mock.patch.object(CreateNSService, "do_biz")
@mock.patch.object(CreateNSService, "do_biz")
def test_create_ns_non_existing_nsd(self, mock_do_biz):
mock_do_biz.side_effect = Exception("nsd not exists.")
- self.create_ns_json["csarId"] = "1"
- response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_json, format='json')
+ self.create_ns_dict["csarId"] = "1"
+ response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_dict, format='json')
self.assertEqual(response.data["error"], "nsd not exists.")
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertIn("error", response.data)
@mock.patch.object(restcall, 'call_req')
def test_create_ns_when_fail_to_get_nsd(self, mock_call_req):
mock_call_req.return_value = [1, "Failed to get nsd.", '500']
- self.create_ns_json["csarId"] = "1"
- response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_json, format='json')
+ self.create_ns_dict["csarId"] = "1"
+ response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_dict, format='json')
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertIn("error", response.data)
})
mock_call_req.return_value = [0, nspackage_info, '200']
NSInstModel(id="1", name="ns").save()
- self.create_ns_json["csarId"] = "1"
- response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_json, format='json')
+ self.create_ns_dict["csarId"] = "1"
+ response = self.client.post("/api/nslcm/v1/ns", data=self.create_ns_dict, format='json')
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertIn("error", response.data)
import json
import mock
-import os
import uuid
from django.test import TestCase, Client
from lcm.pub.database.models import NSInstModel
-from lcm.pub.utils import restcall, fileutil
+from lcm.pub.utils import restcall
from rest_framework import status
+from lcm.ns.tests import NS_INFO_AAI_DICT
class TestNsDelelete(TestCase):
def setUp(self):
self.client = Client()
self.ns_inst_id = str(uuid.uuid1())
- self.cur_path = os.path.dirname(os.path.abspath(__file__))
NSInstModel.objects.filter().delete()
NSInstModel(id=self.ns_inst_id, nspackage_id="7", nsd_id="2").save()
@mock.patch.object(restcall, 'call_req')
def test_delete_ns(self, mock_call_req):
- ns_info_aai = fileutil.read_json_file(self.cur_path + '/data/ns_info_aai.json')
- r1_query_ns_to_aai = [0, json.JSONEncoder().encode(ns_info_aai), '200']
+ r1_query_ns_to_aai = [0, json.JSONEncoder().encode(NS_INFO_AAI_DICT), '200']
r2_delete_ns_to_aai = [0, json.JSONEncoder().encode({}), '200']
mock_call_req.side_effect = [r1_query_ns_to_aai, r2_delete_ns_to_aai]
response = self.client.delete("/api/nslcm/v1/ns/%s" % self.ns_inst_id)
# limitations under the License.
from django.test import TestCase
-import json
import mock
-import os
from rest_framework.test import APIClient
from rest_framework import status
from lcm.ns.biz.ns_heal import NSHealService
from lcm.pub.database.models import NSInstModel, NfInstModel
from lcm.pub.exceptions import NSLCMException
from lcm.pub.utils.jobutil import JobUtil, JOB_TYPE
-from lcm.pub.utils import fileutil
from lcm.ns_vnfs.biz.heal_vnfs import NFHealService
+from lcm.ns.tests import HEAL_NS_DICT, HEAL_VNF_DICT, VNFD_MODEL_DICT
class TestHealNsViews(TestCase):
def setUp(self):
- self.cur_path = os.path.dirname(os.path.abspath(__file__))
self.ns_inst_id = '1'
self.nf_inst_id = '1'
self.nf_uuid = '1-1-1'
self.job_id = JobUtil.create_job("NS", JOB_TYPE.HEAL_VNF, self.ns_inst_id)
self.client = APIClient()
- model = json.dumps(fileutil.read_json_file(self.cur_path + '/data/vnfd_model.json'))
NSInstModel.objects.filter().delete()
NfInstModel.objects.filter().delete()
NSInstModel(id=self.ns_inst_id, name="ns_name", status='null').save()
status='null',
mnfinstid=self.nf_uuid,
package_id='pkg1',
- vnfd_model=model)
+ vnfd_model=VNFD_MODEL_DICT)
def tearDown(self):
pass
@mock.patch.object(NSHealService, 'run')
def test_heal_vnf_url(self, mock_run):
- heal_vnf_json = fileutil.read_json_file(self.cur_path + '/data/heal_vnf.json')
+ heal_vnf_json = HEAL_VNF_DICT.copy()
heal_vnf_json["healVnfData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post("/api/nslcm/v1/ns/%s/heal" % self.ns_inst_id, data=heal_vnf_json, format='json')
self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.data)
# add healNsData
@mock.patch.object(NSHealService, 'run')
def test_heal_ns_url(self, mock_run):
- heal_ns_json = fileutil.read_json_file(self.cur_path + '/data/heal_ns.json')
+ heal_ns_json = HEAL_NS_DICT.copy()
heal_ns_json["healNsData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post("/api/nslcm/v1/ns/%s/heal" % self.ns_inst_id, data=heal_ns_json, format='json')
self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.data)
@mock.patch.object(NSHealService, 'wait_job_finish')
@mock.patch.object(NSHealService, 'update_job')
def test_heal_vnf_thread(self, mock_start, mock_wait, mock_update):
- heal_vnf_json = fileutil.read_json_file(self.cur_path + '/data/heal_vnf.json')
+ heal_vnf_json = HEAL_VNF_DICT.copy()
heal_vnf_json["healVnfData"]["vnfInstanceId"] = self.nf_inst_id
NSHealService(self.ns_inst_id, heal_vnf_json, self.job_id).run()
self.assertEqual(NSInstModel.objects.get(id=self.ns_inst_id).status, NS_INST_STATUS.HEALING)
@mock.patch.object(NSHealService, 'wait_job_finish')
@mock.patch.object(NSHealService, 'update_job')
def test_heal_ns_thread(self, mock_start, mock_wait, mock_update):
- heal_ns_json = fileutil.read_json_file(self.cur_path + '/data/heal_ns.json')
+ heal_ns_json = HEAL_NS_DICT.copy()
heal_ns_json["healNsData"]["vnfInstanceId"] = self.nf_inst_id
NSHealService(self.ns_inst_id, heal_ns_json, self.job_id).run()
self.assertEqual(NSInstModel.objects.get(id=self.ns_inst_id).status, NS_INST_STATUS.HEALING)
def test_heal_vnf_non_existing_ns(self, mock_start):
mock_start.side_effect = NSLCMException("NS Not Found")
ns_inst_id = "2"
- heal_vnf_json = fileutil.read_json_file(self.cur_path + '/data/heal_vnf.json')
+ heal_vnf_json = HEAL_VNF_DICT.copy()
heal_vnf_json["healVnfData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post("/api/nslcm/v1/ns/%s/heal" % ns_inst_id, data=heal_vnf_json, format='json')
self.assertEqual(response.data["error"], "NS Not Found")
def test_heal_ns_heal_non_existing_ns(self, mock_start):
mock_start.side_effect = NSLCMException("NS Not Found")
ns_inst_id = "2"
- heal_ns_json = fileutil.read_json_file(self.cur_path + '/data/heal_ns.json')
+ heal_ns_json = HEAL_NS_DICT.copy()
heal_ns_json["healNsData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post("/api/nslcm/v1/ns/%s/heal" % ns_inst_id, data=heal_ns_json, format='json')
self.assertEqual(response.data["error"], "NS Not Found")
import json
import mock
-import os
from mock import MagicMock
from django.test import TestCase
from rest_framework import status
from lcm.ns.biz.ns_instant import BuildInWorkflowThread
from lcm.ns.biz.ns_instant import InstantNSService
from lcm.pub.database.models import NSInstModel
-from lcm.pub.utils import restcall, fileutil
+from lcm.pub.utils import restcall
from lcm.pub.config import config
+from lcm.ns.tests import NSD_MODEL_DICT, NSD_MODEL_WITH_PNF_DICT, VNFM_LIST_IN_AAI_DICT, VNFM_IN_AAI_DICT, JOB_DICT, INSTANTIATE_NS_DICT, INSTANTIATE_NS_WITH_PNF_DICT
class TestNsInstant(TestCase):
- cur_path = os.path.dirname(os.path.abspath(__file__))
- nsd_model = fileutil.read_json_file(cur_path + '/data/nsd_model.json')
- nsd_model_json = json.dumps({"model": json.dumps(nsd_model)})
- nsd_model_with_pnf_json = json.dumps(fileutil.read_json_file(cur_path + '/data/nsd_model_with_pnf.json'))
- vnfm_list_in_aai = fileutil.read_json_file(cur_path + '/data/vnfm_list_in_aai.json')
- vnfm_in_aai = fileutil.read_json_file(cur_path + '/data/vnfm_in_aai.json')
- job = fileutil.read_json_file(cur_path + '/data/job.json')
- instantiate_ns_with_pnf = fileutil.read_json_file(cur_path + '/data/instantiate_ns_with_pnf.json')
- instantiate_ns_json = fileutil.read_json_file(cur_path + '/data/instantiate_ns.json')
vnfminfo = {"vnfmId": "1"}
def setUp(self):
pass
@mock.patch.object(restcall, 'call_req')
- @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd_model_json))
+ @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=json.dumps({"model": json.dumps(NSD_MODEL_DICT)})))
@mock.patch.object(BuildInWorkflowThread, 'run')
def test_ns_instantiate_when_succeed_to_enter_workflow(self, mock_run, mock_call_req):
config.WORKFLOW_OPTION = "buildin"
mock_call_req.side_effect = [
- [0, TestNsInstant.nsd_model_json, '200'],
- [0, TestNsInstant.vnfm_list_in_aai, '200'],
- [0, TestNsInstant.vnfm_in_aai, '200']
+ [0, {"model": json.dumps(NSD_MODEL_DICT)}, '200'],
+ [0, VNFM_LIST_IN_AAI_DICT, '200'],
+ [0, VNFM_IN_AAI_DICT, '200']
]
- instantiate_ns_json = fileutil.read_json_file(self.cur_path + '/data/instantiate_ns.json')
- resp = self.client.post(self.url, data=instantiate_ns_json, format='json')
+ resp = self.client.post(self.url, data=INSTANTIATE_NS_DICT, format='json')
self.assertEqual(status.HTTP_200_OK, resp.status_code)
self.assertIn("jobId", resp.data)
@mock.patch.object(InstantNSService, 'do_biz')
def test_ns_instantiate_normal(self, mock_do_biz):
- mock_do_biz.return_value = dict(data=TestNsInstant.job, status=status.HTTP_200_OK)
- instantiate_ns_json = fileutil.read_json_file(self.cur_path + '/data/instantiate_ns.json')
- resp = self.client.post(self.url, data=instantiate_ns_json, format='json')
+ mock_do_biz.return_value = dict(data=JOB_DICT, status=status.HTTP_200_OK)
+ resp = self.client.post(self.url, data=INSTANTIATE_NS_DICT, format='json')
self.failUnlessEqual(status.HTTP_200_OK, resp.status_code)
- self.assertEqual(TestNsInstant.job, resp.data)
+ self.assertEqual(JOB_DICT, resp.data)
@mock.patch.object(restcall, 'call_req')
def test_ns_instantiate_when_fail_to_parse_nsd(self, mock_call_req):
mock_call_req.return_value = [1, "Failed to parse nsd", '500']
- resp = self.client.post(self.url, data=TestNsInstant.instantiate_ns_json, format='json')
+ resp = self.client.post(self.url, data=INSTANTIATE_NS_DICT, format='json')
self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertIn("error", resp.data)
@mock.patch('lcm.ns.biz.ns_instantiate_flow.post_deal')
@mock.patch.object(restcall, 'call_req')
@mock.patch('lcm.ns.biz.ns_instantiate_flow.update_job')
- @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd_model_with_pnf_json))
+ @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=json.dumps(NSD_MODEL_WITH_PNF_DICT)))
@mock.patch('lcm.pub.msapi.extsys.select_vnfm', MagicMock(return_value=vnfminfo))
def test_ns_instantiate_with_pnf(self, mock_updata_job, mock_call_req, mock_post_deal):
config.WORKFLOW_OPTION = "grapflow"
NSInstModel(id="1", name="test_ns", nspackage_id="1", status="created").save()
- ret = [0, json.JSONEncoder().encode(TestNsInstant.job), '200']
+ ret = [0, json.JSONEncoder().encode(JOB_DICT), '200']
mock_call_req.side_effect = [ret for i in range(1, 20)]
- ack = InstantNSService(1, TestNsInstant.instantiate_ns_with_pnf).do_biz()
+ ack = InstantNSService(1, INSTANTIATE_NS_WITH_PNF_DICT).do_biz()
self.assertEqual(ack['status'], status.HTTP_200_OK)
@mock.patch.object(restcall, 'call_req')
- @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd_model_with_pnf_json))
+ @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=json.dumps(NSD_MODEL_WITH_PNF_DICT)))
@mock.patch('lcm.pub.msapi.extsys.select_vnfm', MagicMock(return_value=vnfminfo))
def test_ns_instantiate_with_vimid_1(self, mock_call_req):
config.WORKFLOW_OPTION = "grapflow"
NSInstModel(id="1", name="test_ns", nspackage_id="1", status="created").save()
- ret = [0, json.JSONEncoder().encode(TestNsInstant.job), '200']
+ ret = [0, json.JSONEncoder().encode(JOB_DICT), '200']
mock_call_req.side_effect = [ret for i in range(1, 20)]
- ack = InstantNSService(1, TestNsInstant.instantiate_ns_with_pnf).do_biz()
+ ack = InstantNSService(1, INSTANTIATE_NS_WITH_PNF_DICT).do_biz()
self.assertEqual(ack['status'], status.HTTP_200_OK)
@mock.patch.object(restcall, 'call_req')
- @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd_model_with_pnf_json))
+ @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=json.dumps(NSD_MODEL_WITH_PNF_DICT)))
@mock.patch('lcm.pub.msapi.extsys.select_vnfm', MagicMock(return_value=vnfminfo))
def test_ns_instantiate_with_different_vimid_2(self, mock_call_req):
config.WORKFLOW_OPTION = "grapflow"
NSInstModel(id="1", name="test_ns", nspackage_id="1", status="created").save()
- ret = [0, json.JSONEncoder().encode(TestNsInstant.job), '200']
+ ret = [0, json.JSONEncoder().encode(JOB_DICT), '200']
mock_call_req.side_effect = [ret for i in range(1, 20)]
- ack = InstantNSService(1, TestNsInstant.instantiate_ns_with_pnf).do_biz()
+ ack = InstantNSService(1, INSTANTIATE_NS_WITH_PNF_DICT).do_biz()
self.assertEqual(ack['status'], status.HTTP_200_OK)
# limitations under the License.
import json
-import os
import uuid
import mock
from lcm.pub.database.models import NSInstModel, JobModel, NfInstModel
from lcm.pub.exceptions import NSLCMException
from lcm.pub.msapi import catalog
-from lcm.pub.utils import restcall, fileutil
+from lcm.pub.utils import restcall
from lcm.pub.utils.jobutil import JobUtil, JOB_TYPE, JOB_MODEL_STATUS
+from lcm.ns.tests import SCALING_MAP_DICT, VNFD_MODEL_DICT, SCALE_NS_DICT
class TestNsManualScale(TestCase):
def setUp(self):
- self.cur_path = os.path.dirname(os.path.abspath(__file__))
- self.scaling_map_json = fileutil.read_json_file(self.cur_path + '/data/scalemapping.json')
+ self.scaling_map_json = SCALING_MAP_DICT
self.ns_inst_id = str(uuid.uuid4())
self.job_id = JobUtil.create_job("NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
self.client = APIClient()
def insert_new_nf(self):
self.vnfm_inst_id = "1"
- vnfd_model_json = fileutil.read_json_file(self.cur_path + '/data/vnfd_model.json')
NfInstModel.objects.create(
nfinstid="233",
nf_name="name_1",
status='active',
mnfinstid="ab34-3g5j-de13-ab85-ij93",
package_id="nf_zte_hss",
- vnfd_model=json.dumps(vnfd_model_json))
+ vnfd_model=json.dumps(VNFD_MODEL_DICT))
@mock.patch.object(NSManualScaleService, 'run')
def test_ns_manual_scale(self, mock_run):
- scale_ns_json = fileutil.read_json_file(self.cur_path + '/data/scale_ns.json')
- response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data=scale_ns_json, format='json')
+ response = self.client.post("/api/nslcm/v1/ns/%s/scale" % self.ns_inst_id, data=SCALE_NS_DICT, format='json')
self.failUnlessEqual(status.HTTP_202_ACCEPTED, response.status_code)
def test_ns_manual_scale_error_scaletype(self):
- scale_ns_json = fileutil.read_json_file(self.cur_path + '/data/scale_ns.json')
+ scale_ns_json = SCALE_NS_DICT.copy()
scale_ns_json["scaleType"] = "SCALE_ERR"
NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
jobs = JobModel.objects.filter(jobid=self.job_id)
self.assertEqual(255, jobs[0].progress)
def test_ns_manual_scale_error_nsd_id(self):
- scale_ns_json = fileutil.read_json_file(self.cur_path + '/data/scale_ns.json')
+ scale_ns_json = SCALE_NS_DICT.copy()
scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
jobs = JobModel.objects.filter(jobid=self.job_id)
self.assertEqual(255, jobs[0].progress)
def test_ns_manual_scale_error_aspect(self):
- scale_ns_json = fileutil.read_json_file(self.cur_path + '/data/scale_ns.json')
+ scale_ns_json = SCALE_NS_DICT.copy()
scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "sss_zte"
ns_inst_id, job_id = self.insert_new_ns()
job_id = JobUtil.create_job("NS", JOB_TYPE.MANUAL_SCALE_VNF, ns_inst_id)
@mock.patch.object(catalog, 'get_scalingmap_json_package')
@mock.patch.object(NSManualScaleService, 'do_vnfs_scale')
def test_ns_manual_scale_success(self, mock_do_vnfs_scale, mock_get_scalingmap_json_package):
- scale_ns_json = fileutil.read_json_file(self.cur_path + '/data/scale_ns.json')
+ scale_ns_json = SCALE_NS_DICT.copy()
scale_ns_json["scaleNsData"][0]["scaleNsByStepsData"][0]["aspectId"] = "TIC_EDGE_IMS"
mock_get_scalingmap_json_package.return_value = self.scaling_map_json
mock_do_vnfs_scale.return_value = JOB_MODEL_STATUS.FINISHED
@mock.patch.object(restcall, 'call_req')
def test_ns_manual_scale_thread(self, mock_call):
- scale_ns_json = fileutil.read_json_file(self.cur_path + '/data/scale_ns.json')
+ scale_ns_json = SCALE_NS_DICT.copy()
NSManualScaleService(self.ns_inst_id, scale_ns_json, self.job_id).run()
self.assertTrue(NSInstModel.objects.get(id=self.ns_inst_id).status, NS_INST_STATUS.ACTIVE)
@mock.patch.object(NSManualScaleService, 'start')
def test_ns_manual_scale_when_ns_not_exist(self, mock_start):
mock_start.side_effect = NSLCMException("NS scale failed.")
- scale_ns_json = fileutil.read_json_file(self.cur_path + '/data/scale_ns.json')
+ scale_ns_json = SCALE_NS_DICT.copy()
response = self.client.post("/api/nslcm/v1/ns/11/scale", data=scale_ns_json, format='json')
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertIn("error", response.data)
# limitations under the License.
import json
-import os
from rest_framework.test import APIClient
from django.test import TestCase
from rest_framework import status
from lcm.pub.database.models import NSLcmOpOccModel
-from lcm.pub.utils import fileutil
+from lcm.ns.tests import OCCURRENCE_DICT, NSLCMOP_WITH_EXCLUDE_DEFAULT_DICT
class TestNSLcmOpOccs(TestCase):
def setUp(self):
self.client = APIClient()
- self.cur_path = os.path.dirname(os.path.abspath(__file__))
- self.multiple_ns_lcm_op = fileutil.read_json_file(self.cur_path + '/data/occurrence.json')
+ self.multiple_ns_lcm_op = OCCURRENCE_DICT
self.single_ns_lcm_op = self.multiple_ns_lcm_op[0]
- self.nslcmop_with_exclude_default = fileutil.read_json_file(self.cur_path + '/data/occurrence_exclude_default.json')
+ self.nslcmop_with_exclude_default = NSLCMOP_WITH_EXCLUDE_DEFAULT_DICT
NSLcmOpOccModel.objects.all().delete()
def tearDown(self):
# limitations under the License.
import json
-import os
from django.test import TestCase
-from lcm.pub.utils import fileutil
from rest_framework import status
from rest_framework.test import APIClient
from lcm.pub.database.models import SubscriptionModel
+from lcm.ns.tests import SUBSCRIPTION_DICT
class TestQuerySubscriptions(TestCase):
def setUp(self):
self.client = APIClient()
- self.cur_path = os.path.dirname(os.path.abspath(__file__))
- self.test_single_subscription = fileutil.read_json_file(self.cur_path + '/data/subscription.json')
+ self.test_single_subscription = SUBSCRIPTION_DICT.copy()
self.subscription_id = "99442b18-a5c7-11e8-998c-bf1755941f16"
self.ns_instance_id = "cd552c9c-ab6f-11e8-b354-236c32aa91a1"
SubscriptionModel.objects.all().delete()
# limitations under the License.
import json
-import os
from django.test import TestCase
from lcm.pub.database.models import NSInstModel
from lcm.pub.database.models import NfInstModel
from lcm.pub.utils.timeutil import now_time
-from lcm.pub.utils import fileutil
+from lcm.ns.tests import VNFD_MODEL_DICT
class TestScaleAspect(TestCase):
def setUp(self):
- self.cur_path = os.path.dirname(os.path.abspath(__file__))
- self.scaling_map_json = fileutil.read_json_file(self.cur_path + '/data/scalemapping.json')
- self.vnfd_model_json = fileutil.read_json_file(self.cur_path + '/data/vnfd_model.json')
self.initInstModel()
self.init_scale_ns_data()
status='active',
mnfinstid=self.nf_uuid,
package_id=self.package_id,
- vnfd_model=json.dumps(self.vnfd_model_json))
+ vnfd_model=json.dumps(VNFD_MODEL_DICT))
# Create a second vnf instance
self.nf_inst_id = "232"
status='active',
mnfinstid=self.nf_uuid,
package_id=self.package_id,
- vnfd_model=json.dumps(self.vnfd_model_json))
+ vnfd_model=json.dumps(VNFD_MODEL_DICT))
def add_another_nf_instance(self):
# Create a third vnf instance
status='active',
mnfinstid=nf_uuid,
package_id=package_id,
- vnfd_model=json.dumps(self.vnfd_model_json))
+ vnfd_model=json.dumps(VNFD_MODEL_DICT))
def add_new_vnf_instance(self):
# Create a third vnf instance
status='active',
mnfinstid=nf_uuid,
package_id=package_id,
- vnfd_model=json.dumps(self.vnfd_model_json))
+ vnfd_model=json.dumps(VNFD_MODEL_DICT))
# Create a third vnf instance
nf_inst_id = "242"
status='active',
mnfinstid=nf_uuid,
package_id=package_id,
- vnfd_model=json.dumps(self.vnfd_model_json))
+ vnfd_model=json.dumps(VNFD_MODEL_DICT))
def tearDown(self):
NSInstModel().clean()
from lcm.pub.database.models import NSInstModel, NfInstModel
from lcm.pub.exceptions import NSLCMException
from lcm.pub.utils.jobutil import JobUtil, JOB_TYPE
+from lcm.ns.tests import VNFD_MODEL_DICT, HEAL_NS_DICT, HEAL_VNF_DICT
class TestHealNsApi(TestCase):
def setUp(self):
self.url = "/api/nslcm/v1/ns_instances/%s/heal"
- # self.ns_inst_id = '1'
self.ns_inst_id = str(uuid.uuid4())
self.nf_inst_id = '1'
self.nf_uuid = '1-1-1'
-
self.job_id = JobUtil.create_job("NS", JOB_TYPE.HEAL_VNF, self.ns_inst_id)
-
self.client = Client()
-
- model = json.dumps({
- "metadata": {
- "vnfdId": "1",
- "vnfdName": "PGW001",
- "vnfProvider": "zte",
- "vnfdVersion": "V00001",
- "vnfVersion": "V5.10.20",
- "productType": "CN",
- "vnfType": "PGW",
- "description": "PGW VNFD description",
- "isShared": True,
- "vnfExtendType": "driver"
- }
- })
+ model = json.dumps(VNFD_MODEL_DICT)
NSInstModel.objects.filter().delete()
NfInstModel.objects.filter().delete()
NSInstModel(id=self.ns_inst_id, name="ns_name", status='null').save()
- NfInstModel.objects.create(nfinstid=self.nf_inst_id,
- nf_name='name_1',
- vnf_id='1',
- vnfm_inst_id='1',
- ns_inst_id=self.ns_inst_id,
- max_cpu='14',
- max_ram='12296',
- max_hd='101',
- max_shd="20",
- max_net=10,
- status='null',
- mnfinstid=self.nf_uuid,
- package_id='pkg1',
- vnfd_model=model)
+ NfInstModel.objects.create(
+ nfinstid=self.nf_inst_id,
+ nf_name='name_1',
+ vnf_id='1',
+ vnfm_inst_id='1',
+ ns_inst_id=self.ns_inst_id,
+ max_cpu='14',
+ max_ram='12296',
+ max_hd='101',
+ max_shd="20",
+ max_net=10,
+ status='null',
+ mnfinstid=self.nf_uuid,
+ package_id='pkg1',
+ vnfd_model=model)
def tearDown(self):
pass
@mock.patch.object(NSHealService, 'run')
def test_heal_vnf_url(self, mock_run):
-
- data = {
- "healVnfData": [{
- "vnfInstanceId": self.nf_inst_id,
- "cause": "vm is down",
- "additionalParams": {
- "action": "restartvm",
- "actionvminfo": {
- "vmid": "33",
- "vduid": "",
- "vmname": "xgw-smp11"
- }
- }
- }]
- }
-
+ data = HEAL_VNF_DICT.copy()
+ data["healVnfData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post(self.url % self.ns_inst_id, data=data)
self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.data)
self.assertIsNotNone(response.data)
@mock.patch.object(NSHealService, 'run')
def test_heal_ns_url(self, mock_run):
-
- data = {
- "healNsData": {
- "vnfInstanceId": self.nf_inst_id,
- "cause": "",
- "additionalParams": {
- "action": "vmreset",
- "actionvminfo": {
- "vmid": "33",
- "vduid": "",
- "vmname": "xgw-smp11"
- }
- }
- }
- }
-
+ data = HEAL_NS_DICT.copy()
+ data["healNsData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post(self.url % self.ns_inst_id, data=data)
self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.data)
self.assertIsNotNone(response['Location'])
@mock.patch.object(NSHealService, "start")
def test_heal_vnf_non_existing_ns(self, mock_start):
mock_start.side_effect = NSLCMException("NS Not Found")
-
ns_inst_id = "2"
-
- data = {
- "healVnfData": [{
- "vnfInstanceId": self.nf_inst_id,
- "cause": "vm is down",
- "additionalParams": {
- "action": "restartvm",
- "actionvminfo": {
- "vmid": "33",
- "vmname": "xgw-smp11"
- }
- }
- }]
- }
-
+ data = HEAL_VNF_DICT.copy()
+ data["healVnfData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post(self.url % ns_inst_id, data=data)
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
@mock.patch.object(NSHealService, "start")
def test_heal_ns_heal_non_existing_ns(self, mock_start):
mock_start.side_effect = NSLCMException("NS Not Found")
-
ns_inst_id = "2"
-
- data = {
- "healNsData": {
- "vnfInstanceId": self.nf_inst_id,
- "cause": "",
- "additionalParams": {
- "action": "vmreset",
- "actionvminfo": {
- "vmid": "33",
- "vduid": "",
- "vmname": "xgw-smp11"
- }
- }
- }
- }
-
+ data = HEAL_NS_DICT.copy()
+ data["healNsData"]["vnfInstanceId"] = self.nf_inst_id
response = self.client.post(self.url % ns_inst_id, data=data)
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
Code Review / vfc / nfvo / lcm.git / commitdiff