CAIP-80 add a few e2e test scenarious

.github/workflows/e2e-build.yml

@@ -0,0 +1,55 @@
+name: e2e test
+on:
+  push:
+    branches: [ "main" ]
+
+jobs:
+  e2e-test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        id: setup-python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.12'
+
+      - name: Get pip cache dir
+        id: pip-cache
+        run: |
+          echo "dir=$(pip cache dir)" >> $GITHUB_OUTPUT
+      - name: Cache pip dependencies
+        uses: actions/cache@v4
+        with:
+          path: ${{ steps.pip-cache.outputs.dir }}
+          key: ${{ runner.os }}-pip-${{ steps.setup-python.outputs.python-version }}
+
+      - name: Install Poetry
+        run: pip install poetry
+
+      - name: Install Poetry plugins
+        run: poetry self add poetry-plugin-export
+
+      - name: Cache poetry virtualenv
+        uses: actions/cache@v4
+        with:
+          path: ~/.cache/pypoetry/virtualenvs
+          key: ${{ runner.os }}-poetry-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('poetry.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-poetry-${{ steps.setup-python.outputs.python-version }}-
+      - name: Install dependencies
+        run: poetry install
+
+      - name: Run E2E tests
+        env:
+            TEST_RSA_KEY: ${{ secrets.TEST_RSA_KEY }}
+            MSG_BOT_USERNAME: ${{ vars.MSG_BOT_USERNAME }}
+            FEED_BOT_USERNAME: ${{ vars.FEED_BOT_USERNAME }}
+            STREAM_ID: ${{ vars.STREAM_ID }}
+            SYMPHONY_HOST: ${{ vars.SYMPHONY_HOST }}
+            TEST_USER_ID: ${{ vars.TEST_SYM_USER_ID }}
+            BOT_USER_ID: ${{ vars.BOT_USER_ID }}
+        run: poetry run pytest -m e2e --no-cov
+        timeout-minutes: 10

pyproject.toml

@@ -49,6 +49,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.pytest.ini_options]
 addopts = "--junitxml=test-results/junit.xml --cov --cov-report=html"
 testpaths = ["tests"]
+markers = ["e2e: marks tests as end-to-end tests"]
 norecursedirs = ["*.egg", ".*", "build", "dist", "venv", "legacy"]
 junit_logging = "all"
 

tests/bdk/integration/e2e_test.py

@@ -0,0 +1,116 @@
+import asyncio
+from datetime import datetime, timedelta
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+
+from symphony.bdk.core.config.loader import BdkConfigLoader
+from symphony.bdk.core.symphony_bdk import SymphonyBdk
+from symphony.bdk.gen.pod_model.v3_room_attributes import V3RoomAttributes
+from tests.bdk.integration.helpers import (BOT_USER_ID, FEED_BOT_USERNAME,
+                                           MSG_BOT_USERNAME, STREAM_ID,
+                                           SYMPHONY_HOST, TEST_RSA_KEY,
+                                           TEST_USER_ID, MessageListener,
+                                           datafeed_bot_config,
+                                           get_test_messages,
+                                           messenger_bot_config, send_messages)
+
+pytestmark =[
+    pytest.mark.asyncio,
+    pytest.mark.e2e,
+ pytest.mark.skipif(
+    not all(
+        [
+            STREAM_ID,
+            MSG_BOT_USERNAME,
+            FEED_BOT_USERNAME,
+            SYMPHONY_HOST,
+            TEST_RSA_KEY,
+            TEST_USER_ID,
+            BOT_USER_ID,
+        ]
+    ),
+    reason="Required environment variables for integration tests are not set "
+    "(STREAM_ID, MSG_BOT_USERNAME, FEED_BOT_USERNAME, SYMPHONY_HOST, TEST_RSA_KEY, TEST_USER_ID, BOT_USER_ID)",
+)
+]
+NUMBER_OF_MESSAGES = 3
+
+
+@pytest_asyncio.fixture
+async def bdk(messenger_bot_config):
+    config = BdkConfigLoader.load_from_file(str(messenger_bot_config))
+    async with SymphonyBdk(config) as bdk:
+        yield bdk
+
+
+@pytest.mark.asyncio
+async def test_bot_read_write_messages(bdk):
+    uuid = str(uuid4())
+    # Given: test execution start time
+    since = int((datetime.now() - timedelta(seconds=2)).timestamp()) * 1000
+    # Given: BDK is initialized with config
+    # When: messages are sent via bot
+    await send_messages(bdk.messages(), STREAM_ID, since, uuid)
+    # Then: messages are readable with the same bot
+    messages = await get_test_messages(bdk, since, uuid)
+    # Then: Expected messages are posted to the room
+    assert sorted(messages) == [
+        f"{uuid}-{i}-{since}" for i in range(NUMBER_OF_MESSAGES)
+    ]
+
+
+@pytest.mark.asyncio
+async def test_bot_creates_stream_add_delete_user(bdk):
+    test_user = int(TEST_USER_ID)
+    # Given: Stream bdk creates a room
+    streams = bdk.streams()
+    room_result = await streams.create_room(
+        V3RoomAttributes(name="New fancy room", description="test room")
+    )
+    room_id = room_result.room_system_info.id
+    # When: user is added to the room
+    await streams.add_member_to_room(test_user, room_id)
+    members = await streams.list_room_members(room_id)
+    # Then: user is present in the room
+    assert test_user in [m.id for m in members.value]
+    # When: user is removed from the room
+    await streams.remove_member_from_room(test_user, room_id)
+    # Then: user is deleted from the room
+    members_after_removal = await streams.list_room_members(room_id)
+    assert test_user not in [m.id for m in members_after_removal.value]
+
+
+@pytest.mark.asyncio
+async def test_datafeed_receives_message(bdk: SymphonyBdk, datafeed_bot_config):
+    """
+    Test is running 2 bdk instances at the same time.
+    Data feed filters its own events so in order to see that feed is working
+    Two parale bots are added
+
+    """
+    # Given: message listener is initialized with expected message id
+    unique_id = str(uuid4())
+    message_content = f"Message for datafeed test. ID: {unique_id}"
+    listener = MessageListener(message_to_find=message_content)
+    # Given: members are added to the room
+    bdk.datafeed().subscribe(listener)
+    await bdk.streams().add_member_to_room(int(BOT_USER_ID), STREAM_ID)
+    datafeed_task = asyncio.create_task(bdk.datafeed().start())
+    await asyncio.sleep(3)
+    config = BdkConfigLoader.load_from_file(str(datafeed_bot_config))
+    async with SymphonyBdk(config) as another_bot:
+        # When: another bot instance sends a message to the needed room
+        await another_bot.messages().send_message(STREAM_ID, message_content)
+    try:
+        # Then: particular message is received by datafeed instance
+        await asyncio.wait_for(listener.message_received_event.wait(), timeout=300)
+    except asyncio.TimeoutError:
+        pytest.fail("Datafeed did not receive the message within the timeout period.")
+    finally:
+        await bdk.datafeed().stop()
+        await datafeed_task
+        bdk.datafeed().unsubscribe(listener)
+
+    assert listener.message_received_event.is_set()

