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
28 class WorkflowViewTest(unittest.TestCase):
30 self.client = Client()
31 WFPlanModel.objects.filter().delete()
36 @mock.patch.object(restcall, 'upload_by_msb')
37 def test_deploy_workflow(self, mock_upload_by_msb):
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")))
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({
57 mock_upload_by_msb.return_value = [0, json.JSONEncoder().encode({
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")))
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))
77 @mock.patch.object(restcall, 'call_req')
78 def test_buildin_workflow_normal(self, mock_call_req):
83 "nsInstanceId": ns_inst_id,
84 "object_context": '{"a": "b"}',
85 "object_additionalParamForNs": '{"c": "d"}',
86 "object_additionalParamForVnf": '{"e": "f"}',
90 "sdnControllerId": "2"
93 "api/nslcm/v1/ns/vls":
94 [0, json.JSONEncoder().encode({
99 "api/nslcm/v1/ns/vnfs":
100 [0, json.JSONEncoder().encode({
104 "api/nslcm/v1/ns/vnfs/2":
105 [0, json.JSONEncoder().encode({
106 "vnfStatus": "active"
108 "api/nslcm/v1/ns/sfcs":
109 [0, json.JSONEncoder().encode({
113 "api/nslcm/v1/ns/sfcs/3":
114 [0, json.JSONEncoder().encode({
115 "sfcStatus": "active"
117 "/api/nslcm/v1/jobs/11?responseId=0":
118 [0, json.JSONEncoder().encode({"responseDescriptor": {
121 "statusDescription": "ok"
123 "/api/nslcm/v1/jobs/111?responseId=0":
124 [0, json.JSONEncoder().encode({"responseDescriptor": {
127 "statusDescription": "ok"
129 "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
131 "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
135 def side_effect(*args):
136 return mock_vals[args[4]]
137 mock_call_req.side_effect = side_effect
139 self.assertTrue(build_in.run_ns_instantiate(wf_input))
141 @mock.patch.object(restcall, 'call_req')
142 def test_buildin_workflow_when_create_vl_failed(self, mock_call_req):
147 "nsInstanceId": ns_inst_id,
148 "object_context": '{"a": "b"}',
149 "object_additionalParamForNs": '{"c": "d"}',
150 "object_additionalParamForVnf": '{"e": "f"}',
154 "sdnControllerId": "2"
157 "api/nslcm/v1/ns/vls":
158 [0, json.JSONEncoder().encode({
163 "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
165 "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
169 def side_effect(*args):
170 return mock_vals[args[4]]
171 mock_call_req.side_effect = side_effect
173 self.assertFalse(build_in.run_ns_instantiate(wf_input))
175 @mock.patch.object(restcall, 'call_req')
176 def test_buildin_workflow_when_create_vnf_failed(self, mock_call_req):
181 "nsInstanceId": ns_inst_id,
182 "object_context": '{"a": "b"}',
183 "object_additionalParamForNs": '{"c": "d"}',
184 "object_additionalParamForVnf": '{"e": "f"}',
188 "sdnControllerId": "2"
191 "api/nslcm/v1/ns/vls":
192 [0, json.JSONEncoder().encode({
197 "api/nslcm/v1/ns/vnfs":
198 [0, json.JSONEncoder().encode({
202 "api/nslcm/v1/ns/vnfs/2":
203 [0, json.JSONEncoder().encode({
206 "/api/nslcm/v1/jobs/11?responseId=0":
207 [0, json.JSONEncoder().encode({"responseDescriptor": {
210 "statusDescription": "ok"
212 "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id):
214 "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id):
218 def side_effect(*args):
219 return mock_vals[args[4]]
220 mock_call_req.side_effect = side_effect
222 self.assertFalse(build_in.run_ns_instantiate(wf_input))