tests: fix failing unit tests

app/config/di/event_producer_container.py

@@ -1,8 +1,8 @@
 from dependency_injector import containers, providers
-from app.adapters.broker.producers.sms_received_producer import SmsReceivedProducer
-from app.adapters.broker.producers.sms_submitted_producer import SmsSubmittedProducer
-from app.adapters.broker.producers.sms_sent_producer import SmsSentProducer
-from app.adapters.broker.producers.sms_callback_received_producer import SmsCallbackReceivedProducer
+from app.adapters.events.producers.sms_received_producer import SmsReceivedProducer
+from app.adapters.events.producers.sms_submitted_producer import SmsSubmittedProducer
+from app.adapters.events.producers.sms_sent_producer import SmsSentProducer
+from app.adapters.events.producers.sms_callback_received_producer import SmsCallbackReceivedProducer
 
 from app.settings import get_kafka_settings
 

app/workers/consumers/sms_callback_received/__main__.py

@@ -4,8 +4,8 @@
 from dependency_injector.wiring import inject, Provide
 from app.infra.logger import log as logger
 from app.domain.services.sms.create_sms import CreateSmsService
-from app.adapters.broker.consumers.sms_received_consumer import SmsReceivedConsumer
-from app.adapters.broker.producers.sms_submitted_producer import SmsSubmittedProducer
+from app.adapters.events.consumers.sms_received_consumer import SmsReceivedConsumer
+from app.adapters.events.producers.sms_submitted_producer import SmsSubmittedProducer
 from app.config.di.container import ApplicationContainer
 
 

app/workers/consumers/sms_received/__main__.py

@@ -4,8 +4,8 @@
 from dependency_injector.wiring import inject, Provide
 from app.infra.logger import log as logger
 from app.domain.services.sms.create_sms import CreateSmsService
-from app.adapters.broker.consumers.sms_received_consumer import SmsReceivedConsumer
-from app.adapters.broker.producers.sms_submitted_producer import SmsSubmittedProducer
+from app.adapters.events.consumers.sms_received_consumer import SmsReceivedConsumer
+from app.adapters.events.producers.sms_submitted_producer import SmsSubmittedProducer
 from app.config.di.container import ApplicationContainer
 
 

app/workers/consumers/sms_submitted/__main__.py

@@ -4,8 +4,8 @@
 from dependency_injector.wiring import inject, Provide
 from app.infra.logger import log as logger
 from app.domain.services.sms.send_sms import SendSmsService
-from app.adapters.broker.consumers.sms_submitted_consumer import SmsSubmittedConsumer
-from app.adapters.broker.producers.sms_sent_producer import SmsSentProducer
+from app.adapters.events.consumers.sms_submitted_consumer import SmsSubmittedConsumer
+from app.adapters.events.producers.sms_sent_producer import SmsSentProducer
 from app.config.di.container import ApplicationContainer
 
 

tests/integration/services/test_sms_received_consumer_integration.py → tests/integration/adapters/events/producers/test_sms_received_consumer_integration.py

@@ -2,9 +2,11 @@
 import pytest
 from faker import Faker
 import sanctumlabs.messageschema.events.notifications.sms.v1.events_pb2 as events
+from eventmsg_adaptor.event_streams import AsyncEventStream
+
 from app.domain.entities.sms import Sms
-from app.adapters.broker.producers.sms_received_producer import SmsReceivedProducer
-from app.adapters.broker.consumers.sms_received_consumer import SmsReceivedConsumer
+from app.adapters.events.producers.sms_received_producer import SmsReceivedProducer
+from app.adapters.events.consumers.sms_received_consumer import SmsReceivedConsumer
 from app.domain.entities.phone_number import PhoneNumber
 from app.domain.entities.message import Message
 from app.domain.entities.sms_status import SmsDeliveryStatus
@@ -18,7 +20,7 @@
 from app.infra.broker.kafka.consumers.proto_consumer import KafkaProtoConsumer
 from app.infra.broker.kafka.config import KafkaConsumerConfig
 
