1 # Copyright 2018 ZTE Corporation.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
17 from . import enumutil
18 from . import fileutil
19 from urllib import request
21 from . import timeutil
24 from lcm.pub.database.models import JobStatusModel, JobModel
25 from lcm.pub.utils.jobutil import JobUtil
36 class UtilsTest(unittest.TestCase):
44 MY_TYPE = enumutil.enum(SAMLL=0, LARGE=1)
45 self.assertEqual(0, MY_TYPE.SAMLL)
46 self.assertEqual(1, MY_TYPE.LARGE)
48 def test_create_and_delete_dir(self):
50 fileutil.make_dirs(dirs)
51 fileutil.make_dirs(dirs)
52 fileutil.delete_dirs(dirs)
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")
63 def test_query_job_status(self):
65 JobStatusModel.objects.filter().delete()
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()
93 def test_is_job_exists(self):
95 JobModel.objects.filter().delete()
103 self.assertTrue(JobUtil.is_job_exists(job_id))
104 JobModel.objects.filter().delete()
106 def test_create_job(self):
108 JobModel.objects.filter().delete()
116 self.assertEqual(1, len(JobModel.objects.filter(jobid=job_id)))
117 JobModel.objects.filter().delete()
119 def test_clear_job(self):
121 JobModel.objects.filter().delete()
129 JobUtil.clear_job(job_id)
130 self.assertEqual(0, len(JobModel.objects.filter(jobid=job_id)))
132 def test_add_job_status_when_job_is_not_created(self):
133 JobModel.objects.filter().delete()
136 JobUtil.add_job_status,
143 def test_add_job_status_normal(self):
145 JobModel.objects.filter().delete()
146 JobStatusModel.objects.filter().delete()
154 JobUtil.add_job_status(
160 self.assertEqual(1, len(JobStatusModel.objects.filter(jobid=job_id)))
161 JobStatusModel.objects.filter().delete()
162 JobModel.objects.filter().delete()
164 def test_clear_job_status(self):
166 JobStatusModel.objects.filter().delete()
173 JobUtil.clear_job_status(job_id)
174 self.assertEqual(0, len(JobStatusModel.objects.filter(jobid=job_id)))
176 def test_get_unfinished_jobs(self):
177 JobModel.objects.filter().delete()
199 progresses = JobUtil.get_unfinished_jobs(
200 url_prefix="/vnfinst",
204 expect_progresses = ['/vnfinst/11', '/vnfinst/22', '/vnfinst/33']
205 self.assertEqual(expect_progresses, progresses)
206 JobModel.objects.filter().delete()
208 def test_fun_name(self):
209 self.assertEqual("test_fun_name", syscomm.fun_name())
211 def test_now_time(self):
212 self.assertIn(":", timeutil.now_time())
213 self.assertIn("-", timeutil.now_time())
215 def test_ignore_case_get(self):
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'))