File: test_query_knowledgebase_async.py

package info (click to toggle)
python-azure 20251118%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 783,356 kB
  • sloc: python: 6,474,533; ansic: 804; javascript: 287; sh: 205; makefile: 198; xml: 109
file content (104 lines) | stat: -rw-r--r-- 4,531 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
# pylint: disable=line-too-long,useless-suppression
# coding=utf-8
# -------------------------------------------------------------------------
# Runtime async tests: knowledge base querying (authoring removed)
# -------------------------------------------------------------------------
import pytest

from azure.ai.language.questionanswering.aio import QuestionAnsweringClient
from azure.ai.language.questionanswering.models import (
    AnswersOptions,
    KnowledgeBaseAnswerContext,
    ShortAnswerOptions,
    QueryFilters,
    MetadataFilter,
    MetadataRecord,
)
from azure.core.credentials import AzureKeyCredential

from testcase import QuestionAnsweringTestCase


class TestQueryKnowledgeBaseAsync(QuestionAnsweringTestCase):
    @pytest.mark.asyncio
    async def test_query_knowledgebase_basic(self, recorded_test, qna_creds):
        client = QuestionAnsweringClient(qna_creds["qna_endpoint"], AzureKeyCredential(qna_creds["qna_key"]))
        params = AnswersOptions(
            question="Ports and connectors",
            top=3,
            answer_context=KnowledgeBaseAnswerContext(previous_question="Meet Surface Pro 4", previous_qna_id=4),
        )
        async with client:
            output = await client.get_answers(params, project_name=qna_creds["qna_project"], deployment_name="production")
        assert output.answers
        for answer in output.answers:
            assert answer.answer
            assert answer.confidence is not None
            assert answer.qna_id is not None
            assert answer.source
            assert answer.metadata is not None

    @pytest.mark.asyncio
    async def test_query_knowledgebase_with_short_answer(self, recorded_test, qna_creds):
        client = QuestionAnsweringClient(qna_creds["qna_endpoint"], AzureKeyCredential(qna_creds["qna_key"]))
        params = AnswersOptions(
            question="Ports and connectors",
            top=3,
            answer_context=KnowledgeBaseAnswerContext(previous_question="Meet Surface Pro 4", previous_qna_id=4),
            short_answer_options=ShortAnswerOptions(enable=True, confidence_threshold=0.1, top=2),
        )
        async with client:
            output = await client.get_answers(params, project_name=qna_creds["qna_project"], deployment_name="test")
        assert output.answers
        for answer in output.answers:
            if answer.short_answer:
                assert answer.short_answer.text
                assert answer.short_answer.confidence is not None

    @pytest.mark.asyncio
    async def test_query_knowledgebase_filter(self, recorded_test, qna_creds):
        filters = QueryFilters(
            metadata_filter=MetadataFilter(
                metadata=[
                    MetadataRecord(key="explicitlytaggedheading", value="check the battery level"),
                    MetadataRecord(key="explicitlytaggedheading", value="make your battery last"),
                ],
                logical_operation="OR",
            )
        )
        async with QuestionAnsweringClient(
            qna_creds["qna_endpoint"], AzureKeyCredential(qna_creds["qna_key"])
        ) as client:
            params = AnswersOptions(
                question="Battery life",
                top=3,
                filters=filters,
            )
            response = await client.get_answers(
                params,
                project_name=qna_creds["qna_project"],
                deployment_name="production",
            )
            assert response.answers
            assert any(
                [
                    a
                    for a in response.answers
                    if (a.metadata or {}).get("explicitlytaggedheading") == "check the battery level"
                ]
            )
            assert any(
                [
                    a
                    for a in response.answers
                    if (a.metadata or {}).get("explicitlytaggedheading") == "make your battery last"
                ]
            )

    @pytest.mark.asyncio
    async def test_query_knowledgebase_overload_errors(self):  # negative parameter validation
        async with QuestionAnsweringClient("http://fake.com", AzureKeyCredential("123")) as client:
            with pytest.raises(TypeError):
                await client.get_answers("positional_one", "positional_two")  # type: ignore[arg-type]
            with pytest.raises(TypeError):
                await client.get_answers("positional_options_bag", options="options bag by name")  # type: ignore[arg-type]