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 rest_framework.test import APIClient
21 from django.test import TestCase
23 from catalog.pub.database.models import VnfPkgSubscriptionModel, VnfPackageModel
24 from .const import vnf_subscription_data, vnfd_data
25 from catalog.packages.biz.notificationsutil import NotificationsUtil, prepare_vnfpkg_notification
26 from catalog.packages import const
27 from catalog.pub.config import config as pub_config
28 import catalog.pub.utils.timeutil
29 from catalog.pub.utils import toscaparser
30 from catalog.pub.config.config import CATALOG_ROOT_PATH
31 from rest_framework import status
34 class TestNfPackageSubscription(TestCase):
36 self.client = APIClient()
37 VnfPkgSubscriptionModel.objects.filter().delete()
38 self.vnf_subscription_data = vnf_subscription_data
43 @mock.patch("requests.get")
44 @mock.patch.object(uuid, 'uuid4')
45 def test_create_vnf_subscription(self, mock_uuid4, mock_requests):
46 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
47 mock_requests.return_value.status_code = 204
48 mock_requests.get.status_code = 204
49 mock_uuid4.return_value = temp_uuid
50 response = self.client.post(
51 "/api/vnfpkgm/v1/subscriptions",
52 data=self.vnf_subscription_data,
55 self.assertEqual(201, response.status_code)
57 self.vnf_subscription_data["callbackUri"],
58 response.data["callbackUri"]
60 self.assertEqual(temp_uuid, response.data["id"])
62 @mock.patch("requests.get")
63 @mock.patch.object(uuid, 'uuid4')
64 def test_duplicate_subscriptions(self, mock_uuid4, mock_requests):
65 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
66 temp1_uuid = "00342b18-a5c7-11e8-998c-bf1755941f12"
67 mock_requests.return_value.status_code = 204
68 mock_requests.get.status_code = 204
69 mock_uuid4.side_effect = [temp_uuid, temp1_uuid]
70 response = self.client.post(
71 "/api/vnfpkgm/v1/subscriptions",
72 data=self.vnf_subscription_data,
75 self.assertEqual(201, response.status_code)
77 self.vnf_subscription_data["callbackUri"],
78 response.data["callbackUri"]
80 self.assertEqual(temp_uuid, response.data["id"])
81 temp_uuid = "00442b18-a5c7-11e8-998c-bf1755941f12"
82 mock_requests.return_value.status_code = 204
83 mock_requests.get.status_code = 204
84 mock_uuid4.return_value = temp_uuid
85 response = self.client.post(
86 "/api/vnfpkgm/v1/subscriptions",
87 data=self.vnf_subscription_data,
90 self.assertEqual(303, response.status_code)
92 @mock.patch("requests.get")
93 @mock.patch.object(uuid, 'uuid4')
94 def test_get_subscriptions(self, mock_uuid4, mock_requests):
95 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
96 mock_requests.return_value.status_code = 204
97 mock_requests.get.status_code = 204
98 mock_uuid4.return_value = temp_uuid
100 "/api/vnfpkgm/v1/subscriptions",
101 data=self.vnf_subscription_data,
104 response = self.client.get(
105 "/api/vnfpkgm/v1/subscriptions?usageState=IN_USE",
108 self.assertEqual(200, response.status_code)
109 self.assertEqual(1, len(response.data))
111 @mock.patch("requests.get")
112 @mock.patch.object(uuid, 'uuid4')
113 def test_get_subscriptions_with_invalid_params(self, mock_uuid4, mock_requests):
114 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
115 mock_requests.return_value.status_code = 204
116 mock_requests.get.status_code = 204
117 mock_uuid4.return_value = temp_uuid
119 "/api/vnfpkgm/v1/subscriptions",
120 data=self.vnf_subscription_data,
123 response = self.client.get(
124 "/api/vnfpkgm/v1/subscriptions?dummy=dummy",
127 self.assertEqual(400, response.status_code)
129 @mock.patch("requests.get")
130 @mock.patch.object(uuid, 'uuid4')
131 def test_get_subscription_with_id(self, mock_uuid4, mock_requests):
132 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
133 mock_requests.return_value.status_code = 204
134 mock_requests.get.status_code = 204
135 mock_uuid4.return_value = temp_uuid
137 "/api/vnfpkgm/v1/subscriptions",
138 data=self.vnf_subscription_data,
141 response = self.client.get(
142 "/api/vnfpkgm/v1/subscriptions/%s" % temp_uuid,
145 self.assertEqual(200, response.status_code)
146 self.assertEqual(temp_uuid, response.data["id"])
148 @mock.patch("requests.get")
149 @mock.patch.object(uuid, 'uuid4')
150 def test_get_subscription_with_id_not_exists(self, mock_uuid4, mock_requests):
151 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
152 dummy_uuid = str(uuid.uuid4())
153 mock_requests.return_value.status_code = 204
154 mock_requests.get.status_code = 204
155 mock_uuid4.return_value = temp_uuid
157 "/api/vnfpkgm/v1/subscriptions",
158 data=self.vnf_subscription_data,
161 response = self.client.get(
162 "/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid,
165 self.assertEqual(404, response.status_code)
167 @mock.patch("requests.get")
168 @mock.patch.object(uuid, 'uuid4')
169 def test_delete_subscription_with_id(self, mock_uuid4, mock_requests):
170 temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
171 dummy_uuid = str(uuid.uuid4())
172 mock_requests.return_value.status_code = 204
173 mock_requests.get.status_code = 204
174 mock_uuid4.return_value = temp_uuid
176 "/api/vnfpkgm/v1/subscriptions",
177 data=self.vnf_subscription_data,
181 "/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid,
184 response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % temp_uuid)
185 self.assertEqual(204, response.status_code)
187 @mock.patch("requests.get")
188 @mock.patch.object(uuid, 'uuid4')
189 def test_delete_subscription_with_id_not_exists(self, mock_uuid4, mock_requests):
190 dummy_uuid = str(uuid.uuid4())
191 response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid)
192 self.assertEqual(404, response.status_code)
194 @mock.patch("requests.get")
195 @mock.patch.object(toscaparser, 'parse_vnfd')
196 @mock.patch("requests.post")
197 @mock.patch("uuid.uuid4")
198 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
199 def test_vnfpkg_subscript_notify(self, mock_nowtime, mock_uuid, mock_requests_post, mock_parse_vnfd, mock_requests_get):
200 mock_nowtime.return_value = "nowtime()"
201 uuid_subscriptid = "99442b18-a5c7-11e8-998c-bf1755941f13"
202 uuid_vnfPackageId = "3fa85f64-5717-4562-b3fc-2c963f66afa6"
203 uuid_vnfdid = "00342b18-a5c7-11e8-998c-bf1755941f12"
204 mock_uuid.side_effect = [uuid_subscriptid, "1111"]
205 mock_requests_get.return_value.status_code = 204
206 mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
208 response = self.client.post(
209 "/api/vnfpkgm/v1/subscriptions",
210 data=vnf_subscription_data,
212 self.assertEqual(201, response.status_code)
214 data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rt")}
215 VnfPackageModel.objects.create(
216 vnfPackageId=uuid_vnfPackageId,
217 onboardingState="CREATED"
220 response = self.client.put("/api/vnfpkgm/v1/vnf_packages/%s/package_content" % uuid_vnfPackageId, data=data)
221 vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=uuid_vnfPackageId)
222 self.assertEqual(uuid_vnfdid, vnf_pkg[0].vnfdId)
223 self.assertEqual(const.PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
224 self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
226 expect_notification = {
228 'notificationType': const.PKG_NOTIFICATION_TYPE.ONBOARDING,
229 'timeStamp': "nowtime()",
230 'vnfPkgId': uuid_vnfPackageId,
231 'vnfdId': uuid_vnfdid,
232 'changeType': const.PKG_CHANGE_TYPE.OP_STATE_CHANGE,
233 'operationalState': None,
234 "subscriptionId": uuid_subscriptid,
237 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
238 pub_config.MSB_SERVICE_PORT,
239 const.VNFPKG_SUBSCRIPTION_ROOT_URI,
242 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
243 pub_config.MSB_SERVICE_PORT,
244 const.PKG_URL_PREFIX,
249 mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"], data=expect_notification,
250 headers={'Connection': 'close'})
253 class NotificationTest(TestCase):
255 VnfPackageModel.objects.all().delete()
256 VnfPkgSubscriptionModel.objects.all().delete()
259 VnfPackageModel.objects.all().delete()
260 VnfPkgSubscriptionModel.objects.all().delete()
262 @mock.patch("requests.post")
263 @mock.patch("uuid.uuid4")
264 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
265 def test_vnfpkg_manual_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
266 VnfPackageModel(vnfPackageId="vnfpkgid1",
270 VnfPkgSubscriptionModel(subscription_id="1",
271 callback_uri="http://127.0.0.1/self",
272 notification_types=const.NOTIFICATION_TYPES,
274 vnf_pkg_id="vnfpkgid1"
276 mock_nowtime.return_value = "nowtime()"
277 mock_uuid.return_value = "1111"
278 notification_content = prepare_vnfpkg_notification("vnfpkgid1", const.PKG_NOTIFICATION_TYPE.CHANGE,
279 const.PKG_CHANGE_TYPE.OP_STATE_CHANGE, operational_state=None)
282 'vnfPkgId': 'vnf_pkg_id'
284 NotificationsUtil().send_notification(notification_content, filters, True)
285 expect_callbackuri = "http://127.0.0.1/self"
286 expect_notification = {
288 'notificationType': const.PKG_NOTIFICATION_TYPE.CHANGE,
289 'timeStamp': "nowtime()",
290 'vnfPkgId': "vnfpkgid1",
292 'changeType': const.PKG_CHANGE_TYPE.OP_STATE_CHANGE,
293 'operationalState': None,
294 "subscriptionId": "1",
297 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
298 pub_config.MSB_SERVICE_PORT,
299 const.VNFPKG_SUBSCRIPTION_ROOT_URI,
302 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
303 pub_config.MSB_SERVICE_PORT,
304 const.PKG_URL_PREFIX,
309 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification, headers={'Connection': 'close'})