lcm/jobs/tests/tests.py

   1 # Copyright 2016 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 from django.test import TestCase
  16 from lcm.jobs.const import JOB_INSTANCE_URI
  17 from lcm.jobs.enum import JOB_ACTION, JOB_STATUS, JOB_TYPE
  18 from lcm.jobs.tests import UPDATE_JOB_DICT, UPDATE_JOB_BAD_REQ_DICT
  19 from lcm.pub.database.models import JobModel, JobStatusModel
  20 from rest_framework import status
  21 from rest_framework.test import APIClient
  22
  23
  24 class JobsViewTest(TestCase):
  25     def setUp(self):
  26         self.job_id = 'test_job_id'
  27         self.client = APIClient()
  28         JobModel.objects.all().delete()
  29         JobStatusModel.objects.all().delete()
  30
  31     def tearDown(self):
  32         JobModel.objects.all().delete()
  33         JobStatusModel.objects.all().delete()
  34
  35     def test_query_ns_job(self):
  36         JobModel(jobid=self.job_id,
  37                  jobtype=JOB_TYPE.NS,
  38                  jobaction=JOB_ACTION.INSTANTIATE,
  39                  resid='1').save()
  40         JobStatusModel(indexid=1,
  41                        jobid=self.job_id,
  42                        status=JOB_STATUS.PROCESSING,
  43                        progress=20,
  44                        descp='Finish to instantiate NS.',
  45                        errcode="0").save()
  46         response = self.client.get(JOB_INSTANCE_URI % self.job_id)
  47         self.assertEqual(status.HTTP_200_OK, response.status_code, response.data)
  48         self.assertIn('jobId', response.data)
  49         self.assertIn('responseDescriptor', response.data)
  50         self.assertEqual(20, response.data['responseDescriptor']['progress'])
  51
  52     def test_query_ns_job_not_existed(self):
  53         job_id = 'test_job_id_not_existed'
  54         response = self.client.get(JOB_INSTANCE_URI % job_id)
  55         self.assertEqual(status.HTTP_200_OK, response.status_code)
  56         self.assertIn('jobId', response.data)
  57         self.assertNotIn('responseDescriptor', response.data)
  58
  59     def test_query_job_with_response_id(self):
  60         JobModel(jobid=self.job_id,
  61                  jobtype=JOB_TYPE.NS,
  62                  jobaction=JOB_ACTION.INSTANTIATE,
  63                  resid='1').save()
  64         JobStatusModel(indexid=1,
  65                        jobid=self.job_id,
  66                        status=JOB_STATUS.PROCESSING,
  67                        progress=20,
  68                        descp='NS instantiation progress is 20%.',
  69                        errcode="0").save()
  70         JobStatusModel(indexid=2,
  71                        jobid=self.job_id,
  72                        status=JOB_STATUS.PROCESSING,
  73                        progress=50,
  74                        descp='NS instantiation progress is 50%.',
  75                        errcode="0").save()
  76         JobStatusModel(indexid=3,
  77                        jobid=self.job_id,
  78                        status=JOB_STATUS.PROCESSING,
  79                        progress=80,
  80                        descp='NS instantiation progress is 80%.',
  81                        errcode="0").save()
  82         JobStatusModel(indexid=4,
  83                        jobid=self.job_id,
  84                        status=JOB_STATUS.FINISH,
  85                        progress=100,
  86                        descp='Finish to instantiate NS.',
  87                        errcode="0").save()
  88         url = JOB_INSTANCE_URI % self.job_id + "?responseId=2"
  89         response = self.client.get(url)
  90         self.assertEqual(status.HTTP_200_OK, response.status_code)
  91         self.assertEqual(self.job_id, response.data.get('jobId'))
  92         self.assertIn('responseDescriptor', response.data)
  93         self.assertEqual(100, response.data['responseDescriptor']['progress'])
  94         self.assertEqual(1, len(response.data['responseDescriptor']['responseHistoryList']))
  95
  96     def test_update_job(self):
  97         JobModel(
  98             jobid=self.job_id,
  99             jobtype=JOB_TYPE.NS,
 100             jobaction=JOB_ACTION.INSTANTIATE,
 101             resid='1').save()
 102         JobStatusModel(
 103             indexid=1,
 104             jobid=self.job_id,
 105             status=JOB_STATUS.PROCESSING,
 106             progress=20,
 107             descp='NS instantiation progress is 20%.',
 108             errcode="0").save()
 109         response = self.client.post(JOB_INSTANCE_URI % self.job_id, format='json', data=UPDATE_JOB_DICT)
 110         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code)
 111
 112     def test_update_job_not_existed(self):
 113         response = self.client.post(JOB_INSTANCE_URI % self.job_id, format='json', data=UPDATE_JOB_DICT)
 114         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 115
 116     def test_update_job_with_bad_req(self):
 117         JobModel(
 118             jobid=self.job_id,
 119             jobtype=JOB_TYPE.NS,
 120             jobaction=JOB_ACTION.INSTANTIATE,
 121             resid='1').save()
 122         JobStatusModel(
 123             indexid=1,
 124             jobid=self.job_id,
 125             status=JOB_STATUS.PROCESSING,
 126             progress=20,
 127             descp='NS instantiation progress is 20%.',
 128             errcode="0").save()
 129         response = self.client.post(JOB_INSTANCE_URI % self.job_id, format='json', data=UPDATE_JOB_BAD_REQ_DICT)
 130         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)