lcm/ns/tests/sfcs/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         mock_req_by_rest.return_value = [0, '{"test":"test_name","url":"url_add"}']
  53         response = self.client.delete("/api/nslcm/v1/ns/sfcs/%s" % self.sfc_inst_id)
  54         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
  55         expect_resp_data = {"result": 0, "detail": "delete sfc success"}
  56         self.assertEqual(expect_resp_data, response.data)
  57
  58         for vnffg_info in VNFFGInstModel.objects.filter(nsinstid=self.ns_inst_id):
  59             self.assertEqual(vnffg_info.fplist, "test1,test2,test3")
  60         if FPInstModel.objects.filter(fpinstid=self.sfc_inst_id):
  61             self.fail()
  62
  63         response = self.client.delete("/api/nslcm/v1/ns/sfcs/%s" % "notExist")
  64         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
  65         expect_resp_data = {"result": 0, "detail": "sfc is not exist or has been already deleted"}
  66         self.assertEqual(expect_resp_data, response.data)
  67
  68     def test_query_sfc(self):
  69         response = self.client.get("/api/nslcm/v1/ns/sfcs/%s" % self.sfc_inst_id)
  70         self.assertEqual(status.HTTP_200_OK, response.status_code)
  71         expect_resp_data = {'sfcInstId': self.sfc_inst_id,
  72                             'sfcStatus': self.status,
  73                             'sfcName': "xxx"}
  74         self.assertEqual(expect_resp_data, response.data)
  75
  76         response = self.client.get("/api/nslcm/v1/ns/sfcs/%s" % "notExist")
  77         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)