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 unittest
  16 import json
  17 import mock
  18 import os
  19 from django.test import Client
  20 from rest_framework import status
  21
  22 from lcm.pub.utils import restcall
  23 from lcm.pub.database.models import WFPlanModel, JobStatusModel
  24 from lcm.pub.utils.jobutil import JobUtil
  25 from lcm.workflows import build_in
  26
  27 class WorkflowViewTest(unittest.TestCase):
  28     def setUp(self):
  29         self.client = Client()
  30         WFPlanModel.objects.filter().delete()
  31
  32     def tearDown(self):
  33         pass
  34
  35     @mock.patch.object(restcall, 'upload_by_msb')
  36     def test_deploy_workflow(self, mock_upload_by_msb):
  37         mock_upload_by_msb.return_value = [0, json.JSONEncoder().encode({
  38             "status": "1",
  39             "message": "2",
  40             "deployedId": "3",
  41             "processId": "4"
  42             }), '202']
  43         response = self.client.post("/api/nslcm/v1/workflow",
  44             {"filePath": os.path.abspath(__file__)}, format='json')
  45         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
  46         self.assertEqual(1, len(WFPlanModel.objects.filter(deployed_id="3")))
  47
  48     @mock.patch.object(restcall, 'upload_by_msb')
  49     @mock.patch.object(restcall, 'call_req')
  50     def test_force_deploy_workflow(self, mock_call_req, mock_upload_by_msb):
  51         mock_call_req.return_value = [0, json.JSONEncoder().encode({
  52             "status": "1",
  53             "message": "2"
  54             }), '202']
  55         mock_upload_by_msb.return_value = [0, json.JSONEncoder().encode({
  56             "status": "2",
  57             "message": "3",
  58             "deployedId": "4",
  59             "processId": "5"
  60             }), '202']
  61         WFPlanModel(deployed_id="1", process_id="2", status="3", message="4").save()
  62         response = self.client.post("/api/nslcm/v1/workflow",
  63             {"filePath": os.path.abspath(__file__), "forceDeploy": "True"}, format='json')
  64         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
  65         self.assertEqual(0, len(WFPlanModel.objects.filter(deployed_id="1")))
  66         self.assertEqual(1, len(WFPlanModel.objects.filter(deployed_id="4")))
  67
  68     def test_deploy_workflow_when_already_deployed(self):
  69         WFPlanModel(deployed_id="1", process_id="2", status="3", message="4").save()
  70         response = self.client.post("/api/nslcm/v1/workflow",
  71             {"filePath": os.path.abspath(__file__)}, format='json')
  72         self.assertEqual(status.HTTP_202_ACCEPTED, response.status_code, response.content)
  73         self.assertEqual({'msg': 'Already deployed.'}, json.loads(response.content))
  74
  75     @mock.patch.object(restcall, 'call_req')
  76     def test_buildin_workflow_normal(self, mock_call_req):
  77         ns_inst_id = "1"
  78         job_id = "1234"
  79         wf_input = {
  80             "jobId": job_id,
  81             "nsInstanceId": ns_inst_id,
  82             "object_context": '{"a": "b"}',
  83             "object_additionalParamForNs": '{"c": "d"}',
  84             "object_additionalParamForVnf": '{"e": "f"}',
  85             "vlCount": 1,
  86             "vnfCount": 1,
  87             "sfcCount": 1,
  88             "sdnControllerId": "2"
  89         }
  90         mock_vals = {
  91             "api/nslcm/v1/ns/vls":
  92                 [0, json.JSONEncoder().encode({
  93                     "result": "0",
  94                     "detail": "vl1",
  95                     "vlId": "1"
  96                     }), '201'],
  97             "api/nslcm/v1/ns/vnfs":
  98                 [0, json.JSONEncoder().encode({
  99                     "vnfInstId": "2",
 100                     "jobId": "11"
 101                     }), '201'],
 102             "api/nslcm/v1/ns/vnfs/2":
 103                 [0, json.JSONEncoder().encode({
 104                     "vnfStatus": "active"
 105                     }), '201'],
 106             "api/nslcm/v1/ns/sfcs":
 107                 [0, json.JSONEncoder().encode({
 108                     "sfcInstId": "3",
 109                     "jobId": "111"
 110                     }), '201'],
 111             "api/nslcm/v1/ns/sfcs/3":
 112                 [0, json.JSONEncoder().encode({
 113                     "sfcStatus": "active"
 114                     }), '201'],
 115             "/api/nslcm/v1/jobs/11?responseId=0":
 116                 [0, json.JSONEncoder().encode({"responseDescriptor": {
 117                     "responseId": "1",
 118                     "progress": 100,
 119                     "statusDescription": "ok"
 120                     }}), '200'],
 121             "/api/nslcm/v1/jobs/111?responseId=0":
 122                 [0, json.JSONEncoder().encode({"responseDescriptor": {
 123                     "responseId": "1",
 124                     "progress": 100,
 125                     "statusDescription": "ok"
 126                     }}), '200'],
 127             "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
 128                 [0, '{}', '201'],
 129             "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
 130                 [0, '{}', '201']
 131         }
 132
 133         def side_effect(*args):
 134             return mock_vals[args[4]]
 135         mock_call_req.side_effect = side_effect
 136
 137         self.assertTrue(build_in.run_ns_instantiate(wf_input))
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149