-from . import BaseKafkaIntegrationTestCase
+from tests.integration.adapters import BaseKafkaIntegrationTestCase
 
 fake = Faker()
 

tests/integration/services/test_sms_received_producer_integration.py → tests/integration/adapters/events/producers/test_sms_received_producer_integration.py

@@ -1,23 +1,19 @@
 import unittest
+from typing import cast
 import pytest
 from faker import Faker
-import sanctumlabs.messageschema.events.notifications.sms.v1.events_pb2 as events
+from eventmsg_adaptor.event_streams import AsyncEventStream
+from eventmsg_adaptor import factory
+from eventmsg_adaptor.config import Config, AdapterConfigs
+from eventmsg_adaptor.config.kafka import KafkaConfig, KafkaSchemaRegistryConfig
+
 from app.domain.entities.sms import Sms
-from app.adapters.broker.producers.sms_received_producer import SmsReceivedProducer
+from app.adapters.events.producers.sms_received_producer import SmsReceivedProducer
 from app.domain.entities.phone_number import PhoneNumber
 from app.domain.entities.message import Message
 from app.domain.entities.sms_status import SmsDeliveryStatus
-from app.infra.broker.kafka.serializers.protobuf_serializer import (
-    KafkaProtobufSerializer,
-)
-from app.infra.broker.kafka.deserializers.protobuf_deserializer import (
-    KafkaProtobufDeserializer,
-)
-from app.infra.broker.kafka.producers.proto_producer import KafkaProtoProducer
-from app.infra.broker.kafka.consumers.proto_consumer import KafkaProtoConsumer
-from app.infra.broker.kafka.config import KafkaConsumerConfig
 
-from . import BaseKafkaIntegrationTestCase
+from tests.integration.adapters import BaseKafkaIntegrationTestCase
 
 fake = Faker()
 
@@ -28,26 +24,25 @@ def setUp(self) -> None:
         super().setUp()
         self.topic = "test_topic"
 
