allow external port for SDN net
[osm/NBI.git] / osm_nbi / tests / test_instance_topics.py
1 #
2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
3 # not use this file except in compliance with the License. You may obtain
4 # a copy of the License at
5 #
6 # http://www.apache.org/licenses/LICENSE-2.0
7 #
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
11 # License for the specific language governing permissions and limitations
12 # under the License.
13 #
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
16 ##
17
18 import unittest
19 from unittest.mock import Mock, mock_open # patch, MagicMock
20 from osm_common.dbbase import DbException
21 from osm_nbi.engine import EngineException
22 from osm_common.dbmemory import DbMemory
23 from osm_common.fsbase import FsBase
24 from osm_common.msgbase import MsgBase
25 from http import HTTPStatus
26 from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
27 from osm_nbi.tests.test_db_descriptors import db_vim_accounts_text, db_nsds_text, db_vnfds_text, db_nsrs_text,\
28 db_vnfrs_text
29 from copy import deepcopy
30 import yaml
31
32
33 class TestNsLcmOpTopic(unittest.TestCase):
34
35 def setUp(self):
36 self.db = DbMemory()
37 self.fs = Mock(FsBase())
38 self.fs.get_params.return_value = {"./fake/folder"}
39 self.fs.file_open = mock_open()
40 self.msg = Mock(MsgBase())
41 # create class
42 self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
43 self.nslcmop_topic.check_quota = Mock(return_value=None) # skip quota
44
45 self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
46 self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
47 self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
48 self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
49 self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))
50 self.db.create = Mock(return_value="created_id")
51 self.db.set_one = Mock(return_value={"updated": 1})
52 self.nsd = self.db.get_list("nsds")[0]
53 self.nsd_id = self.nsd["_id"]
54 self.nsr = self.db.get_list("nsrs")[0]
55 self.nsr_id = self.nsr["_id"]
56 self.nsr_project = self.nsr["_admin"]["projects_read"][0]
57
58 self.vim = self.db.get_list("vim_accounts")[0]
59 self.vim_id = self.vim["_id"]
60
61 def test_create_instantiate(self):
62 session = {"force": False, "admin": False, "public": False, "project_id": [self.nsr_project], "method": "write"}
63 indata = {
64 "nsdId": self.nsd_id,
65 "nsInstanceId": self.nsr_id,
66 "nsName": "name",
67 "vimAccountId": self.vim_id,
68 "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
69 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}],
70 "vnf": [{"member-vnf-index": "1",
71 "vdu": [{"id": "dataVM", "interface": [{"name": "dataVM-eth0",
72 "ip-address": "10.11.12.13",
73 "floating-ip-required": True}]
74 }],
75 "internal-vld": [{"name": "internal", "vim-network-id": "vim-net-id"}]
76 }],
77 "lcmOperationType": "instantiate",
78
79 }
80 rollback = []
81 headers = {}
82
83 nslcmop_id, _ = self.nslcmop_topic.new(rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers)
84
85 # check nslcmop is created at database
86 self.assertEqual(self.db.create.call_count, 1, "database create not called, or called more than once")
87 _call = self.db.create.call_args_list[0]
88 self.assertEqual(_call[0][0], "nslcmops", "must be create a nslcmops entry at database")
89
90 created_nslcmop = _call[0][1]
91 self.assertEqual(nslcmop_id, created_nslcmop["_id"], "mismatch between return id and database '_id'")
92 self.assertEqual(self.nsr_id, created_nslcmop["nsInstanceId"], "bad reference id from nslcmop to nsr")
93 self.assertTrue(created_nslcmop["_admin"].get("projects_read"),
94 "Database record must contain '_amdin.projects_read'")
95 self.assertIn("created", created_nslcmop["_admin"], "Database record must contain '_admin.created'")
96 self.assertTrue(created_nslcmop["lcmOperationType"] == "instantiate",
97 "Database record must contain 'lcmOperationType=instantiate'")
98
99 self.assertEqual(len(rollback), len(self.db.set_one.call_args_list) + 1,
100 "rollback mismatch with created/set items at database")
101
102 # test parameters with error
103 bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
104 test_set = (
105 ("nsr not found", {"nsInstanceId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
106 # TODO add "nsd"
107 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
108 ("bad member-vnf-index", {"vnf.0.member-vnf-index": "k"}, EngineException, HTTPStatus.BAD_REQUEST,
109 ("k",)),
110 )
111 for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
112 with self.assertRaises(expect_exc, msg=message) as e:
113 self.nslcmop_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
114 if expect_code:
115 self.assertTrue(e.exception.http_code == expect_code)
116 if expect_text_list:
117 for expect_text in expect_text_list:
118 self.assertIn(expect_text, str(e.exception).lower(),
119 "Expected '{}' at exception text".format(expect_text))
120
121 def test_check_ns_operation_action(self):
122 nsrs = self.db.get_list("nsrs")[0]
123 session = {}
124
125 indata = {
126 "member_vnf_index": "1",
127 "vdu_id": None,
128 "primitive": "touch",
129 "primitive_params": {"filename": "file"}
130 }
131
132 self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata)
133 for k in indata:
134 indata_copy = indata.copy()
135 if k == "primitive_params":
136 continue
137 indata_copy[k] = "non_existing"
138 with self.assertRaises(EngineException) as exc_manager:
139 self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata_copy)
140 exc = exc_manager.exception
141 self.assertEqual(exc.http_code, HTTPStatus.BAD_REQUEST, "Engine exception bad http_code with {}".
142 format(indata_copy))
143
144
145 class TestNsrTopic(unittest.TestCase):
146
147 def setUp(self):
148 self.db = DbMemory()
149 self.fs = Mock(FsBase())
150 self.fs.get_params.return_value = {"./fake/folder"}
151 self.fs.file_open = mock_open()
152 self.msg = Mock(MsgBase())
153 # create class
154 self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
155 self.nsr_topic.check_quota = Mock(return_value=None) # skip quota
156
157 self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
158 self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
159 self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
160 self.db.create = Mock(return_value="created_id")
161 self.nsd = self.db.get_list("nsds")[0]
162 self.nsd_id = self.nsd["_id"]
163 self.nsd_project = self.nsd["_admin"]["projects_read"][0]
164
165 self.vim = self.db.get_list("vim_accounts")[0]
166 self.vim_id = self.vim["_id"]
167
168 def test_create(self):
169 session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
170 indata = {
171 "nsdId": self.nsd_id,
172 "nsName": "name",
173 "vimAccountId": self.vim_id,
174 "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
175 {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}]
176 }
177 rollback = []
178 headers = {}
179
180 self.nsr_topic.new(rollback, session, indata=indata, kwargs=None, headers=headers)
181
182 # check vnfrs and nsrs created in whatever order
183 created_vnfrs = []
184 created_nsrs = []
185 nsr_id = None
186 for _call in self.db.create.call_args_list:
187 assert len(_call[0]) >= 2, "called db.create with few parameters"
188 created_item = _call[0][1]
189 if _call[0][0] == "vnfrs":
190 created_vnfrs.append(created_item)
191 self.assertIn("member-vnf-index-ref", created_item,
192 "Created item must contain member-vnf-index-ref section")
193 if nsr_id:
194 self.assertEqual(nsr_id, created_item["nsr-id-ref"], "bad reference id from vnfr to nsr")
195 else:
196 nsr_id = created_item["nsr-id-ref"]
197
198 elif _call[0][0] == "nsrs":
199 created_nsrs.append(created_item)
200 if nsr_id:
201 self.assertEqual(nsr_id, created_item["_id"], "bad reference id from vnfr to nsr")
202 else:
203 nsr_id = created_item["_id"]
204 else:
205 assert True, "created an unknown record {} at database".format(_call[0][0])
206
207 self.assertTrue(created_item["_admin"].get("projects_read"),
208 "Database record must contain '_amdin.projects_read'")
209 self.assertIn("created", created_item["_admin"], "Database record must contain '_admin.created'")
210 self.assertTrue(created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
211 "Database record must contain '_admin.nstate=NOT INSTANTIATE'")
212
213 self.assertEqual(len(created_vnfrs), len(self.nsd["constituent-vnfd"]),
214 "created a mismatch number of vnfr at database")
215 self.assertEqual(len(created_nsrs), 1, "Only one nsrs must be created at database")
216 self.assertEqual(len(rollback), len(created_vnfrs) + 1, "rollback mismatch with created items at database")
217
218 # test parameters with error
219 bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
220 test_set = (
221 # TODO add "nsd"
222 ("nsd not found", {"nsdId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
223 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim"
224 ("additional params not supply", {"additionalParamsForVnf.0.member-vnf-index": "k"}, EngineException,
225 HTTPStatus.BAD_REQUEST, None),
226 )
227 for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
228 with self.assertRaises(expect_exc, msg=message) as e:
229 self.nsr_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
230 if expect_code:
231 self.assertTrue(e.exception.http_code == expect_code)
232 if expect_text_list:
233 for expect_text in expect_text_list:
234 self.assertIn(expect_text, str(e.exception).lower(),
235 "Expected '{}' at exception text".format(expect_text))