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