-        self.serializer = KafkaProtobufSerializer(
-            msg_type=events.SmsReceived, registry_client=self.kafka_schema_registry
-        )
-        self.deserializer = KafkaProtobufDeserializer(msg_type=events.SmsReceived)
-
-        self.kafka_producer = KafkaProtoProducer(
-            params=self.producer_config, serializer=self.serializer
+        event_adapter_config = Config(
+            service_name="ujumbe-test",
+            default_adapter="kafka",
+            adapters=AdapterConfigs(
+                kafka=KafkaConfig(
+                    bootstrap_servers=[self.kafka_bootstrap_server],
+                    schema_registry=KafkaSchemaRegistryConfig(
+                        schema_registry_url=self.kafka_schema_registry,
+                        schema_registry_user_info=_kafka_settings.kafka_schema_registry_user_info
+                    )
+                )
+            )
         )
 
-        self.consumer_config = KafkaConsumerConfig(
-            bootstrap_servers=self.kafka_bootstrap_server,
-            topic=self.topic,
-            group_id="test-sms-received-group",
-        )
-        self.kafka_consumer = KafkaProtoConsumer(
-            params=self.consumer_config, deserializer=self.deserializer
-        )
+        self.event_producer_client = cast(AsyncEventStream,
+                                          factory(config=event_adapter_config, adapter_name="aiokafka"))
 
         self.sms_received_producer = SmsReceivedProducer(
-            topic=self.topic, event_stream=self.kafka_producer
+            topic=self.topic, event_stream=self.event_producer_client
         )
 
     @unittest.skip(

tests/unit/services/test_sms_received_consumer.py → tests/unit/adapters/events/consumers/test_sms_received_consumer.py

@@ -5,7 +5,7 @@
 import sanctumlabs.messageschema.events.notifications.sms.v1.events_pb2 as events
 import sanctumlabs.messageschema.events.notifications.sms.v1.data_pb2 as sms_data
 from app.domain.entities.sms import Sms
-from app.adapters.broker.consumers.sms_received_consumer import SmsReceivedConsumer
+from app.adapters.events.consumers.sms_received_consumer import SmsReceivedConsumer
 from app.infra.broker.kafka.consumers import KafkaConsumer
 from app.domain.entities.phone_number import PhoneNumber
 from app.domain.entities.message import Message

tests/unit/services/test_sms_submitted_consumer.py → tests/unit/adapters/events/consumers/test_sms_submitted_consumer.py

@@ -5,7 +5,7 @@
 import sanctumlabs.messageschema.events.notifications.sms.v1.events_pb2 as events
 import sanctumlabs.messageschema.events.notifications.sms.v1.data_pb2 as sms_data
 from app.domain.entities.sms import Sms
-from app.adapters.broker.consumers.sms_submitted_consumer import SmsSubmittedConsumer
+from app.adapters.events.consumers.sms_submitted_consumer import SmsSubmittedConsumer
 from app.infra.broker.kafka.consumers import KafkaConsumer
 from app.domain.entities.phone_number import PhoneNumber
 from app.domain.entities.message import Message

tests/unit/services/test_sms_received_producer.py → tests/unit/adapters/events/producers/test_sms_received_producer.py

@@ -2,9 +2,9 @@
 from unittest.mock import Mock
 import pytest
 from faker import Faker
+from eventmsg_adaptor.event_streams import AsyncEventStream
 from app.domain.entities.sms import Sms
-from app.adapters.broker.producers.sms_received_producer import SmsReceivedProducer
-from app.infra.broker.kafka.producers import KafkaProducer
+from app.adapters.events.producers.sms_received_producer import SmsReceivedProducer
 from app.domain.entities.phone_number import PhoneNumber
 from app.domain.entities.message import Message
 from app.domain.entities.sms_status import SmsDeliveryStatus
@@ -16,13 +16,13 @@
 class SmsReceivedProducerTestCases(unittest.TestCase):
     def setUp(self) -> None:
         self.topic = "test_topic"
-        self.mock_kafka_producer = Mock(spec=KafkaProducer)
+        self.mock_event_producer_client = Mock(spec=AsyncEventStream)
         self.sms_received_producer = SmsReceivedProducer(
-            topic=self.topic, event_stream=self.mock_kafka_producer
+            topic=self.topic, event_stream=self.mock_event_producer_client
         )
 
-    def test_throws_exception_when_there_is_an_error_producing_message(self):
-        """Should throw exception if kafka client fails to produce message"""
+    async def test_throws_exception_when_there_is_an_error_producing_message(self):
+        """Should throw exception if the event producer client fails to produce a message"""
         sender_phone = "+254700000000"
         sender = PhoneNumber(value=sender_phone)
         recipient_phone = "+254700000000"
@@ -37,21 +37,21 @@ def test_throws_exception_when_there_is_an_error_producing_message(self):
             status=SmsDeliveryStatus.PENDING,
         )
 
-        self.mock_kafka_producer.produce.side_effect = Exception(
+        self.mock_event_producer_client.publish.side_effect = Exception(
             "Failed to publish message"
         )
 
         with self.assertRaises(Exception):
-            self.sms_received_producer.publish_message(mock_sms)
+            await self.sms_received_producer.publish_message(mock_sms)
 
-        self.mock_kafka_producer.produce.assert_called()
+        self.mock_event_producer_client.produce.assert_called()
 
         # there are 3 attempts to call the produce method before giving up
-        call_count = self.mock_kafka_producer.produce.call_count
+        call_count = self.mock_event_producer_client.produce.call_count
         self.assertEquals(3, call_count)
 
-    def test_successfully_publishes_message(self):
-        """Should successfully produce message with Kafka client. No exception is thrown"""
+    async def test_successfully_publishes_message(self):
+        """Should successfully produce a message with event producer client. No exception is thrown"""
         sender_phone = "+254700000000"
         sender = PhoneNumber(value=sender_phone)
         recipient_phone = "+254700000000"
@@ -66,10 +66,10 @@ def test_successfully_publishes_message(self):
             status=SmsDeliveryStatus.PENDING,
         )
 
-        self.sms_received_producer.publish_message(mock_sms)
+        await self.sms_received_producer.publish_message(mock_sms)
 
-        self.mock_kafka_producer.produce.assert_called()
-        self.mock_kafka_producer.produce.assert_called_once()
+        self.mock_event_producer_client.publish.assert_called()
+        self.mock_event_producer_client.publish.assert_called_once()
 
 
 if __name__ == "__main__":

tests/unit/services/test_sms_sent_producer.py → tests/unit/adapters/events/producers/test_sms_sent_producer.py

@@ -2,9 +2,9 @@
 from unittest.mock import Mock
 import pytest
 from faker import Faker
+from eventmsg_adaptor.event_streams import AsyncEventStream
 from app.domain.entities.sms import Sms
-from app.adapters.broker.producers.sms_sent_producer import SmsSentProducer
-from app.infra.broker.kafka.producers import KafkaProducer
+from app.adapters.events.producers.sms_sent_producer import SmsSentProducer
 from app.domain.entities.phone_number import PhoneNumber
 from app.domain.entities.message import Message
 from app.domain.entities.sms_status import SmsDeliveryStatus
@@ -16,13 +16,13 @@
 class SmsSentProducerTestCases(unittest.TestCase):
     def setUp(self) -> None:
         self.topic = "test_topic"
-        self.mock_kafka_producer = Mock(spec=KafkaProducer)
+        self.mock_event_producer_client = Mock(spec=AsyncEventStream)
         self.sms_sent_producer = SmsSentProducer(
-            topic=self.topic, async_event_stream=self.mock_kafka_producer
+            topic=self.topic, async_event_stream=self.mock_event_producer_client
         )
 
-    def test_throws_exception_when_there_is_an_error_producing_message(self):
-        """Should throw exception if kafka client fails to produce message"""
+    async def test_throws_exception_when_there_is_an_error_producing_message(self):
+        """Should throw exception if the event producer client fails to publish a message"""
         sender_phone = "+254700000000"
         sender = PhoneNumber(value=sender_phone)
         recipient_phone = "+254700000000"
@@ -37,18 +37,18 @@ def test_throws_exception_when_there_is_an_error_producing_message(self):
             status=SmsDeliveryStatus.PENDING,
         )
 
