Spaces:
Build error
Build error
# 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") | |
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 | |
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 | |
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 | |
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) | |