lcm/lcm/pub/utils/tests.py

   1 # Copyright 2018 ZTE Corporation.
   2 #
   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
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import unittest
  16 import mock
  17 from . import enumutil
  18 # from . import fileutil
  19 import json
  20 # import urllib.request
  21 from . import syscomm
  22 from . import timeutil
  23 from . import values
  24 import codecs
  25
  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
  29
  30
  31 class MockReq():
  32     def read(self):
  33         return b"1"
  34
  35     def close(self):
  36         pass
  37
  38
  39 class UtilsTest(unittest.TestCase):
  40     def setUp(self):
  41         pass
  42
  43     def tearDown(self):
  44         pass
  45
  46     def test_enum(self):
  47         MY_TYPE = enumutil.enum(SAMLL=0, LARGE=1)
  48         self.assertEqual(0, MY_TYPE.SAMLL)
  49         self.assertEqual(1, MY_TYPE.LARGE)
  50
  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)
  56     #
  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"))
  65     #     else:
  66     #         self.assertTrue(f_name.endswith("abc/1.txt"))
  67     #     fileutil.delete_dirs("abc")
  68
  69     def test_query_job_status(self):
  70         job_id = "1"
  71         JobStatusModel.objects.filter().delete()
  72         JobStatusModel(
  73             indexid=1,
  74             jobid=job_id,
  75             status="success",
  76             progress=10
  77         ).save()
  78         JobStatusModel(
  79             indexid=2,
  80             jobid=job_id,
  81             status="success",
  82             progress=50
  83         ).save()
  84         JobStatusModel(
  85             indexid=3,
  86             jobid=job_id,
  87             status="success",
  88             progress=100
  89         ).save()
  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()
  98
  99     def test_is_job_exists(self):
 100         job_id = "1"
 101         JobModel.objects.filter().delete()
 102         JobModel(
 103             jobid=job_id,
 104             jobtype="1",
 105             jobaction="2",
 106             resid="3",
 107             status=0
 108         ).save()
 109         self.assertTrue(JobUtil.is_job_exists(job_id))
 110         JobModel.objects.filter().delete()
 111
 112     def test_create_job(self):
 113         job_id = "5"
 114         JobModel.objects.filter().delete()
 115         JobUtil.create_job(
 116             inst_type="1",
 117             jobaction="2",
 118             inst_id="3",
 119             user="4",
 120             job_id=5,
 121             res_name="6")
 122         self.assertEqual(1, len(JobModel.objects.filter(jobid=job_id)))
 123         JobModel.objects.filter().delete()
 124
 125     def test_clear_job(self):
 126         job_id = "1"
 127         JobModel.objects.filter().delete()
 128         JobModel(
 129             jobid=job_id,
 130             jobtype="1",
 131             jobaction="2",
 132             resid="3",
 133             status=0
 134         ).save()
 135         JobUtil.clear_job(job_id)
 136         self.assertEqual(0, len(JobModel.objects.filter(jobid=job_id)))
 137
 138     def test_add_job_status_when_job_is_not_created(self):
 139         JobModel.objects.filter().delete()
 140         self.assertRaises(
 141             Exception,
 142             JobUtil.add_job_status,
 143             job_id="1",
 144             progress=1,
 145             status_decs="2",
 146             error_code="0"
 147         )
 148
 149     def test_add_job_status_normal(self):
 150         job_id = "1"
 151         JobModel.objects.filter().delete()
 152         JobStatusModel.objects.filter().delete()
 153         JobModel(
 154             jobid=job_id,
 155             jobtype="1",
 156             jobaction="2",
 157             resid="3",
 158             status=0
 159         ).save()
 160         JobUtil.add_job_status(
 161             job_id="1",
 162             progress=1,
 163             status_decs="2",
 164             error_code="0"
 165         )
 166         self.assertEqual(1, len(JobStatusModel.objects.filter(jobid=job_id)))
 167         JobStatusModel.objects.filter().delete()
 168         JobModel.objects.filter().delete()
 169
 170     def test_clear_job_status(self):
 171         job_id = "1"
 172         JobStatusModel.objects.filter().delete()
 173         JobStatusModel(
 174             indexid=1,
 175             jobid=job_id,
 176             status="success",
 177             progress=10
 178         ).save()
 179         JobUtil.clear_job_status(job_id)
 180         self.assertEqual(0, len(JobStatusModel.objects.filter(jobid=job_id)))
 181
 182     def test_get_unfinished_jobs(self):
 183         JobModel.objects.filter().delete()
 184         JobModel(
 185             jobid="11",
 186             jobtype="InstVnf",
 187             jobaction="2",
 188             resid="3",
 189             status=0
 190         ).save()
 191         JobModel(
 192             jobid="22",
 193             jobtype="InstVnf",
 194             jobaction="2",
 195             resid="3",
 196             status=0
 197         ).save()
 198         JobModel(
 199             jobid="33",
 200             jobtype="InstVnf",
 201             jobaction="2",
 202             resid="3",
 203             status=0
 204         ).save()
 205         progresses = JobUtil.get_unfinished_jobs(
 206             url_prefix="/vnfinst",
 207             inst_id="3",
 208             inst_type="InstVnf"
 209         )
 210         expect_progresses = ['/vnfinst/11', '/vnfinst/22', '/vnfinst/33']
 211         self.assertEqual(expect_progresses, progresses)
 212         JobModel.objects.filter().delete()
 213
 214     def test_fun_name(self):
 215         self.assertEqual("test_fun_name", syscomm.fun_name())
 216
 217     def test_now_time(self):
 218         self.assertIn(":", timeutil.now_time())
 219         self.assertIn("-", timeutil.now_time())
 220
 221     def test_ignore_case_get(self):
 222         data = {
 223             "Abc": "def",
 224             "HIG": "klm"
 225         }
 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'))
 230
 231
 232 class TestNotificationUtils(unittest.TestCase):
 233     def setUp(self):
 234         subscription_id = 1
 235         auth_params = {
 236             "authType": ["BASIC"],
 237             "paramsBasic": {
 238                 "username": "username",
 239                 "password": "password"
 240             }
 241         }
 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',
 251                 'vnfProducts': {
 252                     'vnfProductName': 'string',
 253                     'versions': {
 254                         'vnfSoftwareVersion': 'string',
 255                         'vnfdVersions': 'string'
 256                     }
 257                 }
 258             }
 259         }
 260         links = {
 261             "self": "demo"
 262         }
 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()
 270
 271     def tearDown(self):
 272         SubscriptionModel.objects.all().delete()
 273
 274     @mock.patch('httplib2.Http.request')
 275     def test_send_notification(self, mock_post):
 276         dummy_notification = {
 277             "vnfInstanceId": "9fe4080c-b1a3-11e8-bb96-645106374fd3",
 278             "operationState": "STARTING",
 279             "operation": "INSTANTIATE",
 280             "_links": {}
 281         }
 282         mock_post.return_value = ({"status": "204"}, codecs.encode("1234", encoding='utf-8'))
 283         NotificationsUtil().send_notification(dummy_notification)
 284         mock_post.assert_called_once()
 285
 286     @mock.patch('httplib2.Http.request')
 287     def test_send_notification_with_empty_filters(self, mock_post):
 288         dummy_notification = {
 289             "vnfInstanceId": "9fe4080c-b1a3-11e8-bb96-645106374fd3",
 290             "operationState": "",
 291             "operation": "",
 292             "_links": {}
 293         }
 294         mock_post.return_value = ({"status": "204"}, codecs.encode("1234", encoding='utf-8'))
 295         NotificationsUtil().send_notification(dummy_notification)
 296         mock_post.assert_called_once()
 297
 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",
 304         }
 305         NotificationsUtil().send_notification(dummy_notification)
 306         mock_post.assert_not_called()