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