botbuilder-python/libraries/botbuilder-ai/tests/qna/test_qna.py

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

# pylint: disable=protected-access
# pylint: disable=too-many-lines

import unittest
from os import path
from typing import List, Dict
from unittest.mock import patch

import json
import requests
from aiohttp import ClientSession

import aiounittest
from botbuilder.ai.qna import QnAMakerEndpoint, QnAMaker, QnAMakerOptions
from botbuilder.ai.qna.models import (
    FeedbackRecord,
    JoinOperator,
    Metadata,
    QueryResult,
    QnARequestContext,
)
from botbuilder.ai.qna.utils import HttpRequestUtils, QnATelemetryConstants
from botbuilder.ai.qna.models import GenerateAnswerRequestBody
from botbuilder.core import BotAdapter, BotTelemetryClient, TurnContext
from botbuilder.core.adapters import TestAdapter
from botbuilder.schema import (
    Activity,
    ActivityTypes,
    ChannelAccount,
    ConversationAccount,
)


class TestContext(TurnContext):
    __test__ = False

    def __init__(self, request):
        super().__init__(TestAdapter(), request)
        self.sent: List[Activity] = list()

        self.on_send_activities(self.capture_sent_activities)

    async def capture_sent_activities(
        self, context: TurnContext, activities, next
    ):  # pylint: disable=unused-argument
        self.sent += activities
        context.responded = True


