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"
70 SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
71 auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
72 operation_types="['INSTANTIATE']",
73 operation_states="['STARTING']",
74 links=json.dumps(links),
75 vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
76 response = self.client.get("/api/vnflcm/v1/subscriptions", format='json')
77 self.assertEqual(response.status_code, status.HTTP_200_OK)
78 self.assertEqual([self.test_single_subscription], response.data)
80 def test_get_subscriptions_with_vnf_instance_id(self):
81 vnf_instance_filter = {
83 "vnfInstanceIds": [self.vnf_instance_id],
84 "vnfInstanceNames": [],
85 "vnfProductsFromProviders": {
86 "vnfProvider": "vendor"
91 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
94 SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
95 auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
96 operation_types="['INSTANTIATE']",
97 operation_states="['STARTING']",
98 links=json.dumps(links),
99 vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
100 dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
101 dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
102 vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
103 SubscriptionModel(subscription_id=dummy_subscription_id, callback_uri="http://aurl.com",
104 auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
105 operation_types="['INSTANTIATE']",
106 operation_states="['STARTING']",
107 links=json.dumps(links),
108 vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
110 response = self.client.get("/api/vnflcm/v1/subscriptions?vnfInstanceId=" + dummy_vnf_id, format='json')
111 expected_response = self.test_single_subscription.copy()
112 expected_response["id"] = dummy_subscription_id
113 expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
114 vnf_instance_filter["vnfInstanceIds"]
115 self.assertEqual(response.status_code, status.HTTP_200_OK)
116 self.assertEqual([expected_response], response.data)
118 def test_get_subscriptions_with_unknown_vnf_instance_id(self):
119 vnf_instance_filter = {
121 "vnfInstanceIds": [self.vnf_instance_id],
122 "vnfInstanceNames": [],
123 "vnfProductsFromProviders": {
124 "vnfProvider": "vendor"
129 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
132 SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
133 auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
134 operation_types="['INSTANTIATE']",
135 operation_states="['STARTING']",
136 links=json.dumps(links),
137 vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
138 response = self.client.get("/api/vnflcm/v1/subscriptions?vnfInstanceId=dummy", format='json')
139 self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
141 def test_get_subscriptions_with_invalid_filter(self):
142 vnf_instance_filter = {
144 "vnfInstanceIds": [self.vnf_instance_id],
145 "vnfInstanceNames": [],
146 "vnfProductsFromProviders": {
147 "vnfProvider": "vendor"
152 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
155 SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
156 auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
157 operation_types="['INSTANTIATE']",
158 operation_states="['STARTING']",
159 links=json.dumps(links),
160 vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
161 response = self.client.get("/api/vnflcm/v1/subscriptions?dummy=dummy", format='json')
162 self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
164 def test_get_subscriptions_with_operation_type_filter(self):
165 vnf_instance_filter = {
167 "vnfInstanceIds": [self.vnf_instance_id],
168 "vnfInstanceNames": [],
169 "vnfProductsFromProviders": {
170 "vnfProvider": "vendor"
175 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
178 SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
179 auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
180 operation_types="['INSTANTIATE']",
181 operation_states="['STARTING']",
182 links=json.dumps(links),
183 vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
184 dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
185 dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
186 vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
187 SubscriptionModel(subscription_id=dummy_subscription_id, callback_uri="http://aurl.com",
188 auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
189 operation_types="['SCALE']",
190 operation_states="['STARTING']",
191 links=json.dumps(links),
192 vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
194 response = self.client.get("/api/vnflcm/v1/subscriptions?operationTypes=SCALE", format='json')
195 expected_response = self.test_single_subscription.copy()
196 expected_response["id"] = dummy_subscription_id
197 expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
198 vnf_instance_filter["vnfInstanceIds"]
199 expected_response["filter"]["operationTypes"] = ["SCALE"]
200 self.assertEqual(response.status_code, status.HTTP_200_OK)
201 self.assertEqual([expected_response], response.data)