# Copyright 2018 ZTE Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import unittest import mock from . import enumutil # from . import fileutil import json # import urllib.request from . import syscomm from . import timeutil from . import values import codecs from lcm.pub.database.models import JobStatusModel, JobModel, SubscriptionModel from lcm.pub.utils.jobutil import JobUtil from lcm.pub.utils.notificationsutil import NotificationsUtil class MockReq(): def read(self): return b"1" def close(self): pass class UtilsTest(unittest.TestCase): def setUp(self): pass def tearDown(self): pass def test_enum(self): MY_TYPE = enumutil.enum(SAMLL=0, LARGE=1) self.assertEqual(0, MY_TYPE.SAMLL) self.assertEqual(1, MY_TYPE.LARGE) # def test_create_and_delete_dir(self): # dirs = "/abc/def/hij" # fileutil.make_dirs(dirs) # fileutil.make_dirs(dirs) # fileutil.delete_dirs(dirs) # # @mock.patch.object(urllib.request, 'urlopen') # def test_download_file_from_http(self, mock_urlopen): # mock_urlopen.return_value = MockReq() # fileutil.delete_dirs("abc") # is_ok, f_name = fileutil.download_file_from_http("1", "abc", "1.txt") # self.assertTrue(is_ok) # if platform.system() == 'Windows': # self.assertTrue(f_name.endswith("abc\\1.txt")) # else: # self.assertTrue(f_name.endswith("abc/1.txt")) # fileutil.delete_dirs("abc") def test_query_job_status(self): job_id = "1" JobStatusModel.objects.filter().delete() JobStatusModel( indexid=1, jobid=job_id, status="success", progress=10 ).save() JobStatusModel( indexid=2, jobid=job_id, status="success", progress=50 ).save() JobStatusModel( indexid=3, jobid=job_id, status="success", progress=100 ).save() jobs = JobUtil.query_job_status(job_id) self.assertEqual(1, len(jobs)) self.assertEqual(3, jobs[0].indexid) jobs = JobUtil.query_job_status(job_id, 1) self.assertEqual(2, len(jobs)) self.assertEqual(3, jobs[0].indexid) self.assertEqual(2, jobs[1].indexid) JobStatusModel.objects.filter().delete() def test_is_job_exists(self): job_id = "1" JobModel.objects.filter().delete() JobModel( jobid=job_id, jobtype="1", jobaction="2", resid="3", status=0 ).save() self.assertTrue(JobUtil.is_job_exists(job_id)) JobModel.objects.filter().delete() def test_create_job(self): job_id = "5" JobModel.objects.filter().delete() JobUtil.create_job( inst_type="1", jobaction="2", inst_id="3", user="4", job_id=5, res_name="6") self.assertEqual(1, len(JobModel.objects.filter(jobid=job_id))) JobModel.objects.filter().delete() def test_clear_job(self): job_id = "1" JobModel.objects.filter().delete() JobModel( jobid=job_id, jobtype="1", jobaction="2", resid="3", status=0 ).save() JobUtil.clear_job(job_id) self.assertEqual(0, len(JobModel.objects.filter(jobid=job_id))) def test_add_job_status_when_job_is_not_created(self): JobModel.objects.filter().delete() self.assertRaises( Exception, JobUtil.add_job_status, job_id="1", progress=1, status_decs="2", error_code="0" ) def test_add_job_status_normal(self): job_id = "1" JobModel.objects.filter().delete() JobStatusModel.objects.filter().delete() JobModel( jobid=job_id, jobtype="1", jobaction="2", resid="3", status=0 ).save() JobUtil.add_job_status( job_id="1", progress=1, status_decs="2", error_code="0" ) self.assertEqual(1, len(JobStatusModel.objects.filter(jobid=job_id))) JobStatusModel.objects.filter().delete() JobModel.objects.filter().delete() def test_clear_job_status(self): job_id = "1" JobStatusModel.objects.filter().delete() JobStatusModel( indexid=1, jobid=job_id, status="success", progress=10 ).save() JobUtil.clear_job_status(job_id) self.assertEqual(0, len(JobStatusModel.objects.filter(jobid=job_id))) def test_get_unfinished_jobs(self): JobModel.objects.filter().delete() JobModel( jobid="11", jobtype="InstVnf", jobaction="2", resid="3", status=0 ).save() JobModel( jobid="22", jobtype="InstVnf", jobaction="2", resid="3", status=0 ).save() JobModel( jobid="33", jobtype="InstVnf", jobaction="2", resid="3", status=0 ).save() progresses = JobUtil.get_unfinished_jobs( url_prefix="/vnfinst", inst_id="3", inst_type="InstVnf" ) expect_progresses = ['/vnfinst/11', '/vnfinst/22', '/vnfinst/33'] self.assertEqual(expect_progresses, progresses) JobModel.objects.filter().delete() def test_fun_name(self): self.assertEqual("test_fun_name", syscomm.fun_name()) def test_now_time(self): self.assertIn(":", timeutil.now_time()) self.assertIn("-", timeutil.now_time()) def test_ignore_case_get(self): data = { "Abc": "def", "HIG": "klm" } self.assertEqual("def", values.ignore_case_get(data, 'ABC')) self.assertEqual("def", values.ignore_case_get(data, 'abc')) self.assertEqual("klm", values.ignore_case_get(data, 'hig')) self.assertEqual("bbb", values.ignore_case_get(data, 'aaa', 'bbb')) class TestNotificationUtils(unittest.TestCase): def setUp(self): subscription_id = 1 auth_params = { "authType": ["BASIC"], "paramsBasic": { "username": "username", "password": "password" } } notification_types = ["VnfLcmOperationOccurrenceNotification"] operation_types = ["INSTANTIATE"] operation_states = ["STARTING"] vnf_instance_filter = { 'vnfdIds': ['99442b18-a5c7-11e8-998c-bf1755941f13', '9fe4080c-b1a3-11e8-bb96-645106374fd3'], 'vnfInstanceIds': ['99442b18-a5c7-11e8-998c-bf1755941f12'], 'vnfInstanceNames': ['demo'], 'vnfProductsFromProviders': { 'vnfProvider': 'string', 'vnfProducts': { 'vnfProductName': 'string', 'versions': { 'vnfSoftwareVersion': 'string', 'vnfdVersions': 'string' } } } } links = { "self": "demo" } SubscriptionModel(subscription_id=subscription_id, callback_uri="http://demo", auth_info=json.dumps(auth_params), notification_types=json.dumps(notification_types), operation_types=json.dumps(operation_types), operation_states=json.dumps(operation_states), vnf_instance_filter=json.dumps(vnf_instance_filter), links=json.dumps(links)).save() def tearDown(self): SubscriptionModel.objects.all().delete() @mock.patch('httplib2.Http.request') def test_send_notification(self, mock_post): dummy_notification = { "vnfInstanceId": "9fe4080c-b1a3-11e8-bb96-645106374fd3", "operationState": "STARTING", "operation": "INSTANTIATE", "_links": {} } mock_post.return_value = ({"status": "204"}, codecs.encode("1234", encoding='utf-8')) NotificationsUtil().send_notification(dummy_notification) mock_post.assert_called_once() @mock.patch('httplib2.Http.request') def test_send_notification_with_empty_filters(self, mock_post): dummy_notification = { "vnfInstanceId": "9fe4080c-b1a3-11e8-bb96-645106374fd3", "operationState": "", "operation": "", "_links": {} } mock_post.return_value = ({"status": "204"}, codecs.encode("1234", encoding='utf-8')) NotificationsUtil().send_notification(dummy_notification) mock_post.assert_called_once() @mock.patch('requests.post') def test_send_notification_unmatched_filters(self, mock_post): dummy_notification = { "vnfInstanceId": "9fe4080c-b1a3-11e8-bb96-xxxxx", "operationState": "DUMMY", "operation": "DUMMY", } NotificationsUtil().send_notification(dummy_notification) mock_post.assert_not_called()