Unittest added

Author: symphony-hong

symphony/bdk/core/config/model/bdk_datafeed_config.py

@@ -15,10 +15,11 @@ def __init__(self, config):
 
         :param config: the dict containing the datafeed specific confguration.
         """
-        self.version = config.get(VERSION) if VERSION in config else DF_V1
+        self.version = DF_V1
         self.id_file_path = ""
-        if config is not None and DF_ID_FILE_PATH in config:
-            self.id_file_path = Path(config.get(DF_ID_FILE_PATH))
+        if config is not None:
+            self.id_file_path = Path(config.get(DF_ID_FILE_PATH)) if DF_ID_FILE_PATH in config else ""
+            self.version = config.get(VERSION)
 
     def get_id_file_path(self) -> Path:
         """

symphony/bdk/core/service/datafeed/datafeed_loop_v2.py

@@ -1,11 +1,12 @@
+from typing import Optional
+
 from symphony.bdk.core.auth.auth_session import AuthSession
 from symphony.bdk.core.config.model.bdk_config import BdkConfig
 from symphony.bdk.core.service.datafeed.abstract_datafeed_loop import AbstractDatafeedLoop
 from symphony.bdk.gen import ApiException
 from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
 from symphony.bdk.gen.agent_model.ack_id import AckId
 from symphony.bdk.gen.agent_model.v5_datafeed import V5Datafeed
-from typing import Optional
 
 
 class DatafeedLoopV2(AbstractDatafeedLoop):
@@ -30,27 +31,27 @@ class DatafeedLoopV2(AbstractDatafeedLoop):
     def __init__(self, datafeed_api: DatafeedApi, auth_session: AuthSession, config: BdkConfig):
         super().__init__(datafeed_api, auth_session, config)
         self._ack_id = ""
-        self.started = False
-        self.datafeed_id = None
+        self._started = False
+        self._datafeed_id = None
 
     async def start(self):
         datafeed = await self._retrieve_datafeed()
         if not datafeed:
             datafeed = await self._create_datafeed()
-        self.datafeed_id = datafeed.id
-        self.started = True
-        while self.started:
+        self._datafeed_id = datafeed.id
+        self._started = True
+        while self._started:
             try:
                 await self._read_datafeed()
             except ApiException as e:
                 if e.status == 400:
                     datafeed = await self._recreate_datafeed()
-                    self.datafeed_id = datafeed.id
+                    self._datafeed_id = datafeed.id
                 else:
                     raise e
 
     async def stop(self):
-        self.started = False
+        self._started = False
 
     async def _retrieve_datafeed(self) -> Optional[V5Datafeed]:
         session_token = await self.auth_session.session_token
@@ -70,7 +71,7 @@ async def _read_datafeed(self):
         params = {
             'session_token': await self.auth_session.session_token,
             'key_manager_token': await self.auth_session.key_manager_token,
-            'datafeed_id': self.datafeed_id,
+            'datafeed_id': self._datafeed_id,
             'ack_id': AckId(ack_id=self._ack_id)
         }
         event_list = await self.datafeed_api.read_datafeed(**params)
@@ -81,10 +82,11 @@ async def _read_datafeed(self):
     async def _delete_datafeed(self) -> None:
         session_token = await self.auth_session.session_token
         key_manager_token = await self.auth_session.key_manager_token