tests/bdk/integration/helpers.py

@@ -0,0 +1,92 @@
+import asyncio
+import os
+import re
+
+import pytest
+import pytest_asyncio
+import yaml
+from pathlib import Path
+
+from symphony.bdk.core.config.loader import BdkConfigLoader
+from symphony.bdk.core.service.datafeed.real_time_event_listener import \
+    RealTimeEventListener
+from symphony.bdk.core.symphony_bdk import SymphonyBdk
+from symphony.bdk.gen.agent_model.v4_initiator import V4Initiator
+from symphony.bdk.gen.agent_model.v4_message_sent import V4MessageSent
+
+STREAM_ID = os.getenv("STREAM_ID")
+MSG_BOT_USERNAME = os.getenv("MSG_BOT_USERNAME")
+FEED_BOT_USERNAME = os.getenv("FEED_BOT_USERNAME")
+SYMPHONY_HOST = os.getenv("SYMPHONY_HOST")
+TEST_RSA_KEY = os.getenv("TEST_RSA_KEY")
+TEST_USER_ID = os.getenv("TEST_USER_ID")
+BOT_USER_ID = os.getenv("BOT_USER_ID")
+NUMBER_OF_MESSAGES = 3
+
+
+def generate_config(tmp_dir: Path, bot_username: str):
+    key_path = tmp_dir / "key.pem"
+    config_path = tmp_dir / "config.yaml"
+
+    bot_config_dict = {
+        "host": SYMPHONY_HOST,
+        "bot": {"username": bot_username, "privateKey": {"path": str(key_path)}},
+    }
+
+    key_path.write_text(TEST_RSA_KEY)
+    with config_path.open("w") as config_file:
+        yaml.dump(bot_config_dict, config_file)
+
+    return config_path
+
+
+@pytest.fixture
+def messenger_bot_config(tmp_path_factory):
+    tmp_dir = tmp_path_factory.mktemp("bdk_config_messenger")
+    return generate_config(tmp_dir, MSG_BOT_USERNAME)
+
+
+@pytest.fixture
+def datafeed_bot_config(tmp_path_factory):
+    tmp_dir = tmp_path_factory.mktemp("bdk_config_feed")
+    return generate_config(tmp_dir, FEED_BOT_USERNAME)
+
+
+@pytest_asyncio.fixture
+async def bdk(messenger_bot_config):
+    config = BdkConfigLoader.load_from_file(str(messenger_bot_config))
+    async with SymphonyBdk(config) as bdk:
+        yield bdk
+
+
+async def send_messages(messages, stream_id, since, uuid):
+    for i in range(NUMBER_OF_MESSAGES):
+        await messages.send_message(
+            stream_id, f"<messageML><b>{uuid}-{i}-{since}</b></messageML>"
+        )
+
+
+async def get_test_messages(bdk, since, uuid):
+    messages = await bdk.messages().list_messages(STREAM_ID, since=since)
+    cleaned_messages_text = [
+        re.sub(r"<[^>]+>", " ", msg["message"]).strip() for msg in messages
+    ]
+    return list(
+        filter(
+            lambda msg: msg.startswith(uuid),
+            cleaned_messages_text,
+        )
+    )
+
+
+class MessageListener(RealTimeEventListener):
+    """A simple listener to capture a specific message from the datafeed."""
+
+    def __init__(self, message_to_find: str):
+        self._message_to_find = message_to_find
+        self.message_received_event = asyncio.Event()
+
+    async def on_message_sent(self, initiator: V4Initiator, event: V4MessageSent):
+        message_text = re.sub(r"<[^>]+>", "", event.message.message).strip()
+        if self._message_to_find in message_text:
+            self.message_received_event.set()