1 # Copyright (C) 2018 Verizon. All Rights Reserved
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
17 from django.test import TestCase, Client
18 from rest_framework import status
20 from lcm.pub.database.models import SubscriptionModel
23 class TestQuerySubscriptions(TestCase):
25 self.client = Client()
26 self.subscription_id = "99442b18-a5c7-11e8-998c-bf1755941f16"
27 self.vnf_instance_id = "cd552c9c-ab6f-11e8-b354-236c32aa91a1"
28 SubscriptionModel.objects.all().delete()
29 self.test_single_subscription = {
30 "id": self.subscription_id,
31 "callbackUri": "http://aurl.com",
34 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
38 "notificationTypes": ["VnfLcmOperationOccurrenceNotification"],
39 "operationTypes": ["INSTANTIATE"],
40 "operationStates": ["STARTING"],
41 "vnfInstanceSubscriptionFilter": {
43 "vnfInstanceIds": [self.vnf_instance_id],
44 "vnfInstanceNames": [],
45 "vnfProductsFromProviders": {
46 "vnfProvider": "vendor"
56 def test_get_subscriptions(self):
57 vnf_instance_filter = {
59 "vnfInstanceIds": [self.vnf_instance_id],
60 "vnfInstanceNames": [],
61 "vnfProductsFromProviders": {
62 "vnfProvider": "vendor"
67 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
71 subscription_id=self.subscription_id,
72 callback_uri="http://aurl.com",
74 notification_types="['VnfLcmOperationOccurrenceNotification']",
75 operation_types="['INSTANTIATE']",
76 operation_states="['STARTING']",
77 links=json.dumps(links),
78 vnf_instance_filter=json.dumps(vnf_instance_filter)
80 response = self.client.get(
81 "/api/vnflcm/v1/subscriptions",
84 self.assertEqual(response.status_code, status.HTTP_200_OK)
85 self.assertEqual([self.test_single_subscription], response.data)
87 def test_get_subscriptions_with_vnf_instance_id(self):
88 vnf_instance_filter = {
90 "vnfInstanceIds": [self.vnf_instance_id],
91 "vnfInstanceNames": [],
92 "vnfProductsFromProviders": {
93 "vnfProvider": "vendor"
98 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
102 subscription_id=self.subscription_id,
103 callback_uri="http://aurl.com",
105 notification_types="['VnfLcmOperationOccurrenceNotification']",
106 operation_types="['INSTANTIATE']",
107 operation_states="['STARTING']",
108 links=json.dumps(links),
109 vnf_instance_filter=json.dumps(vnf_instance_filter)
111 dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
112 dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
113 vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
115 subscription_id=dummy_subscription_id,
116 callback_uri="http://aurl.com",
118 notification_types="['VnfLcmOperationOccurrenceNotification']",
119 operation_types="['INSTANTIATE']",
120 operation_states="['STARTING']",
121 links=json.dumps(links),
122 vnf_instance_filter=json.dumps(vnf_instance_filter)
125 response = self.client.get(
126 "/api/vnflcm/v1/subscriptions?vnfInstanceId=" + dummy_vnf_id,
129 expected_response = self.test_single_subscription.copy()
130 expected_response["id"] = dummy_subscription_id
131 expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
132 vnf_instance_filter["vnfInstanceIds"]
133 self.assertEqual(response.status_code, status.HTTP_200_OK)
134 self.assertEqual([expected_response], response.data)
136 def test_get_subscriptions_with_unknown_vnf_instance_id(self):
137 vnf_instance_filter = {
139 "vnfInstanceIds": [self.vnf_instance_id],
140 "vnfInstanceNames": [],
141 "vnfProductsFromProviders": {
142 "vnfProvider": "vendor"
147 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
151 subscription_id=self.subscription_id,
152 callback_uri="http://aurl.com",
154 notification_types="['VnfLcmOperationOccurrenceNotification']",
155 operation_types="['INSTANTIATE']",
156 operation_states="['STARTING']",
157 links=json.dumps(links),
158 vnf_instance_filter=json.dumps(vnf_instance_filter)
160 response = self.client.get(
161 "/api/vnflcm/v1/subscriptions?vnfInstanceId=dummy",
165 response.status_code,
166 status.HTTP_500_INTERNAL_SERVER_ERROR
169 def test_get_subscriptions_with_invalid_filter(self):
170 vnf_instance_filter = {
172 "vnfInstanceIds": [self.vnf_instance_id],
173 "vnfInstanceNames": [],
174 "vnfProductsFromProviders": {
175 "vnfProvider": "vendor"
180 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
184 subscription_id=self.subscription_id,
185 callback_uri="http://aurl.com",
187 notification_types="['VnfLcmOperationOccurrenceNotification']",
188 operation_types="['INSTANTIATE']",
189 operation_states="['STARTING']",
190 links=json.dumps(links),
191 vnf_instance_filter=json.dumps(vnf_instance_filter)
193 response = self.client.get(
194 "/api/vnflcm/v1/subscriptions?dummy=dummy",
197 self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
199 def test_get_subscriptions_with_operation_type_filter(self):
200 vnf_instance_filter = {
202 "vnfInstanceIds": [self.vnf_instance_id],
203 "vnfInstanceNames": [],
204 "vnfProductsFromProviders": {
205 "vnfProvider": "vendor"
210 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
214 subscription_id=self.subscription_id,
215 callback_uri="http://aurl.com",
217 notification_types="['VnfLcmOperationOccurrenceNotification']",
218 operation_types="['INSTANTIATE']",
219 operation_states="['STARTING']",
220 links=json.dumps(links),
221 vnf_instance_filter=json.dumps(vnf_instance_filter)
223 dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
224 dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
225 vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
227 subscription_id=dummy_subscription_id,
228 callback_uri="http://aurl.com",
230 notification_types="['VnfLcmOperationOccurrenceNotification']",
231 operation_types="['SCALE']",
232 operation_states="['STARTING']",
233 links=json.dumps(links),
234 vnf_instance_filter=json.dumps(vnf_instance_filter)
237 response = self.client.get(
238 "/api/vnflcm/v1/subscriptions?operationTypes=SCALE",
241 expected_response = self.test_single_subscription.copy()
242 expected_response["id"] = dummy_subscription_id
243 expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
244 vnf_instance_filter["vnfInstanceIds"]
245 expected_response["filter"]["operationTypes"] = ["SCALE"]
246 self.assertEqual(response.status_code, status.HTTP_200_OK)
247 self.assertEqual([expected_response], response.data)