From 9f7835f755c0dfb94e9eca4a4da4d452ac694cde Mon Sep 17 00:00:00 2001 From: Aleksandr Taranov Date: Fri, 12 May 2023 15:23:19 +0300 Subject: Create SDNC Endpoints Issue-ID: TEST-395 Signed-off-by: Aleksandr Taranov Change-Id: I688e2e96c9b6f3edee59105dcbd05f31f3ad1325 --- src/onapsdk/sdnc/services.py | 131 ++++++++++ .../sdnc/templates/create_node_netconf_api.json.j2 | 37 +++ .../sdnc/templates/create_service_gr_api.json.j2 | 21 ++ src/onapsdk/sdnc/topology.py | 267 +++++++++++++++++++++ tests/test_sdnc_node.py | 102 ++++++++ tests/test_sdnc_service.py | 188 +++++++++++++++ tests/test_sdnc_topology.py | 235 ++++++++++++++++++ 7 files changed, 981 insertions(+) create mode 100644 src/onapsdk/sdnc/services.py create mode 100644 src/onapsdk/sdnc/templates/create_node_netconf_api.json.j2 create mode 100644 src/onapsdk/sdnc/templates/create_service_gr_api.json.j2 create mode 100644 src/onapsdk/sdnc/topology.py create mode 100644 tests/test_sdnc_node.py create mode 100644 tests/test_sdnc_service.py create mode 100644 tests/test_sdnc_topology.py diff --git a/src/onapsdk/sdnc/services.py b/src/onapsdk/sdnc/services.py new file mode 100644 index 0000000..b16551c --- /dev/null +++ b/src/onapsdk/sdnc/services.py @@ -0,0 +1,131 @@ +"""SDNC services module.""" +# Copyright 2023 Deutsche Telekom AG +# +# 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. +from typing import Any, Dict, Iterable + +from onapsdk.utils.headers_creator import headers_sdnc_creator +from onapsdk.utils.jinja import jinja_env + +from .sdnc_element import SdncElement + + +class Service(SdncElement): + """SDNC service.""" + + headers: Dict[str, str] = headers_sdnc_creator(SdncElement.headers) + + def __init__(self, + service_instance_id: str, + service_data: Dict[str, Any] = None, + service_status: Dict[str, Any] = None) -> None: + """Service information initialization. + + Args: + service_instance_id (str): Service instance id + service_data (Dict[str, Any]): Service data + service_status: Dict[str, Any]: Service status + """ + super().__init__() + self.service_instance_id: str = service_instance_id + self.service_data: Dict[str, Any] = service_data + self.service_status: Dict[str, Any] = service_status + + def __repr__(self) -> str: # noqa + """Service information human-readable string. + + Returns: + str: Service information description + + """ + return (f"Service(service_instance_id={self.service_instance_id}, " + f"service_data={self.service_data}, " + f"service_status={self.service_status}") + + @classmethod + def get_all(cls) -> Iterable["Service"]: + """Get all uploaded services using GENERIC-RESOURCES-API. + + Yields: + Services: Service object + """ + for service in cls.send_message_json( + "GET", + "Get SDNC services", + f"{cls.base_url}/restconf/config/GENERIC-RESOURCE-API:services" + ).get('services', {}).get('service', []): + try: + service_data = service["service-data"] + except KeyError: + service_data = {} + try: + service_status = service["service-status"] + except KeyError: + service_status = {} + yield Service(service_instance_id=service["service-instance-id"], + service_data=service_data, + service_status=service_status + ) + + def create(self) -> None: + """Create service using GENERIC-RESOURCES-API.""" + service_data = self.service_data if self.service_data is not None else "" + service_status = self.service_status if self.service_status is not None else "" + self.send_message( + "POST", + "Create a service using GENERIC-RESOURCES-API", + (f"{self.base_url}/restconf/config/" + "GENERIC-RESOURCE-API:services"), + data=jinja_env().get_template( + "create_service_gr_api.json.j2"). + render( + { + "service": { + "service-instance-id": self.service_instance_id, + "service-data": service_data, + "service-status": service_status + } + } + ) + ) + + def update(self) -> None: + """Update service information by service-instance-id using GENERIC-RESOURCES-API.""" + service_data = self.service_data if len(self.service_data) != 0 else "" + service_status = self.service_status if self.service_status != 0 else "" + self.send_message( + "PUT", + "Update service information by service-instance-id using GENERIC-RESOURCES-API", + (f"{self.base_url}/rests/data/" + f"GENERIC-RESOURCE-API:services/service={self.service_instance_id}"), + data=jinja_env().get_template( + "create_service_gr_api.json.j2"). + render( + { + "service": { + "service-instance-id": self.service_instance_id, + "service-data": service_data, + "service-status": service_status + } + } + ) + ) + + def delete(self) -> None: + """Delete service using GENERIC-RESOURCES-API.""" + self.send_message( + "DELETE", + "Delete a service using GENERIC-RESOURCE-API", + (f"{self.base_url}/rests/data/" + f"GENERIC-RESOURCE-API:services/service={self.service_instance_id}") + ) diff --git a/src/onapsdk/sdnc/templates/create_node_netconf_api.json.j2 b/src/onapsdk/sdnc/templates/create_node_netconf_api.json.j2 new file mode 100644 index 0000000..26d4c51 --- /dev/null +++ b/src/onapsdk/sdnc/templates/create_node_netconf_api.json.j2 @@ -0,0 +1,37 @@ +{ + "node": [ + { + {%- for key, value in node.items() %} + {%- if loop.last %} + {%- if key == "node-id" %} + "{{key}}": "{{node_id}}" + {%- elif key == "netconf-node-topology:host" %}"{{key}}": "{{host_}}" + {%- elif key == "netconf-node-topology:port" %}"{{key}}": {{port_}} + {%- elif key == "netconf-node-topology:username" %}"{{key}}": "{{username_}}" + {%- elif key == "netconf-node-topology:password" %}"{{key}}": "{{password_}}" + {%- else %}"{{key}}": + {%- if value is number %}{{value}} + {%- elif value is boolean %}{{value}} + {%- elif value is string %}{{value}} + {%- elif value is mapping %}{{value|replace("'",'"')|replace("True","true")|replace("False","false")}} + {%- endif %} + {%- endif %} + {%- else %} + {%- if key == "node-id" %} + "{{key}}": "{{node_id}}", + {%- elif key == "netconf-node-topology:host" %}"{{key}}": "{{host_}}", + {%- elif key == "netconf-node-topology:port" %}"{{key}}": {{port_}}, + {%- elif key == "netconf-node-topology:username" %}"{{key}}": "{{username_}}", + {%- elif key == "netconf-node-topology:password" %}"{{key}}": "{{password_}}", + {%- else %}"{{key}}": + {%- if value is number %}{{value}}, + {%- elif value is boolean %}{{value|replace("True","true")}}, + {%- elif value is string %}"{{value}}", + {%- elif value is mapping %}{{value|replace("'",'"')|replace("True","true")|replace("False","false")}}, + {%- endif %} + {%- endif %} + {%- endif %} + {% endfor %} + } + ] +} \ No newline at end of file diff --git a/src/onapsdk/sdnc/templates/create_service_gr_api.json.j2 b/src/onapsdk/sdnc/templates/create_service_gr_api.json.j2 new file mode 100644 index 0000000..5f575be --- /dev/null +++ b/src/onapsdk/sdnc/templates/create_service_gr_api.json.j2 @@ -0,0 +1,21 @@ +{ + "service": [ + { + {%- for key, value in service.items() %} + {%- if loop.last %} + "{{key}}": + {%- if value is mapping %}{{value|replace("'",'"')|replace("True","true")|replace("False","false")}} + {%- elif value == "" %}{} + {%- else %}"{{value}}" + {%- endif %} + {%- else %} + "{{key}}": + {%- if value is mapping %}{{value|replace("'",'"')|replace("True","true")|replace("False","false")}}, + {%- elif value == "" %}{}, + {%- else %}"{{value}}", + {%- endif %} + {%- endif %} + {%- endfor %} + } + ] +} \ No newline at end of file diff --git a/src/onapsdk/sdnc/topology.py b/src/onapsdk/sdnc/topology.py new file mode 100644 index 0000000..f618c2c --- /dev/null +++ b/src/onapsdk/sdnc/topology.py @@ -0,0 +1,267 @@ +"""SDNC topology module. NETCONF-API.""" +# Copyright 2023 Deutsche Telekom AG +# +# 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. +from typing import List, Dict, Iterable + +from onapsdk.utils.headers_creator import headers_sdnc_creator +from onapsdk.utils.jinja import jinja_env + +from .sdnc_element import SdncElement + + +class Node(SdncElement): + """SDNC Node.""" + + headers: Dict[str, str] = headers_sdnc_creator(SdncElement.headers) + + def __init__(self, # pylint: disable=too-many-arguments + node_id: str, + host: str, + port: int, + username: str, + password: str, + topology_id: str = "topology-netconf", + **kwargs) -> None: + """Node information initialization. + + Args: + node_id (str): Node id, + host (str): Node IPv4 address, + port (int): Node Netconf port number, + username (str): Node username, + password (str): Node password, + topology_id: (str) : Topology, where node is contained + data (Dict): other possible Node data, + + """ + super().__init__() + self.node_id: str = node_id + self.host: str = host + self.port: int = port + self.username: str = username + self.password: str = password + self.topology_id: str = topology_id + self.data: Dict = kwargs + + def __repr__(self) -> str: + """Node information human-readable string. + + Returns: + str: Node information description + """ + return f"Node(node_id={self.node_id}," \ + f"host={self.host}," \ + f"port={self.port}," \ + f"username={self.username}," \ + f"password={self.password}," \ + f"data={self.data})" + + def create(self) -> None: + """Create the node element of the topology at SDNC via NETCONF-API. + + Returns: + None + """ + node_json_template = { + "node": { + "node-id": "", + "netconf-node-topology:host": "", + "netconf-node-topology:port": 0, + "netconf-node-topology:username": "", + "netconf-node-topology:password": "" + } + } + self.send_message( + "POST", + "Add a node element into the topology at SDNC using NETCONF-API", + (f"{self.base_url}/rests/data/" + f"network-topology:network-topology/topology={self.topology_id}"), + data=jinja_env().get_template( + "create_node_netconf_api.json.j2"). + render( + node_json_template, + node_id=self.node_id, + host_=self.host, + port_=self.port, + username_=self.username, + password_=self.password + ) + ) + + def update(self) -> None: + """Update the node element of the topology at SDNC via NETCONF-API. + + Returns: + None + + """ + node_json_template = { + "node": { + "node-id": "", + "netconf-node-topology:host": "", + "netconf-node-topology:port": 0, + "netconf-node-topology:username": "", + "netconf-node-topology:password": "" + } + } + self.send_message( + "PUT", + "Add a Node element into the topology using NETCONF-API", + (f"{self.base_url}/rests/data/" + f"network-topology:network-topology/topology={self.topology_id}" + f"/node={self.node_id}"), + data=jinja_env().get_template( + "create_node_netconf_api.json.j2"). + render( + node_json_template, + node_id=self.node_id, + host_=self.host, + port_=self.port, + username_=self.username, + password_=self.password) + ) + + def delete(self) -> None: + """Delete the node element of the topology from SDNC via NETCONF-API. + + Returns: + None + + """ + self.send_message( + "DELETE", + "Delete a Node element from the topology using NETCONF-API", + (f"{self.base_url}/rests/data/" + f"network-topology:network-topology/topology={self.topology_id}" + f"/node={self.node_id}") + ) + + +class Topology(SdncElement): + """SDNC topology.""" + + headers: Dict[str, str] = headers_sdnc_creator(SdncElement.headers) + + def __init__(self, + topology_id: str = "topology-netconf", + nodes: List[Node] = None): + """Topology information initialization. + + Args: + topology_id (str): Topology instance id + nodes (list): List of nodes inside the topology + """ + super().__init__() + self.topology_id: str = topology_id + self.nodes: list = nodes + + def __repr__(self) -> str: + """Topology information human-readable string. + + Returns: + str: Topology information description + + """ + return f"Topology(topology_id={self.topology_id}," \ + f"nodes={self.nodes})" + + @classmethod + def get_all(cls) -> Iterable["Topology"]: + """Get all topologies from SDNC using NETCONF-API. + + Yields: + : Topology object + """ + topologies = cls.send_message_json("GET", + "Get all topologies from SDNC using NETCONF-API", + f"{cls.base_url}" + f"/rests/data/network-topology:network-topology" + ).get("network-topology:network-topology", {} + ).get("topology", []) + for topology in topologies: + try: + yield Topology(topology_id=topology["topology-id"], nodes=topology["node"]) + except KeyError: + print(f"Topology with topology-id={topology['topology-id']}" + f" doesn't contain any node") + yield Topology(topology_id=topology["topology-id"]) + + @classmethod + def get(cls, topology_id) -> "Topology": + """Get the topology with a specific topology_id from SDNC via NETCONF-API. + + Returns: + Topology + + """ + topology_object = cls.send_message_json("GET", + "Get all topologies from SDNC using NETCONF-API", + f"{cls.base_url}" + f"/rests/data/network-topology:network-topology/" + f"topology={topology_id}" + ) + try: + topology = topology_object["network-topology:topology"][0] + return Topology(topology_id=topology["topology-id"], + nodes=topology["node"] + ) + except KeyError: + return Topology(topology_id=topology_id) + + def get_node(self, node_id) -> "Node": + """Get the node with a specific node_id form the specific topology at SDNC via NETCONF-API. + + Returns: + Node + + """ + node_object = self.send_message_json("GET", + "Get all nodes from SDNC using NETCONF-API", + f"{self.base_url}" + f"/rests/data/network-topology:network-topology/" + f"topology={self.topology_id}/node={node_id}") + try: + node = node_object["network-topology:node"][0] + return Node(node_id=node["node-id"], + host=node["netconf-node-topology:host"], + port=node["netconf-node-topology:port"], + username=node["netconf-node-topology:username"], + password=node["netconf-node-topology:password"], + topology_id=self.topology_id) + except KeyError: + print(f"Error. Node creation skipped.") + + def get_all_nodes(self) -> Iterable["Node"]: + """Get all nodes of the specific topology from SDNC using NETCONF-API. + + Yields: + : Node object + """ + nodes_object = self.send_message_json("GET", + "Get all nodes from SDNC using NETCONF-API", + f"{self.base_url}/rests/data" + f"/network-topology:network-topology/" + f"topology={self.topology_id}" + ) + nodes = nodes_object["network-topology:topology"][0]["node"] + for node in nodes: + try: + yield Node(node_id=node["node-id"], + host=node["netconf-node-topology:host"], + port=node["netconf-node-topology:port"], + username=node["netconf-node-topology:username"], + password=node["netconf-node-topology:password"], + topology_id=self.topology_id) + except KeyError: + print(f"Error. Node creation skipped. KeyError") diff --git a/tests/test_sdnc_node.py b/tests/test_sdnc_node.py new file mode 100644 index 0000000..1ab3474 --- /dev/null +++ b/tests/test_sdnc_node.py @@ -0,0 +1,102 @@ +"""Test SDNC node creation using NETCONF-API.""" +# Copyright 2023 Deutsche Telekom AG +# +# 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. +from unittest import mock + +from onapsdk.sdnc.topology import Node + +SDNC_NODE = { + "network-topology:node": [ + { + "node-id": "test-node", + "netconf-node-topology:host": "100.70.0.5", + "netconf-node-topology:port": 830, + "netconf-node-topology:keepalive-delay": 100, + "netconf-node-topology:tcp-only": False, + "netconf-node-topology:username": "root", + "netconf-node-topology:password": "password" + } + ] +} + +SDNC_TOPOLOGY_ID = "topology-netconf" + + +def test_sdnc_netconf_api_node_init(): + node = Node(node_id=SDNC_NODE["network-topology:node"][0]["node-id"], + host=SDNC_NODE["network-topology:node"][0]["netconf-node-topology:host"], + port=SDNC_NODE["network-topology:node"][0]["netconf-node-topology:port"], + username=SDNC_NODE["network-topology:node"][0]["netconf-node-topology:username"], + password=SDNC_NODE["network-topology:node"][0]["netconf-node-topology:password"], + ) + assert type(node.node_id) is str + + +@mock.patch.object(Node, "send_message") +def test_sdnc_netconf_api_node_create(mock_send_message): + node = Node(node_id="test-node-02", + host="100.70.0.102", + port=830, + username="admin", + password="2345", + topology_id="topology-netconf" + ) + node.create() + + mock_send_message.assert_called_once() + method, description, url = mock_send_message.call_args[0] + assert method == "POST" + assert description == "Add a node element into the topology at SDNC using NETCONF-API" + assert url == (f"{Node.base_url}/rests/data/" + f"network-topology:network-topology/topology={node.topology_id}") + + +@mock.patch.object(Node, "send_message") +def test_sdnc_netconf_api_node_delete(mock_send_message): + node = Node(node_id="test-node-02", + host="100.70.0.102", + port=830, + username="admin", + password="2345", + topology_id="topology-netconf" + ) + node.delete() + + mock_send_message.assert_called_once() + method, description, url = mock_send_message.call_args[0] + assert method == "DELETE" + assert description == "Delete a Node element from the topology using NETCONF-API" + assert url == (f"{Node.base_url}/rests/data/" + f"network-topology:network-topology/topology={node.topology_id}" + f"/node={node.node_id}") + + +@mock.patch.object(Node, "send_message") +def test_sdnc_netconf_api_node_update(mock_send_message): + node = Node(node_id="test-node-02", + host="100.70.0.102", + port=830, + username="admin", + password="2345", + topology_id="topology-netconf" + ) + node.update() + + mock_send_message.assert_called_once() + method, description, url = mock_send_message.call_args[0] + assert method == "PUT" + assert description == "Add a Node element into the topology using NETCONF-API" + assert url == (f"{Node.base_url}/rests/data/" + f"network-topology:network-topology/topology={node.topology_id}" + f"/node={node.node_id}") diff --git a/tests/test_sdnc_service.py b/tests/test_sdnc_service.py new file mode 100644 index 0000000..bc69797 --- /dev/null +++ b/tests/test_sdnc_service.py @@ -0,0 +1,188 @@ +"""Test SDNC service creation using GENERIC-RESOURCE-API.""" +# Copyright 2023 Deutsche Telekom AG +# +# 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. +from collections.abc import Iterable +from unittest import mock + +from onapsdk.sdnc.services import Service + +SDNC_SERVICES_INFORMATION = { + "services": { + "service": [ + { + "service-instance-id": "sdnc-int-test-fffffffffff", + "service-status": { + "response-code": "string", + "response-message": "string", + "final-indicator": "string", + "request-status": "string", + "action": "string", + "rpc-name": "string", + "rpc-action": "string", + "response-timestamp": "string" + }, + "service-data": { + "service-level-oper-status": { + "last-rpc-action": "assign", + "last-action": "CreateServiceInstance", + "order-status": "Created" + } + } + } + ] + } +} + +SDNC_SERVICES_INFORMATION_GET = { + "GENERIC-RESOURCE-API:service": [ + { + "service-instance-id": "sdnc-int-test-fffffffffff", + "service-status": { + "response-code": "string", + "response-message": "string", + "final-indicator": "string", + "request-status": "string", + "action": "string", + "rpc-name": "string", + "rpc-action": "string", + "response-timestamp": "string" + }, + "service-data": { + "service-level-oper-status": { + "last-rpc-action": "assign", + "last-action": "CreateServiceInstance", + "order-status": "Created" + } + } + } + ] +} + +SDNC_SERVICES_INFORMATION_GET_ALL_SERVICE_DATA_MISSING = { + "services": { + "service": [ + { + "service-instance-id": "sdnc-int-test-fffffffffff", + "service-status": { + "response-code": "string", + "response-message": "string", + "final-indicator": "string", + "request-status": "string", + "action": "string", + "rpc-name": "string", + "rpc-action": "string", + "response-timestamp": "string" + } + } + ] + } +} + +SDNC_SERVICES_INFORMATION_GET_ALL_SERVICE_STATUS_MISSING = { + "services": { + "service": [ + { + "service-instance-id": "sdnc-int-test-fffffffffff", + "service-data": { + "service-level-oper-status": { + "last-rpc-action": "assign", + "last-action": "CreateServiceInstance", + "order-status": "Created" + } + } + } + ] + } +} + +SDNC_SERVICE_ID = "sdnc-int-test-fffffffffff" + + +@mock.patch.object(Service, "send_message_json") +def test_sdnc_service_gr_api_get_all(mock_send_message_json): + mock_send_message_json.return_value = SDNC_SERVICES_INFORMATION + sdnc_all_services = Service.get_all() + assert isinstance(sdnc_all_services, Iterable) + sdnc_all_services_list = list(sdnc_all_services) + assert len(sdnc_all_services_list) == 1 + service = sdnc_all_services_list[0] + assert isinstance(service, Service) + assert service.service_instance_id == SDNC_SERVICE_ID + + +@mock.patch.object(Service, "send_message") +def test_sdnc_service_gr_api_create(mock_send_message): + service = Service(SDNC_SERVICES_INFORMATION["services"]["service"][0]["service-instance-id"], + SDNC_SERVICES_INFORMATION["services"]["service"][0]["service-status"]) + service.create() + + mock_send_message.assert_called_once() + method, description, url = mock_send_message.call_args[0] + assert method == "POST" + assert description == "Create a service using GENERIC-RESOURCES-API" + assert url == (f"{Service.base_url}/restconf/config/" + "GENERIC-RESOURCE-API:services") + + +@mock.patch.object(Service, "send_message") +def test_sdnc_service_gr_api_update(mock_send_message): + service = Service(service_instance_id=SDNC_SERVICES_INFORMATION["services"]["service"][0]["service-instance-id"], + service_status=SDNC_SERVICES_INFORMATION["services"]["service"][0]["service-status"], + service_data=SDNC_SERVICES_INFORMATION["services"]["service"][0]["service-data"]) + service.update() + mock_send_message.assert_called_once() + method, description, url = mock_send_message.call_args[0] + assert method == "PUT" + assert description == "Update service information by service-instance-id using GENERIC-RESOURCES-API" + assert url == (f"{Service.base_url}/rests/data/" + f"GENERIC-RESOURCE-API:services/service={service.service_instance_id}") + + +@mock.patch.object(Service, "send_message") +def test_sdnc_service_gr_api_delete(mock_send_message): + service = Service(SDNC_SERVICES_INFORMATION["services"]["service"][0]["service-instance-id"], + SDNC_SERVICES_INFORMATION["services"]["service"][0]["service-status"]) + service.delete() + + mock_send_message.assert_called_once() + method, description, url = mock_send_message.call_args[0] + assert method == "DELETE" + assert description == "Delete a service using GENERIC-RESOURCE-API" + assert url == (f"{Service.base_url}/rests/data/" + "GENERIC-RESOURCE-API:services/" + f"service={SDNC_SERVICE_ID}") + + +@mock.patch.object(Service, "send_message_json") +def test_sdnc_service_gr_api_get_all_key_error_data(mock_send_message_json): + mock_send_message_json.return_value = SDNC_SERVICES_INFORMATION_GET_ALL_SERVICE_DATA_MISSING + sdnc_all_services = Service.get_all() + assert isinstance(sdnc_all_services, Iterable) + sdnc_all_services_list = list(sdnc_all_services) + assert len(sdnc_all_services_list) == 1 + service = sdnc_all_services_list[0] + assert isinstance(service, Service) + assert service.service_data == {} + + +@mock.patch.object(Service, "send_message_json") +def test_sdnc_service_gr_api_get_all_key_error_status(mock_send_message_json): + mock_send_message_json.return_value = SDNC_SERVICES_INFORMATION_GET_ALL_SERVICE_STATUS_MISSING + sdnc_all_services = Service.get_all() + assert isinstance(sdnc_all_services, Iterable) + sdnc_all_services_list = list(sdnc_all_services) + assert len(sdnc_all_services_list) == 1 + service = sdnc_all_services_list[0] + assert isinstance(service, Service) + assert service.service_status == {} diff --git a/tests/test_sdnc_topology.py b/tests/test_sdnc_topology.py new file mode 100644 index 0000000..6ced081 --- /dev/null +++ b/tests/test_sdnc_topology.py @@ -0,0 +1,235 @@ +"""Test SDNC topology creation using NETCONF-API.""" +# Copyright 2023 Deutsche Telekom AG +# +# 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. +from collections.abc import Iterable +from unittest import mock + +from onapsdk.sdnc.topology import Topology, Node + +SDNC_TOPOLOGY_GET_NODE = { + "network-topology:node": [ + { + "node-id": "TEST", + "netconf-node-topology:sleep-factor": "1.5", + "netconf-node-topology:host": "10.1.1.41", + "netconf-node-topology:reconnect-on-changed-schema": False, + "netconf-node-topology:between-attempts-timeout-millis": 2000, + "netconf-node-topology:connection-status": "connected", + "netconf-node-topology:max-connection-attempts": 0, + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:connection-timeout-millis": 20000, + "netconf-node-topology:port": 32767, + "netconf-node-topology:tcp-only": False, + "netconf-node-topology:keepalive-delay": 120 + } + ] +} +SDNC_TOPOLOGY_GET_NODE_KEY_ERROR = { + "network-topology:node": [ + { + "node-id": "TEST", + "netconf-node-topology:host": "10.1.1.41", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin" + } + ] +} + +SDNC_TOPOLOGY_GET_ALL_NODES_KEY_ERROR = { + "network-topology:topology": [ + { + "topology-id": "topology-netconf", + "node": [ + { + "node-id": "TEST1", + "netconf-node-topology:host": "10.1.1.41", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + }, + { + "node-id": "TEST2", + "netconf-node-topology:port": 830, + "netconf-node-topology:username": "root", + "netconf-node-topology:password": "password", + "netconf-node-topology:host": "100.70.0.53" + } + ] + } + ] +} + +SDNC_TOPOLOGY_WITHOUT_NODES_GET_ALL = { + "network-topology:network-topology": { + "topology": [ + { + "topology-id": "topology-netconf", + } + ] + } +} + +SDNC_TOPOLOGY_WITHOUT_NODES_GET = { + "network-topology:topology": [ + { + "topology-id": "topology-netconf", + } + ] +} + +SDNC_TOPOLOGY_GET_ALL_NODES = { + "network-topology:topology": [ + { + "topology-id": "topology-netconf", + "node": [ + { + "node-id": "TEST", + "netconf-node-topology:host": "10.1.1.41", + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:port": 32767 + } + ] + } + ] +} +SDNC_TOPOLOGY = { + "network-topology:network-topology": { + "topology": [ + { + "topology-id": "topology-netconf", + "node": [ + { + "node-id": "MAAS", + "netconf-node-topology:sleep-factor": "1.5", + "netconf-node-topology:host": "10.32.32.32", + "netconf-node-topology:reconnect-on-changed-schema": False, + "netconf-node-topology:clustered-connection-status": { + "netconf-master-node": "test" + }, + "netconf-node-topology:between-attempts-timeout-millis": 2000, + "netconf-node-topology:connection-status": "connected", + "netconf-node-topology:max-connection-attempts": 0, + "netconf-node-topology:username": "admin", + "netconf-node-topology:password": "admin", + "netconf-node-topology:available-capabilities": { + "available-capability": [ + { + "capability": "urn:ietf:params:netconf:base:1.1", + "capability-origin": "device-advertised" + } + ] + }, + "netconf-node-topology:connection-timeout-millis": 10000, + "netconf-node-topology:port": 32767, + "netconf-node-topology:tcp-only": False, + "netconf-node-topology:keepalive-delay": 120 + } + ] + } + ] + } +} +SDNC_TOPOLOGY_ID = "topology-netconf" +SDNC_NODE = { + "network-topology:node": [ + { + "node-id": "test-node-fff", + "netconf-node-topology:host": "100.70.0.5", + "netconf-node-topology:port": 830, + "netconf-node-topology:keepalive-delay": 100, + "netconf-node-topology:tcp-only": False, + "netconf-node-topology:username": "root", + "netconf-node-topology:password": "password" + } + ] +} + +SDNC_NODE_ID = "test-node-fff" + + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get_all(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY + topology_iterable = Topology.get_all() + assert isinstance(topology_iterable, Iterable) + topology_list = list(topology_iterable) + assert len(topology_list) == 1 + topology = topology_list[0] + assert isinstance(topology, Topology) + assert topology.topology_id == SDNC_TOPOLOGY_ID + + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY_GET_ALL_NODES + topology = Topology() + topology_new = Topology.get(topology.topology_id) + assert isinstance(topology_new, Topology) + + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get_all_nodes(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY_GET_ALL_NODES + topology = Topology() + nodes = topology.get_all_nodes() + assert isinstance(nodes, Iterable) + node_list = list(nodes) + assert len(node_list) == 1 + + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get_node(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY_GET_NODE + topology = Topology() + node = topology.get_node("TEST") + assert isinstance(node, Node) + assert node.node_id == "TEST" + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get_all_key_error(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY_WITHOUT_NODES_GET_ALL + topology_iterable = Topology.get_all() + assert isinstance(topology_iterable, Iterable) + topology_list = list(topology_iterable) + assert len(topology_list) == 1 + topology = topology_list[0] + assert topology.nodes is None + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get_key_error(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY_WITHOUT_NODES_GET + topology = Topology.get(SDNC_TOPOLOGY_ID) + assert isinstance(topology, Topology) + assert topology.nodes is None + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get_all_nodes_key_error(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY_GET_ALL_NODES_KEY_ERROR + topology = Topology() + nodes = topology.get_all_nodes() + assert isinstance(nodes, Iterable) + try: + list(nodes) + except Exception as e: + assert e is KeyError + +@mock.patch.object(Topology, "send_message_json") +def test_sdnc_netconf_api_topology_get_node_key_error(mock_send_message_json): + mock_send_message_json.return_value = SDNC_TOPOLOGY_GET_NODE_KEY_ERROR + topology = Topology() + try: + topology.get_node("TEST") + except Exception as e: + assert e is KeyError -- cgit 1.2.3-korg