return ro_ip_profile
+ @staticmethod
+ def _process_net_params(
+ target_vld: Dict[str, Any],
+ indata: Dict[str, Any],
+ vim_info: Dict[str, Any],
+ target_record_id: str,
+ ) -> Dict[str, Any]:
+ """Function to process network parameters.
+
+ Args:
+ target_vld (Dict[str, Any]): [description]
+ indata (Dict[str, Any]): [description]
+ vim_info (Dict[str, Any]): [description]
+ target_record_id (str): [description]
+
+ Returns:
+ Dict[str, Any]: [description]
+ """
+ extra_dict = {}
+
+ if vim_info.get("sdn"):
+ # vnf_preffix = "vnfrs:{}".format(vnfr_id)
+ # ns_preffix = "nsrs:{}".format(nsr_id)
+ # remove the ending ".sdn
+ vld_target_record_id, _, _ = target_record_id.rpartition(".")
+ extra_dict["params"] = {
+ k: vim_info[k]
+ for k in ("sdn-ports", "target_vim", "vlds", "type")
+ if vim_info.get(k)
+ }
+
+ # TODO needed to add target_id in the dependency.
+ if vim_info.get("target_vim"):
+ extra_dict["depends_on"] = [
+ f"{vim_info.get('target_vim')} {vld_target_record_id}"
+ ]
+
+ return extra_dict
+
+ if vim_info.get("vim_network_name"):
+ extra_dict["find_params"] = {
+ "filter_dict": {
+ "name": vim_info.get("vim_network_name"),
+ },
+ }
+ elif vim_info.get("vim_network_id"):
+ extra_dict["find_params"] = {
+ "filter_dict": {
+ "id": vim_info.get("vim_network_id"),
+ },
+ }
+ elif target_vld.get("mgmt-network"):
+ extra_dict["find_params"] = {
+ "mgmt": True,
+ "name": target_vld["id"],
+ }
+ else:
+ # create
+ extra_dict["params"] = {
+ "net_name": (
+ f"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
+ ),
+ "ip_profile": Ns._ip_profile_to_ro(vim_info.get("ip_profile")),
+ "provider_network_profile": vim_info.get("provider_network"),
+ }
+
+ if not target_vld.get("underlay"):
+ extra_dict["params"]["net_type"] = "bridge"
+ else:
+ extra_dict["params"]["net_type"] = (
+ "ptp" if target_vld.get("type") == "ELINE" else "data"
+ )
+
+ return extra_dict
+
def deploy(self, session, indata, version, nsr_id, *args, **kwargs):
self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata))
validate_input(indata, deploy_schema)
index += 1
- def _process_net_params(target_vld, indata, vim_info, target_record_id):
- extra_dict = {}
-
- if vim_info.get("sdn"):
- # vnf_preffix = "vnfrs:{}".format(vnfr_id)
- # ns_preffix = "nsrs:{}".format(nsr_id)
- # remove the ending ".sdn
- vld_target_record_id, _, _ = target_record_id.rpartition(".")
- extra_dict["params"] = {
- k: vim_info[k]
- for k in ("sdn-ports", "target_vim", "vlds", "type")
- if vim_info.get(k)
- }
-
- # TODO needed to add target_id in the dependency.
- if vim_info.get("target_vim"):
- extra_dict["depends_on"] = [
- vim_info.get("target_vim") + " " + vld_target_record_id
- ]
-
- return extra_dict
-
- if vim_info.get("vim_network_name"):
- extra_dict["find_params"] = {
- "filter_dict": {"name": vim_info.get("vim_network_name")}
- }
- elif vim_info.get("vim_network_id"):
- extra_dict["find_params"] = {
- "filter_dict": {"id": vim_info.get("vim_network_id")}
- }
- elif target_vld.get("mgmt-network"):
- extra_dict["find_params"] = {"mgmt": True, "name": target_vld["id"]}
- else:
- # create
- extra_dict["params"] = {
- "net_name": "{}-{}".format(
- indata["name"][:16],
- target_vld.get("name", target_vld["id"])[:16],
- ),
- "ip_profile": Ns._ip_profile_to_ro(vim_info.get("ip_profile")),
- "provider_network_profile": vim_info.get("provider_network"),
- }
-
- if not target_vld.get("underlay"):
- extra_dict["params"]["net_type"] = "bridge"
- else:
- extra_dict["params"]["net_type"] = (
- "ptp" if target_vld.get("type") == "ELINE" else "data"
- )
-
- return extra_dict
-
def _process_vdu_params(target_vdu, indata, vim_info, target_record_id):
nonlocal vnfr_id
nonlocal nsr_id
db_update=db_nsr_update,
db_path="vld",
item="net",
- process_params=_process_net_params,
+ process_params=Ns._process_net_params,
)
step = "process NS images"
db_update=db_vnfrs_update[vnfr["_id"]],
db_path="vld",
item="net",
- process_params=_process_net_params,
+ process_params=Ns._process_net_params,
)
target_list = target_vnf.get("vdur") if target_vnf else None
)
self.assertDictEqual(expected_result, result)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_empty_params(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "name": "vld-name",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {
+ "provider_network": "some-profile-here",
+ }
+ target_record_id = ""
+ expected_result = {
+ "params": {
+ "net_name": "ns-name-vld-name",
+ "net_type": "bridge",
+ "ip_profile": {
+ "some_ip_profile": "here",
+ },
+ "provider_network_profile": "some-profile-here",
+ }
+ }
+
+ ip_profile_to_ro.return_value = {
+ "some_ip_profile": "here",
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertTrue(ip_profile_to_ro.called)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_vim_info_sdn(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "name": "vld-name",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {
+ "sdn": "some-sdn",
+ "sdn-ports": ["some", "ports", "here"],
+ "vlds": ["some", "vlds", "here"],
+ "type": "sdn-type",
+ }
+ target_record_id = "vld.sdn.something"
+ expected_result = {
+ "params": {
+ "sdn-ports": ["some", "ports", "here"],
+ "vlds": ["some", "vlds", "here"],
+ "type": "sdn-type",
+ }
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertFalse(ip_profile_to_ro.called)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_vim_info_sdn_target_vim(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "name": "vld-name",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {
+ "sdn": "some-sdn",
+ "sdn-ports": ["some", "ports", "here"],
+ "vlds": ["some", "vlds", "here"],
+ "target_vim": "some-vim",
+ "type": "sdn-type",
+ }
+ target_record_id = "vld.sdn.something"
+ expected_result = {
+ "depends_on": ["some-vim vld.sdn"],
+ "params": {
+ "sdn-ports": ["some", "ports", "here"],
+ "vlds": ["some", "vlds", "here"],
+ "target_vim": "some-vim",
+ "type": "sdn-type",
+ },
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertFalse(ip_profile_to_ro.called)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_vim_network_name(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "name": "vld-name",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {
+ "vim_network_name": "some-network-name",
+ }
+ target_record_id = "vld.sdn.something"
+ expected_result = {
+ "find_params": {
+ "filter_dict": {
+ "name": "some-network-name",
+ },
+ },
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertFalse(ip_profile_to_ro.called)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_vim_network_id(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "name": "vld-name",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {
+ "vim_network_id": "some-network-id",
+ }
+ target_record_id = "vld.sdn.something"
+ expected_result = {
+ "find_params": {
+ "filter_dict": {
+ "id": "some-network-id",
+ },
+ },
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertFalse(ip_profile_to_ro.called)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_mgmt_network(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "id": "vld-id",
+ "name": "vld-name",
+ "mgmt-network": "some-mgmt-network",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {}
+ target_record_id = "vld.sdn.something"
+ expected_result = {
+ "find_params": {
+ "mgmt": True,
+ "name": "vld-id",
+ },
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertFalse(ip_profile_to_ro.called)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_underlay_eline(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "name": "vld-name",
+ "underlay": "some-underlay-here",
+ "type": "ELINE",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {
+ "provider_network": "some-profile-here",
+ }
+ target_record_id = ""
+ expected_result = {
+ "params": {
+ "ip_profile": {
+ "some_ip_profile": "here",
+ },
+ "net_name": "ns-name-vld-name",
+ "net_type": "ptp",
+ "provider_network_profile": "some-profile-here",
+ }
+ }
+
+ ip_profile_to_ro.return_value = {
+ "some_ip_profile": "here",
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertTrue(ip_profile_to_ro.called)
+
+ @patch("osm_ng_ro.ns.Ns._ip_profile_to_ro")
+ def test__process_net_params_with_underlay_elan(
+ self,
+ ip_profile_to_ro,
+ ):
+ target_vld = {
+ "name": "vld-name",
+ "underlay": "some-underlay-here",
+ "type": "ELAN",
+ }
+ indata = {
+ "name": "ns-name",
+ }
+ vim_info = {
+ "provider_network": "some-profile-here",
+ }
+ target_record_id = ""
+ expected_result = {
+ "params": {
+ "ip_profile": {
+ "some_ip_profile": "here",
+ },
+ "net_name": "ns-name-vld-name",
+ "net_type": "data",
+ "provider_network_profile": "some-profile-here",
+ }
+ }
+
+ ip_profile_to_ro.return_value = {
+ "some_ip_profile": "here",
+ }
+
+ result = Ns._process_net_params(
+ target_vld=target_vld,
+ indata=indata,
+ vim_info=vim_info,
+ target_record_id=target_record_id,
+ )
+
+ self.assertDictEqual(expected_result, result)
+ self.assertTrue(ip_profile_to_ro.called)
--- /dev/null
+#######################################################################################
+# Copyright ETSI Contributors and Others.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#######################################################################################
+---
+other:
+ - |
+ Extraction of _process_net_params() from being a nested function inside Ns.deploy().
+ The _process_net_params() function is now a static method inside the Ns class. This
+ eases the testability of _process_net_params().
+ With this extraction a unit test was introduced to cover the extracted function.
projects / osm / RO.git / commitdiff