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 +++++++++++++++++++++ 4 files changed, 456 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 (limited to 'src/onapsdk') 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") -- cgit 1.2.3-korg