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 from urllib import request
  20 from . import syscomm
  21 from . import timeutil
  22 from . 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(request, '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("https://git.onap.org/vfc/nfvo/lcm/tree/version.properties", "abc", "1.txt")
  59         self.assertTrue(is_ok)
  60         self.assertTrue(f_name.endswith("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'))