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