botbuilder-python/libraries/botbuilder-integration-aiohttp/tests/skills/test_skill_http_client.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

from uuid import uuid4
from typing import Awaitable, Callable, Dict, Union


from unittest.mock import Mock
import aiounittest

from botbuilder.core import MessageFactory, InvokeResponse
from botbuilder.core.skills import (
    BotFrameworkSkill,
    ConversationIdFactoryBase,
    SkillConversationIdFactoryOptions,
    SkillConversationReference,
)
from botbuilder.integration.aiohttp.skills import SkillHttpClient
from botbuilder.schema import Activity, ConversationAccount, ConversationReference
from botframework.connector.auth import (
    AuthenticationConstants,
    ChannelProvider,
    GovernmentConstants,
)


class SimpleConversationIdFactory(ConversationIdFactoryBase):
    def __init__(self, conversation_id: str):
        self._conversation_id = conversation_id
        self._conversation_refs: Dict[str, SkillConversationReference] = {}
        # Public property to capture and assert the options passed to CreateSkillConversationIdAsync.
        self.creation_options: SkillConversationIdFactoryOptions = None

    async def create_skill_conversation_id(
        self,
        options_or_conversation_reference: Union[
            SkillConversationIdFactoryOptions, ConversationReference
        ],
    ) -> str:
        self.creation_options = options_or_conversation_reference

        key = self._conversation_id
        self._conversation_refs[key] = self._conversation_refs.get(
            key,
            SkillConversationReference(
                conversation_reference=options_or_conversation_reference.activity.get_conversation_reference(),
                oauth_scope=options_or_conversation_reference.from_bot_oauth_scope,
            ),
        )
        return key

    async def get_skill_conversation_reference(
        self, skill_conversation_id: str
    ) -> SkillConversationReference:
        return self._conversation_refs[skill_conversation_id]

    async def delete_conversation_reference(self, skill_conversation_id: str):
        raise NotImplementedError()


class TestSkillHttpClientTests(aiounittest.AsyncTestCase):
    async def test_post_activity_with_originating_audience(self):
        conversation_id = str(uuid4())
        conversation_id_factory = SimpleConversationIdFactory(conversation_id)
        test_activity = MessageFactory.text("some message")
        test_activity.conversation = ConversationAccount()
        skill = BotFrameworkSkill(
            id="SomeSkill",
            app_id="",
            skill_endpoint="https://someskill.com/api/messages",
        )

        async def _mock_post_content(
            to_url: str,
            token: str,  # pylint: disable=unused-argument
            activity: Activity,
        ) -> (int, object):
            nonlocal self
            self.assertEqual(skill.skill_endpoint, to_url)
            # Assert that the activity being sent has what we expect.
            self.assertEqual(conversation_id, activity.conversation.id)
            self.assertEqual("https://parentbot.com/api/messages", activity.service_url)

            # Create mock response.
            return 200, None

        sut = await self._create_http_client_with_mock_handler(
            _mock_post_content, conversation_id_factory
        )

        result = await sut.post_activity_to_skill(
            "",
            skill,
            "https://parentbot.com/api/messages",
            test_activity,
            "someOriginatingAudience",
        )

        # Assert factory options
        self.assertEqual("", conversation_id_factory.creation_options.from_bot_id)
        self.assertEqual(
            "someOriginatingAudience",
            conversation_id_factory.creation_options.from_bot_oauth_scope,
        )
        self.assertEqual(
            test_activity, conversation_id_factory.creation_options.activity
        )
        self.assertEqual(
            skill, conversation_id_factory.creation_options.bot_framework_skill
        )

        # Assert result
        self.assertIsInstance(result, InvokeResponse)
        self.assertEqual(200, result.status)

    async def test_post_activity_using_invoke_response(self):
        for is_gov in [True, False]:
            with self.subTest(is_government=is_gov):
                # pylint: disable=undefined-variable
                # pylint: disable=cell-var-from-loop
                conversation_id = str(uuid4())
                conversation_id_factory = SimpleConversationIdFactory(conversation_id)
                test_activity = MessageFactory.text("some message")
                test_activity.conversation = ConversationAccount()
                expected_oauth_scope = (
                    AuthenticationConstants.TO_CHANNEL_FROM_BOT_OAUTH_SCOPE
                )
                mock_channel_provider: ChannelProvider = Mock(spec=ChannelProvider)

                def is_government_mock():
                    nonlocal expected_oauth_scope
                    if is_government:
                        expected_oauth_scope = (
                            GovernmentConstants.TO_CHANNEL_FROM_BOT_OAUTH_SCOPE
                        )

                    return is_government

                mock_channel_provider.is_government = Mock(
                    side_effect=is_government_mock
                )

                skill = BotFrameworkSkill(
                    id="SomeSkill",
                    app_id="",
                    skill_endpoint="https://someskill.com/api/messages",
                )

                async def _mock_post_content(
                    to_url: str,
                    token: str,  # pylint: disable=unused-argument
                    activity: Activity,
                ) -> (int, object):
                    nonlocal self

                    self.assertEqual(skill.skill_endpoint, to_url)
                    # Assert that the activity being sent has what we expect.
                    self.assertEqual(conversation_id, activity.conversation.id)
                    self.assertEqual(
                        "https://parentbot.com/api/messages", activity.service_url
                    )

                    # Create mock response.
                    return 200, None

                sut = await self._create_http_client_with_mock_handler(
                    _mock_post_content, conversation_id_factory
                )
                result = await sut.post_activity_to_skill(
                    "", skill, "https://parentbot.com/api/messages", test_activity
                )

                # Assert factory options
                self.assertEqual(
                    "", conversation_id_factory.creation_options.from_bot_id
                )
                self.assertEqual(
                    expected_oauth_scope,
                    conversation_id_factory.creation_options.from_bot_oauth_scope,
                )
                self.assertEqual(
                    test_activity, conversation_id_factory.creation_options.activity
                )
                self.assertEqual(
                    skill, conversation_id_factory.creation_options.bot_framework_skill
                )

                # Assert result
                self.assertIsInstance(result, InvokeResponse)
                self.assertEqual(200, result.status)

    # Helper to create an HttpClient with a mock message handler that executes function argument to validate the request
    # and mock a response.
    async def _create_http_client_with_mock_handler(
        self,
        value_function: Callable[[object], Awaitable[object]],
        id_factory: ConversationIdFactoryBase,
        channel_provider: ChannelProvider = None,
    ) -> SkillHttpClient:
        # pylint: disable=protected-access
        client = SkillHttpClient(Mock(), id_factory, channel_provider)
        client._post_content = value_function

        return client