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