lcm/ns_sfcs/tests/tests.py

   1 # Copyright 2016 ZTE Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import json
  16 import uuid
  17
  18 import mock
  19 from django.test import TestCase, Client
  20 from rest_framework import status
  21
  22 from lcm.pub.database.models import FPInstModel, VNFFGInstModel
  23 from lcm.pub.utils import restcall
  24
  25
  26 class TestSfcDetailViews(TestCase):
  27     def setUp(self):
  28         self.client = Client()
  29         self.ns_inst_id = str(uuid.uuid4())
  30         self.sfc_inst_id = str(uuid.uuid4())
  31         self.status = "active"
  32         self.sdn_controler_id = str(uuid.uuid4())
  33         sfc_id = str(uuid.uuid4())
  34         flow_classifiers = "flow1,flow2"
  35         port_pair_groups = json.JSONEncoder().encode(
  36             [{"groupid": "group1", "portpair": [str(uuid.uuid4()), str(uuid.uuid4())]},
  37              {"groupid": "group2", "portpair": [str(uuid.uuid4()), str(uuid.uuid4())]}])
  38         FPInstModel(fpid="", fpinstid=self.sfc_inst_id, nsinstid=self.ns_inst_id, vnffginstid="", policyinfo="",
  39                     status=self.status, sdncontrollerid=self.sdn_controler_id, sfcid=sfc_id,
  40                     flowclassifiers=flow_classifiers,
  41                     portpairgroups=port_pair_groups).save()
  42         VNFFGInstModel(vnffgdid="", vnffginstid="", nsinstid=self.ns_inst_id,
  43                        fplist="test1," + self.sfc_inst_id + ",test2,test3", endpointnumber=0, cplist="", vnflist="",
  44                        vllist="", status="").save()
  45
  46     def tearDown(self):
  47         FPInstModel.objects.all().delete()
  48         VNFFGInstModel.objects.all().delete()
  49
  50     @mock.patch.object(restcall, "call_req")
  51     def test_delete_sfc(self, mock_req_by_rest):
  52         sdnc_info = {
  53             "thirdparty-sdnc-id": "example-thirdparty-sdnc-id-val-6524",
  54             "location": "example-location-val-78867",
  55             "product-name": "example-product-name-val-15818",
  56             "esr-system-info-list": {
  57                 "esr-system-info": [
  58                     {
  59                         "esr-system-info-id": "example-esr-system-info-id-val-24165",
  60                         "system-name": "example-system-name-val-77122",
  61                         "type": "example-type-val-21280",
  62                         "vendor": "example-vendor-val-91275",
  63                         "version": "example-version-val-93343",
  64                         "service-url": "example-service-url-val-81241",
  65                         "user-name": "example-user-name-val-1481",
  66                         "password": "example-password-val-976",
  67                         "system-type": "example-system-type-val-92280",
  68                         "protocal": "example-protocal-val-40984",
  69                         "ssl-cacert": "example-ssl-cacert-val-48921",
  70                         "ssl-insecure": True,
  71                         "ip-address": "example-ip-address-val-1363",
  72                         "port": "example-port-val-90119",
  73                         "cloud-domain": "example-cloud-domain-val-26113",
  74                         "default-tenant": "example-default-tenant-val-5704"
  75                     }
  76                 ]
  77             }
  78         }
  79         mock_req_by_rest.return_value = [0, json.JSONEncoder().encode(sdnc_info), '200']
  80         # mock_req_by_rest.return_value = [0, '{"test":"test_name","url":"url_add"}']
  81         response = self.client.delete("/api/nslcm/v1/ns/sfcs/%s" % self.sfc_inst_id)
  82         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
  83         expect_resp_data = {"result": 0, "detail": "delete sfc success"}
  84         self.assertEqual(expect_resp_data, response.data)
  85
  86         for vnffg_info in VNFFGInstModel.objects.filter(nsinstid=self.ns_inst_id):
  87             self.assertEqual(vnffg_info.fplist, "test1,test2,test3")
  88         if FPInstModel.objects.filter(fpinstid=self.sfc_inst_id):
  89             self.fail()
  90
  91         response = self.client.delete("/api/nslcm/v1/ns/sfcs/%s" % "notExist")
  92         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
  93         expect_resp_data = {"result": 0, "detail": "sfc is not exist or has been already deleted"}
  94         self.assertEqual(expect_resp_data, response.data)
  95
  96     def test_query_sfc(self):
  97         response = self.client.get("/api/nslcm/v1/ns/sfcs/%s" % self.sfc_inst_id)
  98         self.assertEqual(status.HTTP_200_OK, response.status_code)
  99         expect_resp_data = {'sfcInstId': self.sfc_inst_id,
 100                             'sfcStatus': self.status,
 101                             'sfcName': "xxx"}
 102         self.assertEqual(expect_resp_data, response.data)
 103
 104         response = self.client.get("/api/nslcm/v1/ns/ns_sfcs/%s" % "notExist")
 105         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)