test cases to check the fallback feature

doajtest/unit/test_kafka_producer.py

@@ -0,0 +1,98 @@
+from redis import StrictRedis
+from unittest.mock import patch, Mock, MagicMock
+from doajtest.helpers import DoajTestCase
+from portality.events.system_status_check import KafkaStatusCheck
+import portality.events.kafka_producer as kafka_events
+class TestKafkaStatusCheck(DoajTestCase):
+
+    def setUp(self):
+        super(TestKafkaStatusCheck, self).setUp()
+        self.kafka_status = KafkaStatusCheck()
+
+    @patch.object(StrictRedis, "get", return_value=b'True')
+    @patch.object(KafkaStatusCheck, "can_check_redis", return_value=True)
+    def test_active_status(self, mock_kafka_status_check, mock_strict_redis):
+        status = self.kafka_status.is_active()
+        self.assertTrue(status)
+        mock_kafka_status_check.assert_called_once()
+        mock_strict_redis.assert_called_once()
+
+    @patch.object(StrictRedis, "get", return_value=b'False')
+    @patch.object(KafkaStatusCheck, "can_check_redis", return_value=True)
+    def test_inactive_status(self, mock_kafka_status_check, mock_strict_redis):
+        status = self.kafka_status.is_active()
+        self.assertFalse(status)
+        mock_kafka_status_check.assert_called_once()
+        mock_strict_redis.assert_called_once()
+
+class TestKafkaProducer(DoajTestCase):
+
+    # Test for handle_exception
+    @patch('portality.core.app.logger.exception')
+    @patch('portality.app_email.send_mail')
+    def test_handle_exception(self, mock_send_mail, mock_logger_exception):
+        error_msg = "Sample error"
+        exception = Exception("Sample exception")
+        kafka_events.handle_exception(error_msg, exception)
+
+        mock_logger_exception.assert_called_once_with(error_msg + str(exception))
+        mock_send_mail.assert_called_once()
+
+    # Test for kafka_producer when producer is None and no exception raised
+    @patch('kafka.KafkaProducer')
+    def test_kafka_producer_new(self, mock_kafka_producer):
+        kafka_events.producer = None
+        result = kafka_events.kafka_producer()
+
+        self.assertIsNotNone(result)
+        mock_kafka_producer.assert_called_once()
+
+    # Test for kafka_producer when producer is already set
+    def test_kafka_producer_existing(self):
+        kafka_events.producer = Mock()
+        result = kafka_events.kafka_producer()
+
+        self.assertEqual(result, kafka_events.producer)
+
+    # Test for kafka_producer when exception raised
+    @patch('kafka.KafkaProducer', side_effect=Exception("Kafka error"))
+    @patch('portality.events.kafka_producer.handle_exception')
+    def test_kafka_producer_exception(self, mock_handle_exception, _):
+        kafka_events.producer = None
+        result = kafka_events.kafka_producer()
+
+        self.assertIsNone(result)
+        mock_handle_exception.assert_called_once()
+
+    # Test for send_event when kafka_status is None
+    @patch('portality.events.kafka_producer.shortcircuit_send_event')
+    def test_send_event_status_none(self, mock_shortcircuit_send_event):
+        kafka_events.kafka_status = None
+
+        kafka_events.send_event(Mock())
+        mock_shortcircuit_send_event.assert_called_once()
+
+    # Test for send_event when everything is operational
+    @patch.object(KafkaStatusCheck, 'is_active', return_value=True)
+    @patch('portality.events.kafka_producer.kafka_producer', return_value=Mock())
+    @patch('portality.events.shortcircuit')
+    def test_send_event_operational(self, mock_shortcircuit, _, __):
+        kafka_events.kafka_status = KafkaStatusCheck()
+
+        kafka_events.send_event(Mock())
+        mock_shortcircuit.assert_not_called()
+
+    # Test for send_event when exception occurs
+    @patch('kafka.KafkaProducer',  return_value=Mock(send=MagicMock(side_effect=Exception("Send error"))))
+    @patch.object(KafkaStatusCheck, 'set_kafka_inactive_redis')
+    @patch.object(KafkaStatusCheck, 'is_active', return_value=True)
+    @patch('portality.events.kafka_producer.shortcircuit_send_event')
+    @patch('portality.events.kafka_producer.handle_exception')
+    @patch('portality.events.kafka_producer.producer', new=None)
+    def test_send_event_exception(self, __, mock_handle_exception, mock_shortcircuit, _, mock_kafka_producer):
+        kafka_events.kafka_status = KafkaStatusCheck()
+
+        kafka_events.send_event(Mock())
+        mock_handle_exception.assert_called()
+        mock_shortcircuit.assert_called_once()
+        mock_kafka_producer.assert_called_once()

portality/events/kafka_producer.py

@@ -1,5 +1,5 @@
 import json
-from kafka import KafkaProducer
+import kafka
 from portality.core import app
 from portality import app_email
 from portality.events.shortcircuit import send_event as shortcircuit_send_event
@@ -23,7 +23,7 @@ def kafka_producer():
     global producer
     try:
         if producer is None:
-            producer = KafkaProducer(bootstrap_servers=bootstrap_server, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
+            producer = kafka.KafkaProducer(bootstrap_servers=bootstrap_server, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
     except Exception as exp:
         producer = None
         handle_exception("Error in setting up kafka connection", exp)

portality/events/system_status_check.py

@@ -4,6 +4,10 @@
 from portality.lib import dates
 
 class KafkaStatusCheck:
+    """
+    useful to set Kafka status to false when there is any issue connecting to Kafka.
+    If kafka status set to false in Redis, it has to be set to true manually in Radis after fixing the issue.
+    """
 
     def __init__(self):
         self.is_kafka_active = True
@@ -14,11 +18,11 @@ def __init__(self):
         self.redis_conn = redis.StrictRedis(host=redis_host, port=redis_port, db=0)
 
     def is_active(self):
-        if self.can_check_in_redis():
+        if self.can_check_redis():
             self.is_kafka_active = self.is_kafka_active_redis()
         return self.is_kafka_active
 
-    def can_check_in_redis(self):
+    def can_check_redis(self):
         time_diff = dates.now() - self.last_time
         if time_diff.seconds > self.time_gap:
             self.last_time = dates.now()