Create SDNC Endpoints

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

7 files changed, 981 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")
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