lcm/workflows/tests.py

   1 # Copyright 2017 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 json
  16 import os
  17 import unittest
  18
  19 import mock
  20 from django.test import Client
  21 from rest_framework import status
  22
  23 from lcm.pub.database.models import WFPlanModel
  24 from lcm.pub.utils import restcall
  25 from lcm.workflows import build_in
  26 from lcm.ns.biz.ns_lcm_op_occ import NsLcmOpOcc
  27
  28
  29 class WorkflowViewTest(unittest.TestCase):
  30     def setUp(self):
  31         self.client = Client()
  32         WFPlanModel.objects.filter().delete()
  33
  34     def tearDown(self):
  35         pass
  36
  37     @mock.patch.object(restcall, 'upload_by_msb')
  38     def test_deploy_workflow(self, mock_upload_by_msb):
  39         res_data = {
  40             "status": "1",
  41             "message": "2",
  42             "deployedId": "3",
  43             "processId": "4"
  44         }
  45         mock_upload_by_msb.return_value = [0, json.JSONEncoder().encode(res_data), '202']
  46         response = self.client.post("/api/nslcm/v1/workflow",
  47                                     {"filePath": os.path.abspath(__file__)}, format='json')
  48         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
  49         self.assertEqual(1, len(WFPlanModel.objects.filter(deployed_id="3")))
  50
  51     @mock.patch.object(restcall, 'upload_by_msb')
  52     @mock.patch.object(restcall, 'call_req')
  53     def test_force_deploy_workflow(self, mock_call_req, mock_upload_by_msb):
  54         mock_call_req.return_value = [0, json.JSONEncoder().encode({
  55             "status": "1",
  56             "message": "2"
  57         }), '202']
  58         mock_upload_by_msb.return_value = [0, json.JSONEncoder().encode({
  59             "status": "2",
  60             "message": "3",
  61             "deployedId": "4",
  62             "processId": "5"
  63         }), '202']
  64         WFPlanModel(deployed_id="1", process_id="2", status="3", message="4").save()
  65         response = self.client.post("/api/nslcm/v1/workflow",
  66                                     {"filePath": os.path.abspath(__file__), "forceDeploy": "True"}, format='json')
  67         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
  68         self.assertEqual(0, len(WFPlanModel.objects.filter(deployed_id="1")))
  69         self.assertEqual(1, len(WFPlanModel.objects.filter(deployed_id="4")))
  70
  71     def test_deploy_workflow_when_already_deployed(self):
  72         WFPlanModel(deployed_id="1", process_id="2", status="3", message="4").save()
  73         response = self.client.post("/api/nslcm/v1/workflow",
  74                                     {"filePath": os.path.abspath(__file__)}, format='json')
  75         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
  76         self.assertEqual({'msg': 'Already deployed.'}, json.loads(response.content))
  77
  78     @mock.patch.object(restcall, 'call_req')
  79     def test_buildin_workflow_normal(self, mock_call_req):
  80         ns_inst_id = "1"
  81         job_id = "1234"
  82         wf_input = {
  83             "jobId": job_id,
  84             "nsInstanceId": ns_inst_id,
  85             "object_context": '{"a": "b"}',
  86             "object_additionalParamForNs": '{"c": "d"}',
  87             "object_additionalParamForVnf": '{"e": "f"}',
  88             "vlCount": 1,
  89             "vnfCount": 1,
  90             "sfcCount": 1,
  91             "sdnControllerId": "2"
  92         }
  93         mock_vals = {
  94             "api/nslcm/v1/ns/vls":
  95                 [0, json.JSONEncoder().encode({
  96                     "result": "0",
  97                     "detail": "vl1",
  98                     "vlId": "1"
  99                 }), '201'],
 100             "api/nslcm/v1/ns/vnfs":
 101                 [0, json.JSONEncoder().encode({
 102                     "vnfInstId": "2",
 103                     "jobId": "11"
 104                 }), '201'],
 105             "api/nslcm/v1/ns/vnfs/2":
 106                 [0, json.JSONEncoder().encode({
 107                     "vnfStatus": "active"
 108                 }), '201'],
 109             "api/nslcm/v1/ns/sfcs":
 110                 [0, json.JSONEncoder().encode({
 111                     "sfcInstId": "3",
 112                     "jobId": "111"
 113                 }), '201'],
 114             "api/nslcm/v1/ns/sfcs/3":
 115                 [0, json.JSONEncoder().encode({
 116                     "sfcStatus": "active"
 117                 }), '201'],
 118             "/api/nslcm/v1/jobs/11?responseId=0":
 119                 [0, json.JSONEncoder().encode({"responseDescriptor": {
 120                     "responseId": "1",
 121                     "progress": 100,
 122                     "statusDescription": "ok"
 123                 }}), '200'],
 124             "/api/nslcm/v1/jobs/111?responseId=0":
 125                 [0, json.JSONEncoder().encode({"responseDescriptor": {
 126                     "responseId": "1",
 127                     "progress": 100,
 128                     "statusDescription": "ok"
 129                 }}), '200'],
 130             "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
 131                 [0, '{}', '201'],
 132             "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
 133                 [0, '{}', '201']
 134         }
 135
 136         def side_effect(*args):
 137             return mock_vals[args[4]]
 138         mock_call_req.side_effect = side_effect
 139         occ_id = NsLcmOpOcc.create(ns_inst_id, "INSTANTIATE", "PROCESSING", False, wf_input)
 140         self.assertTrue(build_in.run_ns_instantiate(wf_input, occ_id))
 141
 142     @mock.patch.object(restcall, 'call_req')
 143     def test_buildin_workflow_when_create_vl_failed(self, mock_call_req):
 144         ns_inst_id = "1"
 145         job_id = "1234"
 146         wf_input = {
 147             "jobId": job_id,
 148             "nsInstanceId": ns_inst_id,
 149             "object_context": '{"a": "b"}',
 150             "object_additionalParamForNs": '{"c": "d"}',
 151             "object_additionalParamForVnf": '{"e": "f"}',
 152             "vlCount": 1,
 153             "vnfCount": 1,
 154             "sfcCount": 1,
 155             "sdnControllerId": "2"
 156         }
 157         mock_vals = {
 158             "api/nslcm/v1/ns/vls":
 159                 [0, json.JSONEncoder().encode({
 160                     "result": "1",
 161                     "detail": "vl1",
 162                     "vlId": "1"
 163                 }), '201'],
 164             "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
 165                 [0, '{}', '201'],
 166             "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
 167                 [0, '{}', '201']
 168         }
 169
 170         def side_effect(*args):
 171             return mock_vals[args[4]]
 172         mock_call_req.side_effect = side_effect
 173         occ_id = NsLcmOpOcc.create(ns_inst_id, "INSTANTIATE", "PROCESSING", False, wf_input)
 174         self.assertFalse(build_in.run_ns_instantiate(wf_input, occ_id))
 175
 176     @mock.patch.object(restcall, 'call_req')
 177     def test_buildin_workflow_when_create_vnf_failed(self, mock_call_req):
 178         ns_inst_id = "1"
 179         job_id = "1234"
 180         wf_input = {
 181             "jobId": job_id,
 182             "nsInstanceId": ns_inst_id,
 183             "object_context": '{"a": "b"}',
 184             "object_additionalParamForNs": '{"c": "d"}',
 185             "object_additionalParamForVnf": '{"e": "f"}',
 186             "vlCount": 1,
 187             "vnfCount": 1,
 188             "sfcCount": 1,
 189             "sdnControllerId": "2"
 190         }
 191         mock_vals = {
 192             "api/nslcm/v1/ns/vls":
 193                 [0, json.JSONEncoder().encode({
 194                     "result": "0",
 195                     "detail": "vl1",
 196                     "vlId": "1"
 197                 }), '201'],
 198             "api/nslcm/v1/ns/vnfs":
 199                 [0, json.JSONEncoder().encode({
 200                     "vnfInstId": "2",
 201                     "jobId": "11"
 202                 }), '201'],
 203             "api/nslcm/v1/ns/vnfs/2":
 204                 [0, json.JSONEncoder().encode({
 205                     "vnfStatus": "error"
 206                 }), '201'],
 207             "/api/nslcm/v1/jobs/11?responseId=0":
 208                 [0, json.JSONEncoder().encode({"responseDescriptor": {
 209                     "responseId": "1",
 210                     "progress": 100,
 211                     "statusDescription": "ok"
 212                 }}), '200'],
 213             "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
 214                 [0, '{}', '201'],
 215             "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
 216                 [0, '{}', '201']
 217         }
 218
 219         def side_effect(*args):
 220             return mock_vals[args[4]]
 221         mock_call_req.side_effect = side_effect
 222         occ_id = NsLcmOpOcc.create(ns_inst_id, "INSTANTIATE", "PROCESSING", False, wf_input)
 223         self.assertFalse(build_in.run_ns_instantiate(wf_input, occ_id))