lcm/ns/tests/test_ns_instant.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 from rest_framework import status
  16 from django.test import TestCase
  17 from rest_framework.test import APIClient
  18 import mock
  19 import json
  20
  21 from lcm.pub.database.models import NSInstModel
  22 from lcm.ns.ns_instant import InstantNSService
  23 from lcm.ns.ns_instant import BuildInWorkflowThread
  24 from lcm.pub.utils import restcall
  25
  26
  27 class TestNsInstant(TestCase):
  28     def setUp(self):
  29         self.client = APIClient()
  30         NSInstModel.objects.filter().delete()
  31         self.url = "/api/nslcm/v1/ns/2/instantiate"
  32         self.req_data = {
  33             "additionalParamForNs": {
  34                 "sdnControllerId": "2"
  35             },
  36             "locationConstraints": [{
  37                 "vnfProfileId": "vnfd1",
  38                 "locationConstraints": {
  39                     "vimId": "3"
  40                 }
  41             }]
  42         }
  43         self.nsd_model = json.dumps({
  44             "model": json.dumps({
  45                 "vnfs": [{
  46                     "vnf_id": "vnf1",
  47                     "properties": {
  48                         "id": "vnfd1",
  49                         "nf_type": "xgw"
  50                     },
  51                     "dependencies": [{
  52                         "vl_id": "5"
  53                     }]
  54                 }],
  55                 "vls": [{
  56                     "vl_id": "5",
  57                     "properties": {}
  58                 }]
  59             })
  60         })
  61         self.updated_nsd_model = {
  62             "vnfs": [{
  63                 "dependencies": [{
  64                     "vl_id": "5"
  65                 }],
  66                 "vnf_id": "vnf1",
  67                 "properties": {
  68                     "nf_type": "xgw",
  69                     "id": "vnfd1"
  70                 }
  71             }],
  72             "vls": [{
  73                 "vl_id": "5",
  74                 "properties": {
  75                     "location_info": {
  76                         "vimid": "3"
  77                     }
  78                 }
  79             }]
  80         }
  81         self.vnfms = json.dumps({
  82             "esr-vnfm": [{
  83                 "vnfm-id": "4"
  84             }]
  85         })
  86         self.vnfm = json.dumps({
  87             "type": "xgw",
  88             "vim-id": "3",
  89             "vnfm-id": "4",
  90             "certificate-url": "http://127.0.0.0/ztevnfm/v1/auth",
  91             "esr-system-info-list": {
  92                 "esr-system-info": [{
  93                     "type": "xgw",
  94                     "vendor": "zte",
  95                     "version": "1.0",
  96                     "service-url": "http://127.0.0.0/ztevnfm/v1",
  97                     "user-name": "admin",
  98                     "password": "admin123"
  99                 }]
 100             }
 101         })
 102         NSInstModel(id="2", nspackage_id="7", nsd_id="2", status="active").save()
 103
 104     def tearDown(self):
 105         pass
 106
 107     @mock.patch.object(restcall, 'call_req')
 108     @mock.patch.object(BuildInWorkflowThread, 'run')
 109     def test_ns_instantiate_when_succeed_to_enter_workflow(self, mock_run, mock_call_req):
 110         mock_call_req.side_effect = [
 111             [0, self.nsd_model, '200'],
 112             [0, self.vnfms, '200'],
 113             [0, self.vnfm, '200']
 114         ]
 115         resp = self.client.post(self.url, data=self.req_data, format='json')
 116         self.failUnlessEqual(status.HTTP_200_OK, resp.status_code)
 117         self.assertIn("jobId", resp.data)
 118         upd_nsd_model = NSInstModel.objects.filter(id="2").first().nsd_model
 119         self.assertEqual(self.updated_nsd_model, json.loads(upd_nsd_model))
 120
 121     @mock.patch.object(InstantNSService, 'do_biz')
 122     def test_ns_instantiate_normal(self, mock_do_biz):
 123         mock_do_biz.return_value = dict(data={'jobId': "1"}, status=status.HTTP_200_OK)
 124         resp = self.client.post(self.url, data=self.req_data, format='json')
 125         self.failUnlessEqual(status.HTTP_200_OK, resp.status_code)
 126         self.assertEqual({'jobId': "1"}, resp.data)
 127
 128     @mock.patch.object(restcall, 'call_req')
 129     def test_ns_instantiate_when_fail_to_parse_nsd(self, mock_call_req):
 130         mock_call_req.return_value = [1, "Failed to parse nsd", '500']
 131         resp = self.client.post(self.url, data=self.req_data, format='json')
 132         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 133         self.assertIn("error", resp.data)