1 # Copyright (C) 2019 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.
20 from django.test import TestCase
21 from rest_framework import status
22 from rest_framework.test import APIClient
23 from requests.auth import HTTPBasicAuth
25 import catalog.pub.utils.timeutil
26 from catalog.packages import const
27 from catalog.packages.biz.notificationsutil import PkgNotifications
28 from catalog.packages.biz.vnf_pkg_subscription import QuerySubscription, TerminateSubscription
29 from catalog.pub.config import config as pub_config
30 from catalog.pub.config.config import CATALOG_ROOT_PATH, MSB_SERVICE_IP, MSB_SERVICE_PORT
31 from catalog.pub.database.models import VnfPkgSubscriptionModel, VnfPackageModel
32 from catalog.pub.exceptions import SubscriptionDoesNotExistsException
33 from .const import vnf_subscription_data
34 from catalog.pub.utils import toscaparser
35 from catalog.packages.tests.const import vnfd_data
38 class TestNfPackageSubscription(TestCase):
40 self.client = APIClient()
41 VnfPkgSubscriptionModel.objects.filter().delete()
42 self.vnf_subscription_data = vnf_subscription_data
47 @mock.patch("requests.get")
48 @mock.patch.object(uuid, 'uuid4')
49 def test_create_vnf_subscription(self, mock_uuid4, mock_requests):
50 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
51 mock_requests.return_value.status_code = 204
52 mock_requests.get.status_code = 204
53 mock_uuid4.return_value = temp_uuid
54 response = self.client.post(
55 "/api/vnfpkgm/v1/subscriptions",
56 data=self.vnf_subscription_data,
59 self.assertEqual(201, response.status_code)
61 self.vnf_subscription_data["callbackUri"],
62 response.data["callbackUri"]
64 self.assertEqual(temp_uuid, response.data["id"])
66 @mock.patch("requests.get")
67 @mock.patch.object(uuid, 'uuid4')
68 def test_duplicate_subscriptions(self, mock_uuid4, mock_requests):
69 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
70 temp1_uuid = "00342b18-a5c7-11e8-998c-bf1755941f12"
71 mock_requests.return_value.status_code = 204
72 mock_requests.get.status_code = 204
73 mock_uuid4.side_effect = [temp_uuid, temp1_uuid]
74 response = self.client.post(
75 "/api/vnfpkgm/v1/subscriptions",
76 data=self.vnf_subscription_data,
79 self.assertEqual(201, response.status_code)
81 self.vnf_subscription_data["callbackUri"],
82 response.data["callbackUri"]
84 self.assertEqual(temp_uuid, response.data["id"])
85 temp2_uuid = "00442b18-a5c7-11e8-998c-bf1755941f12"
86 mock_requests.return_value.status_code = 204
87 mock_requests.get.status_code = 204
88 mock_uuid4.return_value = temp2_uuid
89 response = self.client.post(
90 "/api/vnfpkgm/v1/subscriptions",
91 data=self.vnf_subscription_data,
94 self.assertEqual(303, response.status_code)
95 redirect_addr = "https://%s:%s/%s" % (MSB_SERVICE_IP, MSB_SERVICE_PORT,
96 os.path.join(const.VNFPKG_SUBSCRIPTION_ROOT_URI, temp_uuid))
97 self.assertEqual(redirect_addr, response["Location"])
99 @mock.patch("requests.get")
100 def test_callbackuri_same_subscriptions(self, mock_requests):
101 mock_requests.return_value.status_code = 204
102 mock_requests.get.status_code = 204
103 response = self.client.post(
104 "/api/vnfpkgm/v1/subscriptions",
105 data=self.vnf_subscription_data,
108 self.assertEqual(201, response.status_code)
110 self.vnf_subscription_data["callbackUri"],
111 response.data["callbackUri"]
113 newsubs = self.vnf_subscription_data
114 newsubs["filter"]["vnfdId"] = ["ssss"]
115 response = self.client.post(
116 "/api/vnfpkgm/v1/subscriptions",
117 data=self.vnf_subscription_data,
120 self.assertEqual(201, response.status_code)
122 newsubs["callbackUri"],
123 response.data["callbackUri"]
126 @mock.patch("requests.get")
127 @mock.patch.object(uuid, 'uuid4')
128 def test_get_subscriptions(self, mock_uuid4, mock_requests):
129 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
130 mock_requests.return_value.status_code = 204
131 mock_requests.get.status_code = 204
132 mock_uuid4.return_value = temp_uuid
134 "/api/vnfpkgm/v1/subscriptions",
135 data=self.vnf_subscription_data,
138 response = self.client.get(
139 "/api/vnfpkgm/v1/subscriptions?usageState=IN_USE",
142 self.assertEqual(200, response.status_code)
143 self.assertEqual(1, len(response.data))
145 @mock.patch("requests.get")
146 @mock.patch.object(uuid, 'uuid4')
147 def test_get_subscriptions_with_invalid_params(self, mock_uuid4, mock_requests):
148 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
149 mock_requests.return_value.status_code = 204
150 mock_requests.get.status_code = 204
151 mock_uuid4.return_value = temp_uuid
153 "/api/vnfpkgm/v1/subscriptions",
154 data=self.vnf_subscription_data,
157 response = self.client.get(
158 "/api/vnfpkgm/v1/subscriptions?dummy=dummy",
161 self.assertEqual(400, response.status_code)
163 @mock.patch("requests.get")
164 @mock.patch.object(uuid, 'uuid4')
165 def test_get_subscription_with_id(self, mock_uuid4, mock_requests):
166 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
167 mock_requests.return_value.status_code = 204
168 mock_requests.get.status_code = 204
169 mock_uuid4.return_value = temp_uuid
171 "/api/vnfpkgm/v1/subscriptions",
172 data=self.vnf_subscription_data,
175 response = self.client.get(
176 "/api/vnfpkgm/v1/subscriptions/%s" % temp_uuid,
179 self.assertEqual(200, response.status_code)
180 self.assertEqual(temp_uuid, response.data["id"])
182 @mock.patch("requests.get")
183 def test_get_subscription_with_id_not_exists(self, mock_requests):
184 mock_requests.return_value.status_code = 204
185 mock_requests.get.status_code = 204
187 "/api/vnfpkgm/v1/subscriptions",
188 data=self.vnf_subscription_data,
191 response = self.client.get(
192 "/api/vnfpkgm/v1/subscriptions/111",
195 self.assertEqual(404, response.status_code)
197 @mock.patch("requests.get")
198 @mock.patch.object(uuid, 'uuid4')
199 def test_delete_subscription_with_id(self, mock_uuid4, mock_requests):
200 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
201 dummy_uuid = str(uuid.uuid4())
202 mock_requests.return_value.status_code = 204
203 mock_requests.get.status_code = 204
204 mock_uuid4.return_value = temp_uuid
206 "/api/vnfpkgm/v1/subscriptions",
207 data=self.vnf_subscription_data,
211 "/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid,
214 response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % temp_uuid)
215 self.assertEqual(204, response.status_code)
217 def test_delete_subscription_with_id_not_exists(self):
218 dummy_uuid = str(uuid.uuid4())
219 response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid)
220 self.assertEqual(404, response.status_code)
222 @mock.patch("requests.get")
223 @mock.patch.object(toscaparser, 'parse_vnfd')
224 @mock.patch("requests.post")
225 @mock.patch("uuid.uuid4")
226 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
227 def test_vnfpkg_subscript_notify(self, mock_nowtime, mock_uuid, mock_requests_post, mock_parse_vnfd,
229 mock_nowtime.return_value = "2019-02-16 14:41:16"
230 uuid_subscriptid = "99442b18-a5c7-11e8-998c-bf1755941f13"
231 uuid_vnfPackageId = "3fa85f64-5717-4562-b3fc-2c963f66afa6"
232 uuid_vnfdid = "00342b18-a5c7-11e8-998c-bf1755941f12"
233 mock_uuid.side_effect = [uuid_subscriptid, "1111"]
234 mock_requests_get.return_value.status_code = 204
235 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
237 response = self.client.post(
238 "/api/vnfpkgm/v1/subscriptions",
239 data=vnf_subscription_data,
241 self.assertEqual(201, response.status_code)
243 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rt")}
244 VnfPackageModel.objects.create(
245 vnfPackageId=uuid_vnfPackageId,
246 onboardingState="CREATED"
249 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/%s/package_content" % uuid_vnfPackageId, data=data)
250 vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=uuid_vnfPackageId)
251 self.assertEqual(uuid_vnfdid, vnf_pkg[0].vnfdId)
252 self.assertEqual(const.PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
253 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
255 expect_notification = {
257 'notificationType': const.PKG_NOTIFICATION_TYPE.ONBOARDING,
258 'timeStamp': "2019-02-16 14:41:16",
259 'vnfPkgId': uuid_vnfPackageId,
260 'vnfdId': uuid_vnfdid,
263 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
264 pub_config.MSB_SERVICE_PORT,
265 const.PKG_URL_PREFIX,
269 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
270 pub_config.MSB_SERVICE_PORT,
271 const.VNFPKG_SUBSCRIPTION_ROOT_URI,
274 "subscriptionId": uuid_subscriptid
276 mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"], data=json.dumps(expect_notification),
277 headers={'Connection': 'close',
278 'content-type': 'application/json',
279 'accept': 'application/json'},
280 auth=HTTPBasicAuth("admin", "pwd1234"),
283 def test_service_query_single_subscription_not_found(self):
285 subscription_id = "test_not_found"
286 QuerySubscription().query_single_subscription(subscription_id)
287 except SubscriptionDoesNotExistsException as e:
288 self.assertEqual("Subscription with ID: %s does not exist" % subscription_id, e.args[0])
290 def test_service_delete_single_subscription_not_found(self):
292 subscription_id = "test_not_found"
293 TerminateSubscription().terminate(subscription_id)
294 except SubscriptionDoesNotExistsException as e:
295 self.assertEqual("Subscription with ID: %s does not exist" % subscription_id, e.args[0])
298 class NotificationTest(TestCase):
300 VnfPackageModel.objects.all().delete()
301 VnfPkgSubscriptionModel.objects.all().delete()
304 VnfPackageModel.objects.all().delete()
305 VnfPkgSubscriptionModel.objects.all().delete()
307 @mock.patch("requests.post")
308 @mock.patch("uuid.uuid4")
309 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
310 def test_vnfpkg_manual_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
311 VnfPackageModel(vnfPackageId="vnfpkgid1",
315 VnfPkgSubscriptionModel(subscription_id="1",
316 callback_uri="http://127.0.0.1/self",
317 notification_types=const.NOTIFICATION_TYPES,
319 vnf_pkg_id="vnfpkgid1"
321 mock_nowtime.return_value = "2019-12-16 14:41:16"
322 mock_uuid.return_value = "1111"
323 notify = PkgNotifications(const.PKG_NOTIFICATION_TYPE.CHANGE, "vnfpkgid1",
324 const.PKG_CHANGE_TYPE.OP_STATE_CHANGE, operational_state=None)
326 notify.send_notification()
327 expect_callbackuri = "http://127.0.0.1/self"
328 expect_notification = {
330 'notificationType': const.PKG_NOTIFICATION_TYPE.CHANGE,
331 'timeStamp': "2019-12-16 14:41:16",
332 'vnfPkgId': "vnfpkgid1",
336 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
337 pub_config.MSB_SERVICE_PORT,
338 const.PKG_URL_PREFIX,
342 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
343 pub_config.MSB_SERVICE_PORT,
344 const.VNFPKG_SUBSCRIPTION_ROOT_URI,
347 'changeType': const.PKG_CHANGE_TYPE.OP_STATE_CHANGE,
348 'operationalState': None,
349 "subscriptionId": "1"
351 mock_requests_post.assert_called_with(expect_callbackuri, data=json.dumps(expect_notification),
352 headers={'Connection': 'close',
353 'content-type': 'application/json',
354 'accept': 'application/json'},