-        self.mock_kafka_producer.produce.side_effect = Exception(
+        self.mock_event_producer_client.produce.side_effect = Exception(
             "Failed to publish message"
         )
 
         with self.assertRaises(Exception):
-            self.sms_sent_producer.publish_message(mock_sms)
+            await self.sms_sent_producer.publish_message(mock_sms)
 
-        self.mock_kafka_producer.produce.assert_called()
-        self.mock_kafka_producer.produce.assert_called_once()
+        self.mock_event_producer_client.publish.assert_called()
+        self.mock_event_producer_client.publish.assert_called_once()
 
-    def test_successfully_publishes_message(self):
-        """Should successfully produce message with Kafka client. No exception is thrown"""
+    async def test_successfully_publishes_message(self):
+        """Should successfully produce message using the event producer client. No exception is thrown"""
         sender_phone = "+254700000000"
         sender = PhoneNumber(value=sender_phone)
         recipient_phone = "+254700000000"
@@ -63,10 +63,10 @@ def test_successfully_publishes_message(self):
             status=SmsDeliveryStatus.PENDING,
         )
 
-        self.sms_sent_producer.publish_message(mock_sms)
+        await self.sms_sent_producer.publish_message(mock_sms)
 
-        self.mock_kafka_producer.produce.assert_called()
-        self.mock_kafka_producer.produce.assert_called_once()
+        self.mock_event_producer_client.produce.assert_called()
+        self.mock_event_producer_client.produce.assert_called_once()
 
 
 if __name__ == "__main__":