class QnaApplicationTest(aiounittest.AsyncTestCase):
    # Note this is NOT a real QnA Maker application ID nor a real QnA Maker subscription-key
    # theses are GUIDs edited to look right to the parsing and validation code.

    _knowledge_base_id: str = "f028d9k3-7g9z-11d3-d300-2b8x98227q8w"
    _endpoint_key: str = "1k997n7w-207z-36p3-j2u1-09tas20ci6011"
    _host: str = "https://dummyqnahost.azurewebsites.net/qnamaker"

    tests_endpoint = QnAMakerEndpoint(_knowledge_base_id, _endpoint_key, _host)

    def test_qnamaker_construction(self):
        # Arrange
        endpoint = self.tests_endpoint

        # Act
        qna = QnAMaker(endpoint)
        endpoint = qna._endpoint

        # Assert
        self.assertEqual(
            "f028d9k3-7g9z-11d3-d300-2b8x98227q8w", endpoint.knowledge_base_id
        )
        self.assertEqual("1k997n7w-207z-36p3-j2u1-09tas20ci6011", endpoint.endpoint_key)
        self.assertEqual(
            "https://dummyqnahost.azurewebsites.net/qnamaker", endpoint.host
        )

    def test_endpoint_with_empty_kbid(self):
        empty_kbid = ""

        with self.assertRaises(TypeError):
            QnAMakerEndpoint(empty_kbid, self._endpoint_key, self._host)

    def test_endpoint_with_empty_endpoint_key(self):
        empty_endpoint_key = ""

        with self.assertRaises(TypeError):
            QnAMakerEndpoint(self._knowledge_base_id, empty_endpoint_key, self._host)

    def test_endpoint_with_emptyhost(self):
        with self.assertRaises(TypeError):
            QnAMakerEndpoint(self._knowledge_base_id, self._endpoint_key, "")

    def test_qnamaker_with_none_endpoint(self):
        with self.assertRaises(TypeError):
            QnAMaker(None)

    def test_set_default_options_with_no_options_arg(self):
        qna_without_options = QnAMaker(self.tests_endpoint)
        options = qna_without_options._generate_answer_helper.options

        default_threshold = 0.3
        default_top = 1
        default_strict_filters = []

        self.assertEqual(default_threshold, options.score_threshold)
        self.assertEqual(default_top, options.top)
        self.assertEqual(default_strict_filters, options.strict_filters)

    def test_options_passed_to_ctor(self):
        options = QnAMakerOptions(
            score_threshold=0.8,
            timeout=9000,
            top=5,
            strict_filters=[Metadata(**{"movie": "disney"})],
        )

        qna_with_options = QnAMaker(self.tests_endpoint, options)
        actual_options = qna_with_options._generate_answer_helper.options

        expected_threshold = 0.8
        expected_timeout = 9000
        expected_top = 5
        expected_strict_filters = [Metadata(**{"movie": "disney"})]

        self.assertEqual(expected_threshold, actual_options.score_threshold)
        self.assertEqual(expected_timeout, actual_options.timeout)
        self.assertEqual(expected_top, actual_options.top)
        self.assertEqual(
            expected_strict_filters[0].name, actual_options.strict_filters[0].name
        )
        self.assertEqual(
            expected_strict_filters[0].value, actual_options.strict_filters[0].value
        )

    async def test_returns_answer(self):
        # Arrange
        question: str = "how do I clean the stove?"
        response_path: str = "ReturnsAnswer.json"

        # Act
        result = await QnaApplicationTest._get_service_result(question, response_path)

        first_answer = result[0]

        # Assert
        self.assertIsNotNone(result)
        self.assertEqual(1, len(result))
        self.assertEqual(
            "BaseCamp: You can use a damp rag to clean around the Power Pack",
            first_answer.answer,
        )

    async def test_active_learning_enabled_status(self):
        # Arrange
        question: str = "how do I clean the stove?"
        response_path: str = "ReturnsAnswer.json"

        # Act
        result = await QnaApplicationTest._get_service_result_raw(
            question, response_path
        )

        # Assert
        self.assertIsNotNone(result)
        self.assertEqual(1, len(result.answers))
        self.assertFalse(result.active_learning_enabled)

    async def test_returns_answer_with_strict_filters_with_or_operator(self):
        # Arrange
        question: str = "Where can you find"
        response_path: str = "RetrunsAnswer_WithStrictFilter_Or_Operator.json"
        response_json = QnaApplicationTest._get_json_for_file(response_path)

        strict_filters = [
            Metadata(name="species", value="human"),
            Metadata(name="type", value="water"),
        ]
        options = QnAMakerOptions(
            top=5,
            strict_filters=strict_filters,
            strict_filters_join_operator=JoinOperator.OR,
        )
        qna = QnAMaker(endpoint=QnaApplicationTest.tests_endpoint)
        context = QnaApplicationTest._get_context(question, TestAdapter())

        # Act
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ) as mock_http_client:
            result = await qna.get_answers_raw(context, options)

            serialized_http_req_args = mock_http_client.call_args[1]["data"]
            req_args = json.loads(serialized_http_req_args)

            # Assert
            self.assertIsNotNone(result)
            self.assertEqual(3, len(result.answers))
            self.assertEqual(
                JoinOperator.OR, req_args["strictFiltersCompoundOperationType"]
            )

            req_args_strict_filters = req_args["strictFilters"]

            first_filter = strict_filters[0]
            self.assertEqual(first_filter.name, req_args_strict_filters[0]["name"])
            self.assertEqual(first_filter.value, req_args_strict_filters[0]["value"])

            second_filter = strict_filters[1]
            self.assertEqual(second_filter.name, req_args_strict_filters[1]["name"])
            self.assertEqual(second_filter.value, req_args_strict_filters[1]["value"])

    async def test_returns_answer_with_strict_filters_with_and_operator(self):
        # Arrange
        question: str = "Where can you find"
        response_path: str = "RetrunsAnswer_WithStrictFilter_And_Operator.json"
        response_json = QnaApplicationTest._get_json_for_file(response_path)

        strict_filters = [
            Metadata(name="species", value="human"),
            Metadata(name="type", value="water"),
        ]
        options = QnAMakerOptions(
            top=5,
            strict_filters=strict_filters,
            strict_filters_join_operator=JoinOperator.AND,
        )
        qna = QnAMaker(endpoint=QnaApplicationTest.tests_endpoint)
        context = QnaApplicationTest._get_context(question, TestAdapter())

        # Act
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ) as mock_http_client:
            result = await qna.get_answers_raw(context, options)

            serialized_http_req_args = mock_http_client.call_args[1]["data"]
            req_args = json.loads(serialized_http_req_args)

            # Assert
            self.assertIsNotNone(result)
            self.assertEqual(1, len(result.answers))
            self.assertEqual(
                JoinOperator.AND, req_args["strictFiltersCompoundOperationType"]
            )

            req_args_strict_filters = req_args["strictFilters"]

            first_filter = strict_filters[0]
            self.assertEqual(first_filter.name, req_args_strict_filters[0]["name"])
            self.assertEqual(first_filter.value, req_args_strict_filters[0]["value"])

            second_filter = strict_filters[1]
            self.assertEqual(second_filter.name, req_args_strict_filters[1]["name"])
            self.assertEqual(second_filter.value, req_args_strict_filters[1]["value"])

    async def test_returns_answer_using_requests_module(self):
        question: str = "how do I clean the stove?"
        response_path: str = "ReturnsAnswer.json"
        response_json = QnaApplicationTest._get_json_for_file(response_path)

        qna = QnAMaker(endpoint=QnaApplicationTest.tests_endpoint, http_client=requests)
        context = QnaApplicationTest._get_context(question, TestAdapter())

        with patch("requests.post", return_value=response_json):
            result = await qna.get_answers_raw(context)
            answers = result.answers

            self.assertIsNotNone(result)
            self.assertEqual(1, len(answers))
            self.assertEqual(
                "BaseCamp: You can use a damp rag to clean around the Power Pack",
                answers[0].answer,
            )

    async def test_returns_answer_using_options(self):
        # Arrange
        question: str = "up"
        response_path: str = "AnswerWithOptions.json"
        options = QnAMakerOptions(
            score_threshold=0.8, top=5, strict_filters=[Metadata(**{"movie": "disney"})]
        )

        # Act
        result = await QnaApplicationTest._get_service_result(
            question, response_path, options=options
        )

        first_answer = result[0]
        has_at_least_1_ans = True
        first_metadata = first_answer.metadata[0]

        # Assert
        self.assertIsNotNone(result)
        self.assertEqual(has_at_least_1_ans, len(result) >= 1)
        self.assertTrue(first_answer.answer[0])
        self.assertEqual("is a movie", first_answer.answer)
        self.assertTrue(first_answer.score >= options.score_threshold)
        self.assertEqual("movie", first_metadata.name)
        self.assertEqual("disney", first_metadata.value)

    async def test_trace_test(self):
        activity = Activity(
            type=ActivityTypes.message,
            text="how do I clean the stove?",
            conversation=ConversationAccount(),
            recipient=ChannelAccount(),
            from_property=ChannelAccount(),
        )

        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)

        context = TestContext(activity)

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            result = await qna.get_answers(context)

            qna_trace_activities = list(
                filter(
                    lambda act: act.type == "trace" and act.name == "QnAMaker",
                    context.sent,
                )
            )
            trace_activity = qna_trace_activities[0]

            self.assertEqual("trace", trace_activity.type)
            self.assertEqual("QnAMaker", trace_activity.name)
            self.assertEqual("QnAMaker Trace", trace_activity.label)
            self.assertEqual(
                "https://www.qnamaker.ai/schemas/trace", trace_activity.value_type
            )
            self.assertEqual(True, hasattr(trace_activity, "value"))
            self.assertEqual(True, hasattr(trace_activity.value, "message"))
            self.assertEqual(True, hasattr(trace_activity.value, "query_results"))
            self.assertEqual(True, hasattr(trace_activity.value, "score_threshold"))
            self.assertEqual(True, hasattr(trace_activity.value, "top"))
            self.assertEqual(True, hasattr(trace_activity.value, "strict_filters"))
            self.assertEqual(
                self._knowledge_base_id, trace_activity.value.knowledge_base_id
            )

            return result

    async def test_returns_answer_with_timeout(self):
        question: str = "how do I clean the stove?"
        options = QnAMakerOptions(timeout=999999)
        qna = QnAMaker(QnaApplicationTest.tests_endpoint, options)
        context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            result = await qna.get_answers(context, options)

            self.assertIsNotNone(result)
            self.assertEqual(
                options.timeout, qna._generate_answer_helper.options.timeout
            )

    async def test_returns_answer_using_requests_module_with_no_timeout(self):
        url = f"{QnaApplicationTest._host}/knowledgebases/{QnaApplicationTest._knowledge_base_id}/generateAnswer"
        question = GenerateAnswerRequestBody(
            question="how do I clean the stove?",
            top=1,
            score_threshold=0.3,
            strict_filters=[],
            context=None,
            qna_id=None,
            is_test=False,
            ranker_type="Default",
        )
        response_path = "ReturnsAnswer.json"
        response_json = QnaApplicationTest._get_json_for_file(response_path)

        http_request_helper = HttpRequestUtils(requests)

        with patch("requests.post", return_value=response_json):
            result = await http_request_helper.execute_http_request(
                url, question, QnaApplicationTest.tests_endpoint, timeout=None
            )
            answers = result["answers"]

            self.assertIsNotNone(result)
            self.assertEqual(1, len(answers))
            self.assertEqual(
                "BaseCamp: You can use a damp rag to clean around the Power Pack",
                answers[0]["answer"],
            )

    async def test_telemetry_returns_answer(self):
        # Arrange
        question: str = "how do I clean the stove?"
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")
        telemetry_client = unittest.mock.create_autospec(BotTelemetryClient)
        log_personal_information = True
        context = QnaApplicationTest._get_context(question, TestAdapter())
        qna = QnAMaker(
            QnaApplicationTest.tests_endpoint,
            telemetry_client=telemetry_client,
            log_personal_information=log_personal_information,
        )

        # Act
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context)

            telemetry_args = telemetry_client.track_event.call_args_list[0][1]
            telemetry_properties = telemetry_args["properties"]
            telemetry_metrics = telemetry_args["measurements"]
            number_of_args = len(telemetry_args)
            first_answer = telemetry_args["properties"][
                QnATelemetryConstants.answer_property
            ]
            expected_answer = (
                "BaseCamp: You can use a damp rag to clean around the Power Pack"
            )

            # Assert - Check Telemetry logged.
            self.assertEqual(1, telemetry_client.track_event.call_count)
            self.assertEqual(3, number_of_args)
            self.assertEqual("QnaMessage", telemetry_args["name"])
            self.assertTrue("answer" in telemetry_properties)
            self.assertTrue("knowledgeBaseId" in telemetry_properties)
            self.assertTrue("matchedQuestion" in telemetry_properties)
            self.assertTrue("question" in telemetry_properties)
            self.assertTrue("questionId" in telemetry_properties)
            self.assertTrue("articleFound" in telemetry_properties)
            self.assertEqual(expected_answer, first_answer)
            self.assertTrue("score" in telemetry_metrics)
            self.assertEqual(1, telemetry_metrics["score"])

            # Assert - Validate we didn't break QnA functionality.
            self.assertIsNotNone(results)
            self.assertEqual(1, len(results))
            self.assertEqual(expected_answer, results[0].answer)
            self.assertEqual("Editorial", results[0].source)

    async def test_telemetry_returns_answer_when_no_answer_found_in_kb(self):
        # Arrange
        question: str = "gibberish question"
        response_json = QnaApplicationTest._get_json_for_file("NoAnswerFoundInKb.json")
        telemetry_client = unittest.mock.create_autospec(BotTelemetryClient)
        qna = QnAMaker(
            QnaApplicationTest.tests_endpoint,
            telemetry_client=telemetry_client,
            log_personal_information=True,
        )
        context = QnaApplicationTest._get_context(question, TestAdapter())

        # Act
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context)

            telemetry_args = telemetry_client.track_event.call_args_list[0][1]
            telemetry_properties = telemetry_args["properties"]
            number_of_args = len(telemetry_args)
            first_answer = telemetry_args["properties"][
                QnATelemetryConstants.answer_property
            ]
            expected_answer = "No Qna Answer matched"
            expected_matched_question = "No Qna Question matched"

            # Assert - Check Telemetry logged.
            self.assertEqual(1, telemetry_client.track_event.call_count)
            self.assertEqual(3, number_of_args)
            self.assertEqual("QnaMessage", telemetry_args["name"])
            self.assertTrue("answer" in telemetry_properties)
            self.assertTrue("knowledgeBaseId" in telemetry_properties)
            self.assertTrue("matchedQuestion" in telemetry_properties)
            self.assertEqual(
                expected_matched_question,
                telemetry_properties[QnATelemetryConstants.matched_question_property],
            )
            self.assertTrue("question" in telemetry_properties)
            self.assertTrue("questionId" in telemetry_properties)
            self.assertTrue("articleFound" in telemetry_properties)
            self.assertEqual(expected_answer, first_answer)

            # Assert - Validate we didn't break QnA functionality.
            self.assertIsNotNone(results)
            self.assertEqual(0, len(results))

    async def test_telemetry_pii(self):
        # Arrange
        question: str = "how do I clean the stove?"
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")
        telemetry_client = unittest.mock.create_autospec(BotTelemetryClient)
        log_personal_information = False
        context = QnaApplicationTest._get_context(question, TestAdapter())
        qna = QnAMaker(
            QnaApplicationTest.tests_endpoint,
            telemetry_client=telemetry_client,
            log_personal_information=log_personal_information,
        )

        # Act
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context)

            telemetry_args = telemetry_client.track_event.call_args_list[0][1]
            telemetry_properties = telemetry_args["properties"]
            telemetry_metrics = telemetry_args["measurements"]
            number_of_args = len(telemetry_args)
            first_answer = telemetry_args["properties"][
                QnATelemetryConstants.answer_property
            ]
            expected_answer = (
                "BaseCamp: You can use a damp rag to clean around the Power Pack"
            )

            # Assert - Validate PII properties not logged.
            self.assertEqual(1, telemetry_client.track_event.call_count)
            self.assertEqual(3, number_of_args)
            self.assertEqual("QnaMessage", telemetry_args["name"])
            self.assertTrue("answer" in telemetry_properties)
            self.assertTrue("knowledgeBaseId" in telemetry_properties)
            self.assertTrue("matchedQuestion" in telemetry_properties)
            self.assertTrue("question" not in telemetry_properties)
            self.assertTrue("questionId" in telemetry_properties)
            self.assertTrue("articleFound" in telemetry_properties)
            self.assertEqual(expected_answer, first_answer)
            self.assertTrue("score" in telemetry_metrics)
            self.assertEqual(1, telemetry_metrics["score"])

            # Assert - Validate we didn't break QnA functionality.
            self.assertIsNotNone(results)
            self.assertEqual(1, len(results))
            self.assertEqual(expected_answer, results[0].answer)
            self.assertEqual("Editorial", results[0].source)

    async def test_telemetry_override(self):
        # Arrange
        question: str = "how do I clean the stove?"
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")
        context = QnaApplicationTest._get_context(question, TestAdapter())
        options = QnAMakerOptions(top=1)
        telemetry_client = unittest.mock.create_autospec(BotTelemetryClient)
        log_personal_information = False

        # Act - Override the QnAMaker object to log custom stuff and honor params passed in.
        telemetry_properties: Dict[str, str] = {"id": "MyId"}
        qna = QnaApplicationTest.OverrideTelemetry(
            QnaApplicationTest.tests_endpoint,
            options,
            None,
            telemetry_client,
            log_personal_information,
        )
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context, options, telemetry_properties)

            telemetry_args = telemetry_client.track_event.call_args_list
            first_call_args = telemetry_args[0][0]
            first_call_properties = first_call_args[1]
            second_call_args = telemetry_args[1][0]
            second_call_properties = second_call_args[1]
            expected_answer = (
                "BaseCamp: You can use a damp rag to clean around the Power Pack"
            )

            # Assert
            self.assertEqual(2, telemetry_client.track_event.call_count)
            self.assertEqual(2, len(first_call_args))
            self.assertEqual("QnaMessage", first_call_args[0])
            self.assertEqual(2, len(first_call_properties))
            self.assertTrue("my_important_property" in first_call_properties)
            self.assertEqual(
                "my_important_value", first_call_properties["my_important_property"]
            )
            self.assertTrue("id" in first_call_properties)
            self.assertEqual("MyId", first_call_properties["id"])

            self.assertEqual("my_second_event", second_call_args[0])
            self.assertTrue("my_important_property2" in second_call_properties)
            self.assertEqual(
                "my_important_value2", second_call_properties["my_important_property2"]
            )

            # Validate we didn't break QnA functionality.
            self.assertIsNotNone(results)
            self.assertEqual(1, len(results))
            self.assertEqual(expected_answer, results[0].answer)
            self.assertEqual("Editorial", results[0].source)

    async def test_telemetry_additional_props_metrics(self):
        # Arrange
        question: str = "how do I clean the stove?"
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")
        context = QnaApplicationTest._get_context(question, TestAdapter())
        options = QnAMakerOptions(top=1)
        telemetry_client = unittest.mock.create_autospec(BotTelemetryClient)
        log_personal_information = False

        # Act
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            qna = QnAMaker(
                QnaApplicationTest.tests_endpoint,
                options,
                None,
                telemetry_client,
                log_personal_information,
            )
            telemetry_properties: Dict[str, str] = {
                "my_important_property": "my_important_value"
            }
            telemetry_metrics: Dict[str, float] = {"my_important_metric": 3.14159}

            results = await qna.get_answers(
                context, None, telemetry_properties, telemetry_metrics
            )

            # Assert - Added properties were added.
            telemetry_args = telemetry_client.track_event.call_args_list[0][1]
            telemetry_properties = telemetry_args["properties"]
            expected_answer = (
                "BaseCamp: You can use a damp rag to clean around the Power Pack"
            )

            self.assertEqual(1, telemetry_client.track_event.call_count)
            self.assertEqual(3, len(telemetry_args))
            self.assertEqual("QnaMessage", telemetry_args["name"])
            self.assertTrue("knowledgeBaseId" in telemetry_properties)
            self.assertTrue("question" not in telemetry_properties)
            self.assertTrue("matchedQuestion" in telemetry_properties)
            self.assertTrue("questionId" in telemetry_properties)
            self.assertTrue("answer" in telemetry_properties)
            self.assertTrue(expected_answer, telemetry_properties["answer"])
            self.assertTrue("my_important_property" in telemetry_properties)
            self.assertEqual(
                "my_important_value", telemetry_properties["my_important_property"]
            )

            tracked_metrics = telemetry_args["measurements"]

            self.assertEqual(2, len(tracked_metrics))
            self.assertTrue("score" in tracked_metrics)
            self.assertTrue("my_important_metric" in tracked_metrics)
            self.assertEqual(3.14159, tracked_metrics["my_important_metric"])

            # Assert - Validate we didn't break QnA functionality.
            self.assertIsNotNone(results)
            self.assertEqual(1, len(results))
            self.assertEqual(expected_answer, results[0].answer)
            self.assertEqual("Editorial", results[0].source)

    async def test_telemetry_additional_props_override(self):
        question: str = "how do I clean the stove?"
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")
        context = QnaApplicationTest._get_context(question, TestAdapter())
        options = QnAMakerOptions(top=1)
        telemetry_client = unittest.mock.create_autospec(BotTelemetryClient)
        log_personal_information = False

        # Act - Pass in properties during QnA invocation that override default properties
        # NOTE: We are invoking this with PII turned OFF, and passing a PII property (originalQuestion).
        qna = QnAMaker(
            QnaApplicationTest.tests_endpoint,
            options,
            None,
            telemetry_client,
            log_personal_information,
        )
        telemetry_properties = {
            "knowledge_base_id": "my_important_value",
            "original_question": "my_important_value2",
        }
        telemetry_metrics = {"score": 3.14159}

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(
                context, None, telemetry_properties, telemetry_metrics
            )

            # Assert - Added properties were added.
            tracked_args = telemetry_client.track_event.call_args_list[0][1]
            tracked_properties = tracked_args["properties"]
            expected_answer = (
                "BaseCamp: You can use a damp rag to clean around the Power Pack"
            )
            tracked_metrics = tracked_args["measurements"]

            self.assertEqual(1, telemetry_client.track_event.call_count)
            self.assertEqual(3, len(tracked_args))
            self.assertEqual("QnaMessage", tracked_args["name"])
            self.assertTrue("knowledge_base_id" in tracked_properties)
            self.assertEqual(
                "my_important_value", tracked_properties["knowledge_base_id"]
            )
            self.assertTrue("original_question" in tracked_properties)
            self.assertTrue("matchedQuestion" in tracked_properties)
            self.assertEqual(
                "my_important_value2", tracked_properties["original_question"]
            )
            self.assertTrue("question" not in tracked_properties)
            self.assertTrue("questionId" in tracked_properties)
            self.assertTrue("answer" in tracked_properties)
            self.assertEqual(expected_answer, tracked_properties["answer"])
            self.assertTrue("my_important_property" not in tracked_properties)
            self.assertEqual(1, len(tracked_metrics))
            self.assertTrue("score" in tracked_metrics)
            self.assertEqual(3.14159, tracked_metrics["score"])

            # Assert - Validate we didn't break QnA functionality.
            self.assertIsNotNone(results)
            self.assertEqual(1, len(results))
            self.assertEqual(expected_answer, results[0].answer)
            self.assertEqual("Editorial", results[0].source)

    async def test_telemetry_fill_props_override(self):
        # Arrange
        question: str = "how do I clean the stove?"
        response_json = QnaApplicationTest._get_json_for_file("ReturnsAnswer.json")
        context: TurnContext = QnaApplicationTest._get_context(question, TestAdapter())
        options = QnAMakerOptions(top=1)
        telemetry_client = unittest.mock.create_autospec(BotTelemetryClient)
        log_personal_information = False

        # Act - Pass in properties during QnA invocation that override default properties
        #       In addition Override with derivation.  This presents an interesting question of order of setting
        #       properties.
        #       If I want to override "originalQuestion" property:
        #           - Set in "Stock" schema
        #           - Set in derived QnAMaker class
        #           - Set in GetAnswersAsync
        #       Logically, the GetAnswersAync should win.  But ultimately OnQnaResultsAsync decides since it is the last
        #       code to touch the properties before logging (since it actually logs the event).
        qna = QnaApplicationTest.OverrideFillTelemetry(
            QnaApplicationTest.tests_endpoint,
            options,
            None,
            telemetry_client,
            log_personal_information,
        )
        telemetry_properties: Dict[str, str] = {
            "knowledgeBaseId": "my_important_value",
            "matchedQuestion": "my_important_value2",
        }
        telemetry_metrics: Dict[str, float] = {"score": 3.14159}

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(
                context, None, telemetry_properties, telemetry_metrics
            )

            # Assert - Added properties were added.
            first_call_args = telemetry_client.track_event.call_args_list[0][0]
            first_properties = first_call_args[1]
            expected_answer = (
                "BaseCamp: You can use a damp rag to clean around the Power Pack"
            )
            first_metrics = first_call_args[2]

            self.assertEqual(2, telemetry_client.track_event.call_count)
            self.assertEqual(3, len(first_call_args))
            self.assertEqual("QnaMessage", first_call_args[0])
            self.assertEqual(6, len(first_properties))
            self.assertTrue("knowledgeBaseId" in first_properties)
            self.assertEqual("my_important_value", first_properties["knowledgeBaseId"])
            self.assertTrue("matchedQuestion" in first_properties)
            self.assertEqual("my_important_value2", first_properties["matchedQuestion"])
            self.assertTrue("questionId" in first_properties)
            self.assertTrue("answer" in first_properties)
            self.assertEqual(expected_answer, first_properties["answer"])
            self.assertTrue("articleFound" in first_properties)
            self.assertTrue("my_important_property" in first_properties)
            self.assertEqual(
                "my_important_value", first_properties["my_important_property"]
            )

            self.assertEqual(1, len(first_metrics))
            self.assertTrue("score" in first_metrics)
            self.assertEqual(3.14159, first_metrics["score"])

            # Assert - Validate we didn't break QnA functionality.
            self.assertIsNotNone(results)
            self.assertEqual(1, len(results))
            self.assertEqual(expected_answer, results[0].answer)
            self.assertEqual("Editorial", results[0].source)

    async def test_call_train(self):
        feedback_records = []

        feedback1 = FeedbackRecord(
            qna_id=1, user_id="test", user_question="How are you?"
        )

        feedback2 = FeedbackRecord(qna_id=2, user_id="test", user_question="What up??")

        feedback_records.extend([feedback1, feedback2])

        with patch.object(
            QnAMaker, "call_train", return_value=None
        ) as mocked_call_train:
            qna = QnAMaker(QnaApplicationTest.tests_endpoint)
            qna.call_train(feedback_records)

            mocked_call_train.assert_called_once_with(feedback_records)

    async def test_should_filter_low_score_variation(self):
        options = QnAMakerOptions(top=5)
        qna = QnAMaker(QnaApplicationTest.tests_endpoint, options)
        question: str = "Q11"
        context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file("TopNAnswer.json")

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context)
            self.assertEqual(4, len(results), "Should have received 4 answers.")

            filtered_results = qna.get_low_score_variation(results)
            self.assertEqual(
                3,
                len(filtered_results),
                "Should have 3 filtered answers after low score variation.",
            )

    async def test_should_answer_with_is_test_true(self):
        options = QnAMakerOptions(top=1, is_test=True)
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        question: str = "Q11"
        context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "QnaMaker_IsTest_true.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context, options=options)
            self.assertEqual(0, len(results), "Should have received zero answer.")

    async def test_should_answer_with_ranker_type_question_only(self):
        options = QnAMakerOptions(top=1, ranker_type="QuestionOnly")
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        question: str = "Q11"
        context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "QnaMaker_RankerType_QuestionOnly.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(context, options=options)
            self.assertEqual(2, len(results), "Should have received two answers.")

    async def test_should_answer_with_prompts(self):
        options = QnAMakerOptions(top=2)
        qna = QnAMaker(QnaApplicationTest.tests_endpoint, options)
        question: str = "how do I clean the stove?"
        turn_context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file("AnswerWithPrompts.json")

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertEqual(1, len(results), "Should have received 1 answers.")
            self.assertEqual(
                1, len(results[0].context.prompts), "Should have received 1 prompt."
            )

    async def test_should_answer_with_high_score_provided_context(self):
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        question: str = "where can I buy?"
        context = QnARequestContext(
            previous_qna_id=5, previous_user_query="how do I clean the stove?"
        )
        options = QnAMakerOptions(top=2, qna_id=55, context=context)
        turn_context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "AnswerWithHighScoreProvidedContext.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertEqual(1, len(results), "Should have received 1 answers.")
            self.assertEqual(1, results[0].score, "Score should be high.")

    async def test_should_answer_with_high_score_provided_qna_id(self):
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        question: str = "where can I buy?"

        options = QnAMakerOptions(top=2, qna_id=55)
        turn_context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "AnswerWithHighScoreProvidedContext.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertEqual(1, len(results), "Should have received 1 answers.")
            self.assertEqual(1, results[0].score, "Score should be high.")

    async def test_should_answer_with_low_score_without_provided_context(self):
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        question: str = "where can I buy?"
        options = QnAMakerOptions(top=2, context=None)

        turn_context = QnaApplicationTest._get_context(question, TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "AnswerWithLowScoreProvidedWithoutContext.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertEqual(
                2, len(results), "Should have received more than one answers."
            )
            self.assertEqual(True, results[0].score < 1, "Score should be low.")

    async def test_low_score_variation(self):
        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        options = QnAMakerOptions(top=5, context=None)

        turn_context = QnaApplicationTest._get_context("Q11", TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "QnaMaker_TopNAnswer.json"
        )

        # active learning enabled
        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertIsNotNone(results)
            self.assertEqual(4, len(results), "should get four results")

            filtered_results = qna.get_low_score_variation(results)
            self.assertIsNotNone(filtered_results)
            self.assertEqual(3, len(filtered_results), "should get three results")

        # active learning disabled
        turn_context = QnaApplicationTest._get_context("Q11", TestAdapter())
        response_json = QnaApplicationTest._get_json_for_file(
            "QnaMaker_TopNAnswer_DisableActiveLearning.json"
        )

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            results = await qna.get_answers(turn_context, options)
            self.assertIsNotNone(results)
            self.assertEqual(4, len(results), "should get four results")

            filtered_results = qna.get_low_score_variation(results)
            self.assertIsNotNone(filtered_results)
            self.assertEqual(3, len(filtered_results), "should get three results")

    @classmethod
    async def _get_service_result(
        cls,
        utterance: str,
        response_file: str,
        bot_adapter: BotAdapter = TestAdapter(),
        options: QnAMakerOptions = None,
    ) -> [dict]:
        response_json = QnaApplicationTest._get_json_for_file(response_file)

        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        context = QnaApplicationTest._get_context(utterance, bot_adapter)

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            result = await qna.get_answers(context, options)

            return result

    @classmethod
    async def _get_service_result_raw(
        cls,
        utterance: str,
        response_file: str,
        bot_adapter: BotAdapter = TestAdapter(),
        options: QnAMakerOptions = None,
    ) -> [dict]:
        response_json = QnaApplicationTest._get_json_for_file(response_file)

        qna = QnAMaker(QnaApplicationTest.tests_endpoint)
        context = QnaApplicationTest._get_context(utterance, bot_adapter)

        with patch(
            "aiohttp.ClientSession.post",
            return_value=aiounittest.futurized(response_json),
        ):
            result = await qna.get_answers_raw(context, options)

            return result

    @classmethod
    def _get_json_for_file(cls, response_file: str) -> object:
        curr_dir = path.dirname(path.abspath(__file__))
        response_path = path.join(curr_dir, "test_data", response_file)

        with open(response_path, "r", encoding="utf-8-sig") as file:
            response_str = file.read()
        response_json = json.loads(response_str)

        return response_json

    @staticmethod
    def _get_context(question: str, bot_adapter: BotAdapter) -> TurnContext:
        test_adapter = bot_adapter or TestAdapter()
        activity = Activity(
            type=ActivityTypes.message,
            text=question,
            conversation=ConversationAccount(),
            recipient=ChannelAccount(),
            from_property=ChannelAccount(),
        )

        return TurnContext(test_adapter, activity)

    class OverrideTelemetry(QnAMaker):
        def __init__(  # pylint: disable=useless-super-delegation
            self,
            endpoint: QnAMakerEndpoint,
            options: QnAMakerOptions,
            http_client: ClientSession,
            telemetry_client: BotTelemetryClient,
            log_personal_information: bool,
        ):
            super().__init__(
                endpoint,
                options,
                http_client,
                telemetry_client,
                log_personal_information,
            )

        async def on_qna_result(  # pylint: disable=unused-argument
            self,
            query_results: [QueryResult],
            turn_context: TurnContext,
            telemetry_properties: Dict[str, str] = None,
            telemetry_metrics: Dict[str, float] = None,
        ):
            properties = telemetry_properties or {}

            # get_answers overrides derived class
            properties["my_important_property"] = "my_important_value"

            # Log event
            self.telemetry_client.track_event(
                QnATelemetryConstants.qna_message_event, properties
            )

            # Create 2nd event.
            second_event_properties = {"my_important_property2": "my_important_value2"}
            self.telemetry_client.track_event(
                "my_second_event", second_event_properties
            )

    class OverrideFillTelemetry(QnAMaker):
        def __init__(  # pylint: disable=useless-super-delegation
            self,
            endpoint: QnAMakerEndpoint,
            options: QnAMakerOptions,
            http_client: ClientSession,
            telemetry_client: BotTelemetryClient,
            log_personal_information: bool,
        ):
            super().__init__(
                endpoint,
                options,
                http_client,
                telemetry_client,
                log_personal_information,
            )

        async def on_qna_result(
            self,
            query_results: [QueryResult],
            turn_context: TurnContext,
            telemetry_properties: Dict[str, str] = None,
            telemetry_metrics: Dict[str, float] = None,
        ):
            event_data = await self.fill_qna_event(
                query_results, turn_context, telemetry_properties, telemetry_metrics
            )

            # Add my property.
            event_data.properties.update(
                {"my_important_property": "my_important_value"}
            )

            # Log QnaMessage event.
            self.telemetry_client.track_event(
                QnATelemetryConstants.qna_message_event,
                event_data.properties,
                event_data.metrics,
            )

            # Create second event.
            second_event_properties: Dict[str, str] = {
                "my_important_property2": "my_important_value2"
            }

            self.telemetry_client.track_event("MySecondEvent", second_event_properties)