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