1 # Copyright 2018 ZTE Corporation.
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 . import enumutil
18 # from . import fileutil
20 # import urllib.request
22 from . import timeutil
26 from lcm.pub.database.models import JobStatusModel, JobModel, SubscriptionModel
27 from lcm.pub.utils.jobutil import JobUtil
28 from lcm.pub.utils.notificationsutil import NotificationsUtil
39 class UtilsTest(unittest.TestCase):
47 MY_TYPE = enumutil.enum(SAMLL=0, LARGE=1)
48 self.assertEqual(0, MY_TYPE.SAMLL)
49 self.assertEqual(1, MY_TYPE.LARGE)
51 # def test_create_and_delete_dir(self):
52 # dirs = "/abc/def/hij"
53 # fileutil.make_dirs(dirs)
54 # fileutil.make_dirs(dirs)
55 # fileutil.delete_dirs(dirs)
57 # @mock.patch.object(urllib.request, 'urlopen')
58 # def test_download_file_from_http(self, mock_urlopen):
59 # mock_urlopen.return_value = MockReq()
60 # fileutil.delete_dirs("abc")
61 # is_ok, f_name = fileutil.download_file_from_http("1", "abc", "1.txt")
62 # self.assertTrue(is_ok)
63 # if platform.system() == 'Windows':
64 # self.assertTrue(f_name.endswith("abc\\1.txt"))
66 # self.assertTrue(f_name.endswith("abc/1.txt"))
67 # fileutil.delete_dirs("abc")
69 def test_query_job_status(self):
71 JobStatusModel.objects.filter().delete()
90 jobs = JobUtil.query_job_status(job_id)
91 self.assertEqual(1, len(jobs))
92 self.assertEqual(3, jobs[0].indexid)
93 jobs = JobUtil.query_job_status(job_id, 1)
94 self.assertEqual(2, len(jobs))
95 self.assertEqual(3, jobs[0].indexid)
96 self.assertEqual(2, jobs[1].indexid)
97 JobStatusModel.objects.filter().delete()
99 def test_is_job_exists(self):
101 JobModel.objects.filter().delete()
109 self.assertTrue(JobUtil.is_job_exists(job_id))
110 JobModel.objects.filter().delete()
112 def test_create_job(self):
114 JobModel.objects.filter().delete()
122 self.assertEqual(1, len(JobModel.objects.filter(jobid=job_id)))
123 JobModel.objects.filter().delete()
125 def test_clear_job(self):
127 JobModel.objects.filter().delete()
135 JobUtil.clear_job(job_id)
136 self.assertEqual(0, len(JobModel.objects.filter(jobid=job_id)))
138 def test_add_job_status_when_job_is_not_created(self):
139 JobModel.objects.filter().delete()
142 JobUtil.add_job_status,
149 def test_add_job_status_normal(self):
151 JobModel.objects.filter().delete()
152 JobStatusModel.objects.filter().delete()
160 JobUtil.add_job_status(
166 self.assertEqual(1, len(JobStatusModel.objects.filter(jobid=job_id)))
167 JobStatusModel.objects.filter().delete()
168 JobModel.objects.filter().delete()
170 def test_clear_job_status(self):
172 JobStatusModel.objects.filter().delete()
179 JobUtil.clear_job_status(job_id)
180 self.assertEqual(0, len(JobStatusModel.objects.filter(jobid=job_id)))
182 def test_get_unfinished_jobs(self):
183 JobModel.objects.filter().delete()
205 progresses = JobUtil.get_unfinished_jobs(
206 url_prefix="/vnfinst",
210 expect_progresses = ['/vnfinst/11', '/vnfinst/22', '/vnfinst/33']
211 self.assertEqual(expect_progresses, progresses)
212 JobModel.objects.filter().delete()
214 def test_fun_name(self):
215 self.assertEqual("test_fun_name", syscomm.fun_name())
217 def test_now_time(self):
218 self.assertIn(":", timeutil.now_time())
219 self.assertIn("-", timeutil.now_time())
221 def test_ignore_case_get(self):
226 self.assertEqual("def", values.ignore_case_get(data, 'ABC'))
227 self.assertEqual("def", values.ignore_case_get(data, 'abc'))
228 self.assertEqual("klm", values.ignore_case_get(data, 'hig'))
229 self.assertEqual("bbb", values.ignore_case_get(data, 'aaa', 'bbb'))
232 class TestNotificationUtils(unittest.TestCase):
236 "authType": ["BASIC"],
238 "username": "username",
239 "password": "password"
242 notification_types = ["VnfLcmOperationOccurrenceNotification"]
243 operation_types = ["INSTANTIATE"]
244 operation_states = ["STARTING"]
245 vnf_instance_filter = {
246 'vnfdIds': ['99442b18-a5c7-11e8-998c-bf1755941f13', '9fe4080c-b1a3-11e8-bb96-645106374fd3'],
247 'vnfInstanceIds': ['99442b18-a5c7-11e8-998c-bf1755941f12'],
248 'vnfInstanceNames': ['demo'],
249 'vnfProductsFromProviders': {
250 'vnfProvider': 'string',
252 'vnfProductName': 'string',
254 'vnfSoftwareVersion': 'string',
255 'vnfdVersions': 'string'
263 SubscriptionModel(subscription_id=subscription_id, callback_uri="http://demo",
264 auth_info=json.dumps(auth_params),
265 notification_types=json.dumps(notification_types),
266 operation_types=json.dumps(operation_types),
267 operation_states=json.dumps(operation_states),
268 vnf_instance_filter=json.dumps(vnf_instance_filter),
269 links=json.dumps(links)).save()
272 SubscriptionModel.objects.all().delete()
274 @mock.patch('requests.post')
275 def test_send_notification(self, mock_post):
276 dummy_notification = {
277 "vnfInstanceId": "99442b18-a5c7-11e8-998c-bf1755941f13",
278 "operationState": "STARTING",
279 "operation": "INSTANTIATE",
282 mock_post.return_value.status_code = 204
283 NotificationsUtil().send_notification(dummy_notification)
284 mock_post.assert_called_once()
286 @mock.patch('requests.post')
287 def test_send_notification_with_empty_filters(self, mock_post):
288 dummy_notification = {
289 "vnfInstanceId": "9fe4080c-b1a3-11e8-bb96-645106374fd3",
290 "operationState": "",
294 mock_post.return_value.status_code = 204
295 NotificationsUtil().send_notification(dummy_notification)
296 mock_post.assert_called_once()
298 @mock.patch('requests.post')
299 def test_send_notification_unmatched_filters(self, mock_post):
300 dummy_notification = {
301 "vnfInstanceId": "9fe4080c-b1a3-11e8-bb96-xxxxx",
302 "operationState": "DUMMY",
303 "operation": "DUMMY",
305 NotificationsUtil().send_notification(dummy_notification)
306 mock_post.assert_not_called()