-        await self.datafeed_api.delete_datafeed(datafeed_id=self.datafeed_id,
+        await self.datafeed_api.delete_datafeed(datafeed_id=self._datafeed_id,
                                                 session_token=session_token,
                                                 key_manager_token=key_manager_token)
 
     async def _recreate_datafeed(self):
         await self._delete_datafeed()
+        self._ack_id = ""
         return await self._create_datafeed()

tests/core/config/models/test_bdk_datafeed_config.py

@@ -3,7 +3,7 @@
 import pytest
 
 
-@pytest.fixture(params=["v1", 25, True])
+@pytest.fixture(params=["v1"])
 def datafeed_version(request):
     return request.param
 

tests/core/service/datafeed/datafeed_loop_v2_test.py

@@ -0,0 +1,174 @@
+from unittest.mock import MagicMock, AsyncMock, call
+
+import pytest
+
+from symphony.bdk.core.auth.auth_session import AuthSession
+from symphony.bdk.core.config.loader import BdkConfigLoader
+from symphony.bdk.core.service.datafeed.abstract_datafeed_loop import RealTimeEvent
+from symphony.bdk.core.service.datafeed.datafeed_loop_v2 import DatafeedLoopV2
+from symphony.bdk.core.service.datafeed.real_time_event_listener import RealTimeEventListener
+from symphony.bdk.gen import ApiClient, ApiException
+from symphony.bdk.gen.agent_api.datafeed_api import DatafeedApi
+from symphony.bdk.gen.agent_model.ack_id import AckId
+from symphony.bdk.gen.agent_model.datafeed import Datafeed
+from symphony.bdk.gen.agent_model.v4_event import V4Event
+from symphony.bdk.gen.agent_model.v4_initiator import V4Initiator
+from symphony.bdk.gen.agent_model.v4_message_sent import V4MessageSent
+from symphony.bdk.gen.agent_model.v4_payload import V4Payload
+from symphony.bdk.gen.agent_model.v4_user import V4User
+from tests.utils.resource_utils import get_config_resource_filepath
+
+
+@pytest.fixture()
+def auth_session():
+    auth_session = AuthSession(None)
+    auth_session.session_token = "session_token"
+    auth_session.key_manager_token = "km_token"
+    return auth_session
+
+
+@pytest.fixture()
+def config():
+    config = BdkConfigLoader.load_from_file(get_config_resource_filepath("config.yaml"))
+    config.datafeed.version = "v2"
+    return config
+
+
+@pytest.fixture()
+def datafeed_api():
+    datafeed_api = MagicMock(DatafeedApi)
+    datafeed_api.api_client = MagicMock(ApiClient)
+    datafeed_api.list_datafeed = AsyncMock()
+    datafeed_api.create_datafeed = AsyncMock()
+    datafeed_api.read_datafeed = AsyncMock()
+    datafeed_api.delete_datafeed = AsyncMock()
+    return datafeed_api
+
+
+@pytest.fixture()
+def mock_listener():
+    return AsyncMock(wraps=RealTimeEventListener())
+
+
+@pytest.fixture()
+def initiator():
+    return V4Initiator(user=V4User(username="username"))
+
+
+@pytest.fixture()
+def message_sent_event(initiator):
+    class EventsMock:
+        def __init__(self, individual_event):
+            self.events = [individual_event]
+            self.ack_id = "ack_id"
+
+    v4_event = V4Event(type=RealTimeEvent.MESSAGESENT.name,
+                       payload=V4Payload(message_sent=V4MessageSent()),
+                       initiator=initiator)
+    event = EventsMock(v4_event)
+    return event
+
+
+@pytest.fixture()
+def datafeed_loop(datafeed_api, auth_session, config):
+    datafeed_loop = DatafeedLoopV2(datafeed_api, auth_session, config)
+
+    class RealTimeEventListenerImpl(RealTimeEventListener):
+
+        async def on_message_sent(self, initiator: V4Initiator, event: V4MessageSent):
+            await datafeed_loop.stop()
+
+    datafeed_loop.subscribe(RealTimeEventListenerImpl())
+    return datafeed_loop
+
+
+@pytest.mark.asyncio
+async def test_start(datafeed_loop, datafeed_api, message_sent_event):
+    datafeed_api.list_datafeed.return_value = []
+    datafeed_api.create_datafeed.return_value = Datafeed(id="test_id")
+    datafeed_api.read_datafeed.return_value = message_sent_event
+
+    await datafeed_loop.start()
+
+    datafeed_api.list_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token"
+    )
+    datafeed_api.create_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token"
+    )
+    datafeed_api.read_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token",
+        datafeed_id="test_id",
+        ack_id=AckId(ack_id="")
+    )
+
+    assert datafeed_loop._datafeed_id == "test_id"
+    assert datafeed_loop._ack_id == "ack_id"
+
+
+@pytest.mark.asyncio
+async def test_start_datafeed_exist(datafeed_loop, datafeed_api, message_sent_event):
+    datafeed_api.list_datafeed.return_value = [Datafeed(id="test_id_exist")]
+    datafeed_api.read_datafeed.return_value = message_sent_event
+
+    await datafeed_loop.start()
+
+    datafeed_api.list_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token"
+    )
+    datafeed_api.read_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token",
+        datafeed_id="test_id_exist",
+        ack_id=AckId(ack_id="")
+    )
+
+    assert datafeed_loop._datafeed_id == "test_id_exist"
+    assert datafeed_loop._ack_id == "ack_id"
+
+
+@pytest.mark.asyncio
+async def test_start_datafeed_stale_datafeed(datafeed_loop, datafeed_api, message_sent_event):
+    datafeed_api.list_datafeed.return_value = [Datafeed(id="fault_datafeed_id")]
+    datafeed_api.create_datafeed.return_value = Datafeed(id="test_id")
+    datafeed_api.read_datafeed.side_effect = [ApiException(400), message_sent_event]
+
+    await datafeed_loop.start()
+
+    datafeed_api.list_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token"
+    )
+
+    datafeed_api.delete_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token",
+        datafeed_id="fault_datafeed_id"
+    )
+
+    datafeed_api.create_datafeed.assert_called_with(
+        session_token="session_token",
+        key_manager_token="km_token"
+    )
+
+    datafeed_api.read_datafeed.assert_has_calls([
+        call(
+            session_token="session_token",
+            key_manager_token="km_token",
+            datafeed_id="fault_datafeed_id",
+            ack_id=AckId(ack_id="")
+        ),
+        call(
+            session_token="session_token",
+            key_manager_token="km_token",
+            datafeed_id="test_id",
+            ack_id=AckId(ack_id="")
+        )
+    ])
+
+    assert datafeed_loop._datafeed_id == "test_id"
+    assert datafeed_loop._ack_id == "ack_id"

