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/topology.py | 267 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 267 insertions(+) create mode 100644 src/onapsdk/sdnc/topology.py (limited to 'src/onapsdk/sdnc/topology.py') 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") -- cgit 1.2.3-korg