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 urllib2
  20 import syscomm
  21 import timeutil
  22 import values
  23 import platform
  24
  25 from lcm.pub.database.models import JobStatusModel, JobModel
  26 from lcm.pub.utils.jobutil import JobUtil
  27
  28
  29 class MockReq():
  30     def read(self):
  31         return "1"
  32
  33     def close(self):
  34         pass
  35
  36
  37 class UtilsTest(unittest.TestCase):
  38     def setUp(self):
  39         pass
  40
  41     def tearDown(self):
  42         pass
  43
  44     def test_enum(self):
  45         MY_TYPE = enumutil.enum(SAMLL=0, LARGE=1)
  46         self.assertEqual(0, MY_TYPE.SAMLL)
  47         self.assertEqual(1, MY_TYPE.LARGE)
  48
  49     def test_create_and_delete_dir(self):
  50         dirs = "abc/def/hij"
  51         fileutil.make_dirs(dirs)
  52         fileutil.make_dirs(dirs)
  53         fileutil.delete_dirs(dirs)
  54
  55     @mock.patch.object(urllib2, 'urlopen')
  56     def test_download_file_from_http(self, mock_urlopen):
  57         mock_urlopen.return_value = MockReq()
  58         fileutil.delete_dirs("abc")
  59         is_ok, f_name = fileutil.download_file_from_http("1", "abc", "1.txt")
  60         self.assertTrue(is_ok)
  61         if platform.system() == 'Windows':
  62             self.assertTrue(f_name.endswith("abc\\1.txt"))
  63         else:
  64             self.assertTrue(f_name.endswith("abc/1.txt"))
  65         fileutil.delete_dirs("abc")
  66
  67     def test_query_job_status(self):
  68         job_id = "1"
  69         JobStatusModel.objects.filter().delete()
  70         JobStatusModel(
  71             indexid=1,
  72             jobid=job_id,
  73             status="success",
  74             progress=10
  75         ).save()
  76         JobStatusModel(
  77             indexid=2,
  78             jobid=job_id,
  79             status="success",
  80             progress=50
  81         ).save()
  82         JobStatusModel(
  83             indexid=3,
  84             jobid=job_id,
  85             status="success",
  86             progress=100
  87         ).save()
  88         jobs = JobUtil.query_job_status(job_id)
  89         self.assertEqual(1, len(jobs))
  90         self.assertEqual(3, jobs[0].indexid)
  91         jobs = JobUtil.query_job_status(job_id, 1)
  92         self.assertEqual(2, len(jobs))
  93         self.assertEqual(3, jobs[0].indexid)
  94         self.assertEqual(2, jobs[1].indexid)
  95         JobStatusModel.objects.filter().delete()
  96
  97     def test_is_job_exists(self):
  98         job_id = "1"
  99         JobModel.objects.filter().delete()
 100         JobModel(
 101             jobid=job_id,
 102             jobtype="1",
 103             jobaction="2",
 104             resid="3",
 105             status=0
 106         ).save()
 107         self.assertTrue(JobUtil.is_job_exists(job_id))
 108         JobModel.objects.filter().delete()
 109
 110     def test_create_job(self):
 111         job_id = "5"
 112         JobModel.objects.filter().delete()
 113         JobUtil.create_job(
 114             inst_type="1",
 115             jobaction="2",
 116             inst_id="3",
 117             user="4",
 118             job_id=5,
 119             res_name="6")
 120         self.assertEqual(1, len(JobModel.objects.filter(jobid=job_id)))
 121         JobModel.objects.filter().delete()
 122
 123     def test_clear_job(self):
 124         job_id = "1"
 125         JobModel.objects.filter().delete()
 126         JobModel(
 127             jobid=job_id,
 128             jobtype="1",
 129             jobaction="2",
 130             resid="3",
 131             status=0
 132         ).save()
 133         JobUtil.clear_job(job_id)
 134         self.assertEqual(0, len(JobModel.objects.filter(jobid=job_id)))
 135
 136     def test_add_job_status_when_job_is_not_created(self):
 137         JobModel.objects.filter().delete()
 138         self.assertRaises(
 139             Exception,
 140             JobUtil.add_job_status,
 141             job_id="1",
 142             progress=1,
 143             status_decs="2",
 144             error_code="0"
 145         )
 146
 147     def test_add_job_status_normal(self):
 148         job_id = "1"
 149         JobModel.objects.filter().delete()
 150         JobStatusModel.objects.filter().delete()
 151         JobModel(
 152             jobid=job_id,
 153             jobtype="1",
 154             jobaction="2",
 155             resid="3",
 156             status=0
 157         ).save()
 158         JobUtil.add_job_status(
 159             job_id="1",
 160             progress=1,
 161             status_decs="2",
 162             error_code="0"
 163         )
 164         self.assertEqual(1, len(JobStatusModel.objects.filter(jobid=job_id)))
 165         JobStatusModel.objects.filter().delete()
 166         JobModel.objects.filter().delete()
 167
 168     def test_clear_job_status(self):
 169         job_id = "1"
 170         JobStatusModel.objects.filter().delete()
 171         JobStatusModel(
 172             indexid=1,
 173             jobid=job_id,
 174             status="success",
 175             progress=10
 176         ).save()
 177         JobUtil.clear_job_status(job_id)
 178         self.assertEqual(0, len(JobStatusModel.objects.filter(jobid=job_id)))
 179
 180     def test_get_unfinished_jobs(self):
 181         JobModel.objects.filter().delete()
 182         JobModel(
 183             jobid="11",
 184             jobtype="InstVnf",
 185             jobaction="2",
 186             resid="3",
 187             status=0
 188         ).save()
 189         JobModel(
 190             jobid="22",
 191             jobtype="InstVnf",
 192             jobaction="2",
 193             resid="3",
 194             status=0
 195         ).save()
 196         JobModel(
 197             jobid="33",
 198             jobtype="InstVnf",
 199             jobaction="2",
 200             resid="3",
 201             status=0
 202         ).save()
 203         progresses = JobUtil.get_unfinished_jobs(
 204             url_prefix="/vnfinst",
 205             inst_id="3",
 206             inst_type="InstVnf"
 207         )
 208         expect_progresses = ['/vnfinst/11', '/vnfinst/22', '/vnfinst/33']
 209         self.assertEqual(expect_progresses, progresses)
 210         JobModel.objects.filter().delete()
 211
 212     def test_fun_name(self):
 213         self.assertEqual("test_fun_name", syscomm.fun_name())
 214
 215     def test_now_time(self):
 216         self.assertIn(":", timeutil.now_time())
 217         self.assertIn("-", timeutil.now_time())
 218
 219     def test_ignore_case_get(self):
 220         data = {
 221             "Abc": "def",
 222             "HIG": "klm"
 223         }
 224         self.assertEqual("def", values.ignore_case_get(data, 'ABC'))
 225         self.assertEqual("def", values.ignore_case_get(data, 'abc'))
 226         self.assertEqual("klm", values.ignore_case_get(data, 'hig'))
 227         self.assertEqual("bbb", values.ignore_case_get(data, 'aaa', 'bbb'))