tests/core/service_factory_test.py

@@ -0,0 +1,82 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from symphony.bdk.core.auth.auth_session import AuthSession
+from symphony.bdk.core.client.api_client_factory import ApiClientFactory
+from symphony.bdk.core.config.loader import BdkConfigLoader
+from symphony.bdk.core.service.connection.connection_service import ConnectionService
+from symphony.bdk.core.service.datafeed.datafeed_loop_v1 import DatafeedLoopV1
+from symphony.bdk.core.service.datafeed.datafeed_loop_v2 import DatafeedLoopV2
+from symphony.bdk.core.service.message.message_service import MessageService
+from symphony.bdk.core.service.stream.stream_service import StreamService
+from symphony.bdk.core.service.user.user_service import UserService
+from symphony.bdk.core.service_factory import ServiceFactory
+from symphony.bdk.gen import ApiClient
+from tests.utils.resource_utils import get_config_resource_filepath
+
+
+@pytest.fixture()
+def api_client_factory():
+    factory = MagicMock(ApiClientFactory)
+    api_client = MagicMock(ApiClient)
+    factory.get_pod_client.return_value = api_client
+    factory.get_agent_client.return_value = api_client
+    return factory
+
+
+@pytest.fixture()
+def config():
+    config = BdkConfigLoader.load_from_file(get_config_resource_filepath("config.yaml"))
+    return config
+
+
+@pytest.fixture()
+def service_factory(api_client_factory, config):
+    factory = ServiceFactory(api_client_factory, AuthSession(None), config)
+    return factory
+
+
+def test_get_user_service(service_factory):
+    user_service = service_factory.get_user_service()
+    assert user_service is not None
+    assert type(user_service) == UserService
+
+
+def test_get_message_service(service_factory):
+    message_service = service_factory.get_message_service()
+    assert message_service is not None
+    assert type(message_service) == MessageService
+
+
+def test_get_connection_service(service_factory):
+    connection_service = service_factory.get_connection_service()
+    assert connection_service is not None
+    assert type(connection_service) == ConnectionService
+
+
+def test_get_stream_service(service_factory):
+    stream_service = service_factory.get_stream_service()
+    assert stream_service is not None
+    assert type(stream_service) == StreamService
+
+
+def test_get_datafeed_loop(config, service_factory):
+    datafeed_loop = service_factory.get_datafeed_loop()
+    assert datafeed_loop is not None
+    assert type(datafeed_loop) == DatafeedLoopV1
+
+    config.datafeed.version = "v1"
+    datafeed_loop = service_factory.get_datafeed_loop()
+    assert datafeed_loop is not None
+    assert type(datafeed_loop) == DatafeedLoopV1
+
+    config.datafeed.version = "something"
+    datafeed_loop = service_factory.get_datafeed_loop()
+    assert datafeed_loop is not None
+    assert type(datafeed_loop) == DatafeedLoopV1
+
+    config.datafeed.version = "v2"
+    datafeed_loop = service_factory.get_datafeed_loop()
+    assert datafeed_loop is not None
+    assert type(datafeed_loop) == DatafeedLoopV2