1 # Copyright 2017 ZTE Corporation.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
20 from django.test import Client
21 from rest_framework import status
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
29 class WorkflowViewTest(unittest.TestCase):
31 self.client = Client()
32 WFPlanModel.objects.filter().delete()
37 @mock.patch.object(restcall, 'upload_by_msb')
38 def test_deploy_workflow(self, mock_upload_by_msb):
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")))
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({
58 mock_upload_by_msb.return_value = [0, json.JSONEncoder().encode({
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")))
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))
78 @mock.patch.object(restcall, 'call_req')
79 def test_buildin_workflow_normal(self, mock_call_req):
84 "nsInstanceId": ns_inst_id,
85 "object_context": '{"a": "b"}',
86 "object_additionalParamForNs": '{"c": "d"}',
87 "object_additionalParamForVnf": '{"e": "f"}',
91 "sdnControllerId": "2"
94 "api/nslcm/v1/ns/vls":
95 [0, json.JSONEncoder().encode({
100 "api/nslcm/v1/ns/vnfs":
101 [0, json.JSONEncoder().encode({
105 "api/nslcm/v1/ns/vnfs/2":
106 [0, json.JSONEncoder().encode({
107 "vnfStatus": "active"
109 "api/nslcm/v1/ns/sfcs":
110 [0, json.JSONEncoder().encode({
114 "api/nslcm/v1/ns/sfcs/3":
115 [0, json.JSONEncoder().encode({
116 "sfcStatus": "active"
118 "/api/nslcm/v1/jobs/11?responseId=0":
119 [0, json.JSONEncoder().encode({"responseDescriptor": {
122 "statusDescription": "ok"
124 "/api/nslcm/v1/jobs/111?responseId=0":
125 [0, json.JSONEncoder().encode({"responseDescriptor": {
128 "statusDescription": "ok"
130 "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
132 "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
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))
142 @mock.patch.object(restcall, 'call_req')
143 def test_buildin_workflow_when_create_vl_failed(self, mock_call_req):
148 "nsInstanceId": ns_inst_id,
149 "object_context": '{"a": "b"}',
150 "object_additionalParamForNs": '{"c": "d"}',
151 "object_additionalParamForVnf": '{"e": "f"}',
155 "sdnControllerId": "2"
158 "api/nslcm/v1/ns/vls":
159 [0, json.JSONEncoder().encode({
164 "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
166 "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
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))
176 @mock.patch.object(restcall, 'call_req')
177 def test_buildin_workflow_when_create_vnf_failed(self, mock_call_req):
182 "nsInstanceId": ns_inst_id,
183 "object_context": '{"a": "b"}',
184 "object_additionalParamForNs": '{"c": "d"}',
185 "object_additionalParamForVnf": '{"e": "f"}',
189 "sdnControllerId": "2"
192 "api/nslcm/v1/ns/vls":
193 [0, json.JSONEncoder().encode({
198 "api/nslcm/v1/ns/vnfs":
199 [0, json.JSONEncoder().encode({
203 "api/nslcm/v1/ns/vnfs/2":
204 [0, json.JSONEncoder().encode({
207 "/api/nslcm/v1/jobs/11?responseId=0":
208 [0, json.JSONEncoder().encode({"responseDescriptor": {
211 "statusDescription": "ok"
213 "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
215 "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
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))