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 import fileutil
  19 import urllib2
  20 import syscomm
  21 import timeutil
  22 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(urllib2, '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'))