Add Kafka envs

Signed-off-by: Bogumil Zebek <bogumil.zebek@nokia.com>
Issue-ID: INT-1869
Change-Id: Id8f4d3061fc16e8cb2c16bf32058ec8527e4c6c3

11 files changed, 114 insertions, 0 deletions

diff --git a/src/python/tests/snippet/README.md b/src/python/tests/snippet/README.md
new file mode 100644
index 0000000..a120182
--- /dev/null
+++ b/src/python/tests/snippet/README.md
@@ -0,0 +1,16 @@
+How to use
+----------
+    1. Make modifications in 'docker-compose.yml' in root directory
+
+    Replace 'KAFKA_ADVERTISED_HOST_NAME: kafka1' by 'KAFKA_ADVERTISED_HOST_NAME: localhost'
+
+    2. Build local netconf-server image in root directory
+
+        mvn clean package -Pdocker
+
+    3. Run docker-compose in root directory
+
+        docker-compose up -d
+
+    4. Run kafka_producer.py. Wait on results.
+    5. Run kafka_consumer.py. Wait on results.
diff --git a/src/python/tests/netconf_server/sysrepo_configuration/__init__.py b/src/python/tests/snippet/kafka_consumer.py
index eeb06d5..eef3e4d 100644
--- a/src/python/tests/netconf_server/sysrepo_configuration/__init__.py
+++ b/src/python/tests/snippet/kafka_consumer.py
@@ -17,3 +17,18 @@
 # limitations under the License.
 # ============LICENSE_END=========================================================
 ###
+import logging
+
+from netconf_server.netconf_kafka_client import NetconfKafkaClient
+
+logging.basicConfig(filename='kafka_consumer.log', level=logging.DEBUG)
+
+if __name__ == "__main__":
+
+    client = NetconfKafkaClient.create(
+        host="localhost",
+        port=9092
+    )
+
+    messages = client.get_all_messages_from(topic='config')
+    print("Received {}".format(messages))
diff --git a/src/python/tests/netconf_server/sysrepo_interface/__init__.py b/src/python/tests/snippet/kafka_producer.py
index eeb06d5..30a898c 100644
--- a/src/python/tests/netconf_server/sysrepo_interface/__init__.py
+++ b/src/python/tests/snippet/kafka_producer.py
@@ -17,3 +17,26 @@
 # limitations under the License.
 # ============LICENSE_END=========================================================
 ###
+import logging
+
+from netconf_server.netconf_kafka_client import NetconfKafkaClient
+
+NUMBER_OF_MESSAGES = 1000
+
+logging.basicConfig(filename='kafka_producer.log', level=logging.DEBUG)
+
+if __name__ == "__main__":
+
+    client = NetconfKafkaClient.create(
+        host="localhost",
+        port=9092
+    )
+
+    for number in range(NUMBER_OF_MESSAGES):
+        print("Send {}".format(number))
+        data = {'number': number}
+        resp = client.send(
+            topic='config',
+            value=data
+        )
+        print("Response: {}".format(resp.get(timeout=4)))
diff --git a/src/python/tests/netconf_server/__init__.py b/src/python/tests/unit/__init__.py
index eeb06d5..eeb06d5 100644
--- a/src/python/tests/netconf_server/__init__.py
+++ b/src/python/tests/unit/__init__.py
diff --git a/src/python/tests/unit/sysrepo_configuration/__init__.py b/src/python/tests/unit/sysrepo_configuration/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/src/python/tests/unit/sysrepo_configuration/__init__.py
diff --git a/src/python/tests/netconf_server/sysrepo_configuration/test_sysrepo_configuration_loader.py b/src/python/tests/unit/sysrepo_configuration/test_sysrepo_configuration_loader.py
index e5462e4..e5462e4 100644
--- a/src/python/tests/netconf_server/sysrepo_configuration/test_sysrepo_configuration_loader.py
+++ b/src/python/tests/unit/sysrepo_configuration/test_sysrepo_configuration_loader.py
diff --git a/src/python/tests/netconf_server/sysrepo_configuration/test_sysrepo_configuration_manager_.py b/src/python/tests/unit/sysrepo_configuration/test_sysrepo_configuration_manager_.py
index 5194218..5194218 100644
--- a/src/python/tests/netconf_server/sysrepo_configuration/test_sysrepo_configuration_manager_.py
+++ b/src/python/tests/unit/sysrepo_configuration/test_sysrepo_configuration_manager_.py
diff --git a/src/python/tests/unit/sysrepo_interface/__init__.py b/src/python/tests/unit/sysrepo_interface/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/src/python/tests/unit/sysrepo_interface/__init__.py
diff --git a/src/python/tests/netconf_server/sysrepo_interface/test_config_change_subscriber.py b/src/python/tests/unit/sysrepo_interface/test_config_change_subscriber.py
index 9817ba4..9817ba4 100644
--- a/src/python/tests/netconf_server/sysrepo_interface/test_config_change_subscriber.py
+++ b/src/python/tests/unit/sysrepo_interface/test_config_change_subscriber.py
diff --git a/src/python/tests/netconf_server/test_netconf_chang_listener.py b/src/python/tests/unit/test_netconf_chang_listener.py
index c2889f1..c2889f1 100644
--- a/src/python/tests/netconf_server/test_netconf_chang_listener.py
+++ b/src/python/tests/unit/test_netconf_chang_listener.py
diff --git a/src/python/tests/unit/test_netconf_kafka_client.py b/src/python/tests/unit/test_netconf_kafka_client.py
new file mode 100644
index 0000000..9eff761
--- /dev/null
+++ b/src/python/tests/unit/test_netconf_kafka_client.py
@@ -0,0 +1,60 @@
+###
+# ============LICENSE_START=======================================================
+# Netconf Server
+# ================================================================================
+# Copyright (C) 2021 Nokia. All rights reserved.
+# ================================================================================
+# 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.
+# ============LICENSE_END=========================================================
+###
+from unittest import TestCase
+from unittest.mock import MagicMock, call
+
+from netconf_server.netconf_kafka_client import NetconfKafkaClient
+
+MESSAGE_1 = "{'number':1}"
+MESSAGE_2 = "{'number':2}"
+
+TOPIC_NAME = 'config'
+
+
+class TestNetconfKafkaClient(TestCase):
+
+    def setUp(self):
+        self.producer = MagicMock()
+        self.kafka_customer_func = MagicMock(
+            return_value=[MagicMock(value=MESSAGE_1), MagicMock(value=MESSAGE_2)]
+        )
+        self.test_obj = NetconfKafkaClient(
+            producer=self.producer,
+            get_kafka_consumer_func=self.kafka_customer_func
+        )
+
+    def test_create_instance(self):
+        self.assertIsNotNone(self.test_obj)
+
+    def test_send_a_message_to_kafka(self):
+        # when
+        self.test_obj.send(TOPIC_NAME, MESSAGE_1)
+
+        # then
+        self.producer.assert_has_calls([call.send(topic=TOPIC_NAME, value=MESSAGE_1)])
+
+    def test_get_all_messages_from_kafka_topic(self):
+        # when
+        messages = self.test_obj.get_all_messages_from(TOPIC_NAME)
+
+        # then
+        self.assertTrue(len(messages) == 2)
+        self.assertTrue(MESSAGE_1 in messages)
+        self.assertTrue(MESSAGE_2 in messages)