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 import enumutil
  18 import fileutil
  19 import json
  20 import urllib2
  21 import syscomm
  22 import timeutil
  23 import values
  24 import platform
  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 "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(urllib2, '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': u'string',
 251                 'vnfProducts': {
 252                     'vnfProductName': 'string',
 253                     'versions': {
 254                         'vnfSoftwareVersion': u'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('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",
 280             "_links": {}
 281         }
 282         mock_post.return_value.status_code = 204
 283         NotificationsUtil().send_notification(dummy_notification)
 284         mock_post.assert_called_once()
 285
 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": "",
 291             "operation": "",
 292             "_links": {}
 293         }
 294         mock_post.return_value.status_code = 204
 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()