lcm/ns/tests/test_sol_ns_instances_api.py

   1 # Copyright 2019 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
  20 from rest_framework import status
  21 from rest_framework.test import APIClient
  22 from lcm.pub.database.models import NSInstModel
  23 from lcm.pub.utils import restcall
  24
  25
  26 class TestNsInstanceApi(TestCase):
  27
  28     def setUp(self):
  29         self.apiClient = APIClient()
  30         self.format = 'json'
  31         self.ns_instances_url = '/api/nslcm/v1/ns_instances'
  32         self.nsd_id = str(uuid.uuid4())
  33         self.ns_package_id = str(uuid.uuid4())
  34
  35     def tearDown(self):
  36         NSInstModel.objects.all().delete()
  37
  38     @mock.patch.object(restcall, 'call_req')
  39     def test_create_ns(self, mock_call_req):
  40         nspackage_info = {
  41             "csarId": self.ns_package_id,
  42             "packageInfo": {
  43                 "nsPackageId": self.ns_package_id,
  44                 "nsdId": self.nsd_id
  45             }
  46         }
  47         r1_query_nspackage_from_catalog = [0, json.JSONEncoder().encode(nspackage_info), '201']
  48         r2_create_ns_to_aai = [0, json.JSONEncoder().encode({}), '201']
  49         mock_call_req.side_effect = [r1_query_nspackage_from_catalog, r2_create_ns_to_aai]
  50
  51         header = {
  52             'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
  53             'HTTP_SERVICETYPE': 'service-type-test1'
  54         }
  55
  56         data = {
  57             "nsdId": self.nsd_id,
  58             "nsName": "ns",
  59             "nsDescription": "description"
  60         }
  61         response = self.apiClient.post(self.ns_instances_url, data=data, format=self.format, **header)
  62         self.failUnlessEqual(status.HTTP_201_CREATED, response.status_code)
  63         return response.data['id']
  64
  65     def test_ns_instances_method_not_allowed(self):
  66         header = {
  67             'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
  68             'HTTP_SERVICETYPE': 'service-type-test1'
  69         }
  70
  71         data = {
  72             "nsdId": self.nsd_id,
  73             "nsName": "ns",
  74             "nsDescription": "description"
  75         }
  76         response = self.apiClient.delete(self.ns_instances_url, data=data, format=self.format, **header)
  77         self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
  78         response = self.apiClient.put(self.ns_instances_url, data=data, format=self.format, **header)
  79         self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
  80         response = self.apiClient.patch(self.ns_instances_url, data=data, format=self.format, **header)
  81         self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
  82
  83     def test_invidual_ns_instance_method_not_allowed(self):
  84         header = {
  85             'HTTP_GLOBALCUSTOMERID': 'global-customer-id-test1',
  86             'HTTP_SERVICETYPE': 'service-type-test1'
  87         }
  88
  89         data = {
  90             "nsdId": self.nsd_id,
  91             "nsName": "ns",
  92             "nsDescription": "description"
  93         }
  94         url = self.ns_instances_url + '/1'
  95         response = self.apiClient.post(url, data=data, format=self.format, **header)
  96         self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
  97         response = self.apiClient.put(url, data=data, format=self.format, **header)
  98         self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
  99         response = self.apiClient.patch(url, data=data, format=self.format, **header)
 100         self.failUnlessEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
 101
 102     def test_query_ns(self):
 103         NSInstModel.objects.all().delete()
 104         self.test_create_ns()
 105         response = self.apiClient.get(self.ns_instances_url)
 106         self.failUnlessEqual(status.HTTP_200_OK, response.status_code, response.data)
 107         self.assertIsNotNone(response.data)
 108         self.assertEqual(1, len(response.data))
 109         self.assertEquals(self.nsd_id, response.data[0]['nsdId'])
 110         self.assertEquals('ns', response.data[0]['nsInstanceName'])
 111         self.assertEquals('NOT_INSTANTIATED', response.data[0]['nsState'])
 112
 113     def test_query_one_ns(self):
 114         NSInstModel.objects.all().delete()
 115         id = self.test_create_ns()
 116         url = self.ns_instances_url + '/' + id
 117         response = self.apiClient.get(url)
 118         self.failUnlessEqual(status.HTTP_200_OK, response.status_code, response.data)
 119         self.assertIsNotNone(response.data)
 120         self.assertEquals(self.nsd_id, response.data['nsdId'])
 121         self.assertEquals('ns', response.data['nsInstanceName'])
 122         self.assertEquals('NOT_INSTANTIATED', response.data['nsState'])
 123
 124     @mock.patch.object(restcall, 'call_req')
 125     def test_delete_ns(self, mock_call_req):
 126         NSInstModel(id="1", nspackage_id="7", nsd_id="2").save()
 127         ns_info = {
 128             "service-instance-id": "service-instance-id-9b9348f2-f75d-4559-823d-db7ac138ed34",
 129             "service-instance-name": "service-instance-name-9b9348f2-f75d-4559-823d-db7ac138ed34",
 130             "service-type": "service-type-9b9348f2-f75d-4559-823d-db7ac138ed34",
 131             "service-role": "service-role-9b9348f2-f75d-4559-823d-db7ac138ed34",
 132             "resource-version": "1505350720009"
 133         }
 134         r1_query_ns_to_aai = [0, json.JSONEncoder().encode(ns_info), '200']
 135         r2_delete_ns_to_aai = [0, json.JSONEncoder().encode({}), '200']
 136         mock_call_req.side_effect = [r1_query_ns_to_aai, r2_delete_ns_to_aai]
 137         url = self.ns_instances_url + '/1'
 138         response = self.apiClient.delete(url)
 139         self.failUnlessEqual(status.HTTP_204_NO_CONTENT, response.status_code)