Create SDNC Endpoints

Issue-ID: TEST-395
Signed-off-by: Aleksandr Taranov <aleksandr.taranov@telekom.com>
Change-Id: I688e2e96c9b6f3edee59105dcbd05f31f3ad1325

4 files changed, 456 insertions, 0 deletions

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")