tests/unit/services/test_sms_submitted_producer.py → tests/unit/adapters/events/producers/test_sms_submitted_producer.py

@@ -2,9 +2,9 @@
 from unittest.mock import Mock
 import pytest
 from faker import Faker
+from eventmsg_adaptor.event_streams import AsyncEventStream
 from app.domain.entities.sms import Sms
-from app.adapters.broker.producers.sms_submitted_producer import SmsSubmittedProducer
-from app.infra.broker.kafka.producers import KafkaProducer
+from app.adapters.events.producers.sms_submitted_producer import SmsSubmittedProducer
 from app.domain.entities.phone_number import PhoneNumber
 from app.domain.entities.message import Message
 from app.domain.entities.sms_status import SmsDeliveryStatus
@@ -16,13 +16,13 @@
 class SmsSubmittedProducerTestCases(unittest.TestCase):
     def setUp(self) -> None:
         self.topic = "test_topic"
-        self.mock_kafka_producer = Mock(spec=KafkaProducer)
+        self.mock_event_producer_client = Mock(spec=AsyncEventStream)
         self.sms_submitted_producer = SmsSubmittedProducer(
-            topic=self.topic, async_event_stream=self.mock_kafka_producer
+            topic=self.topic, async_event_stream=self.mock_event_producer_client
         )
 
-    def test_throws_exception_when_there_is_an_error_producing_message(self):
-        """Should throw exception if kafka client fails to produce message"""
+    async def test_throws_exception_when_there_is_an_error_producing_message(self):
+        """Should throw exception if the event producer client fails to publish a message"""
         sender_phone = "+254700000000"
         sender = PhoneNumber(value=sender_phone)
         recipient_phone = "+254700000000"
@@ -37,18 +37,18 @@ def test_throws_exception_when_there_is_an_error_producing_message(self):
             status=SmsDeliveryStatus.PENDING,
         )
 
-        self.mock_kafka_producer.produce.side_effect = Exception(
+        self.mock_event_producer_client.produce.side_effect = Exception(
             "Failed to publish message"
         )
 
         with self.assertRaises(Exception):
-            self.sms_submitted_producer.publish_message(mock_sms)
+            await self.sms_submitted_producer.publish_message(mock_sms)
 
-        self.mock_kafka_producer.produce.assert_called()
-        self.mock_kafka_producer.produce.assert_called_once()
+        self.mock_event_producer_client.produce.assert_called()
+        self.mock_event_producer_client.produce.assert_called_once()
 
-    def test_successfully_publishes_message(self):
-        """Should successfully produce message with Kafka client. No exception is thrown"""
+    async def test_successfully_publishes_message(self):
+        """Should successfully produce a message with the event producer client. No exception is thrown"""
         sender_phone = "+254700000000"
         sender = PhoneNumber(value=sender_phone)
         recipient_phone = "+254700000000"
@@ -63,10 +63,10 @@ def test_successfully_publishes_message(self):
             status=SmsDeliveryStatus.PENDING,
         )
 
-        self.sms_submitted_producer.publish_message(mock_sms)
+        await self.sms_submitted_producer.publish_message(mock_sms)
 
-        self.mock_kafka_producer.produce.assert_called()
-        self.mock_kafka_producer.produce.assert_called_once()
+        self.mock_event_producer_client.produce.assert_called()
+        self.mock_event_producer_client.produce.assert_called_once()
 
 
 if __name__ == "__main__":

tests/unit/workers/consumers/test_sms_received_consumer.py

@@ -10,7 +10,7 @@
 from app.domain.entities.sms_status import SmsDeliveryStatus
 from app.domain.entities.sms import Sms
 from app.domain.services.sms.create_sms import CreateSmsService
-from app.adapters.broker.consumers.sms_received_consumer import SmsReceivedConsumer
+from app.adapters.events.consumers.sms_received_consumer import SmsReceivedConsumer
 from app.workers.consumers.sms_received.__main__ import main
 
 fake = Faker()