File: test_update_knowledge_sources_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 (111 lines) | stat: -rw-r--r-- 5,041 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
105
106
107
108
109
110
111
import pytest
from typing import cast
from azure.core.credentials import AzureKeyCredential
from azure.ai.language.questionanswering.authoring.aio import QuestionAnsweringAuthoringClient
from azure.ai.language.questionanswering.authoring import models as _models

from helpers import AuthoringAsyncTestHelper
from testcase import QuestionAnsweringAuthoringTestCase


class TestSourcesQnasSynonymsAsync(QuestionAnsweringAuthoringTestCase):
    @pytest.mark.asyncio
    async def test_add_source(self, recorded_test, qna_authoring_creds):  # type: ignore[name-defined]
        client = QuestionAnsweringAuthoringClient(
            qna_authoring_creds["endpoint"], AzureKeyCredential(qna_authoring_creds["key"])
        )
        project_name = "IsaacNewton"
        async with client:
            await AuthoringAsyncTestHelper.create_test_project(
                client, project_name=project_name, polling_interval=0 if self.is_playback else None
            )
            update_source_ops = [
                _models.UpdateSourceRecord(
                    {
                        "op": "add",
                        "value": {
                            "displayName": "MicrosoftFAQ",
                            "source": "https://www.microsoft.com/en-in/software-download/faq",
                            "sourceUri": "https://www.microsoft.com/en-in/software-download/faq",
                            "sourceKind": "url",
                            "contentStructureKind": "unstructured",
                            "refresh": False,
                        },
                    }
                )
            ]
            poller = await client.begin_update_sources(
                project_name=project_name,
                sources=cast(list[_models.UpdateSourceRecord], update_source_ops),
                content_type="application/json",
                polling_interval=0 if self.is_playback else None,  # type: ignore[arg-type]
            )
            await poller.result()
            found = False
            async for s in client.list_sources(project_name=project_name):
                if s.get("displayName") == "MicrosoftFAQ":
                    found = True
            assert found

    @pytest.mark.asyncio
    async def test_add_qna(self, recorded_test, qna_authoring_creds):  # type: ignore[name-defined]
        client = QuestionAnsweringAuthoringClient(
            qna_authoring_creds["endpoint"], AzureKeyCredential(qna_authoring_creds["key"])
        )
        project_name = "IsaacNewton"
        async with client:
            await AuthoringAsyncTestHelper.create_test_project(
                client, project_name=project_name, polling_interval=0 if self.is_playback else None
            )
            question = "What is the easiest way to use azure services in my .NET project?"
            answer = "Using Microsoft's Azure SDKs"
            update_qna_ops = [
                _models.UpdateQnaRecord(
                    {
                        "op": "add",
                        "value": {
                            "id": 0,
                            "answer": answer,
                            "questions": [question],
                        },
                    }
                )
            ]
            poller = await client.begin_update_qnas(
                project_name=project_name,
                qnas=cast(list[_models.UpdateQnaRecord], update_qna_ops),
                content_type="application/json",
                polling_interval=0 if self.is_playback else None,  # type: ignore[arg-type]
            )
            await poller.result()
            found = False
            async for qna in client.list_qnas(project_name=project_name):
                if qna.get("answer") == answer and question in qna.get("questions", []):
                    found = True
            assert found

    @pytest.mark.asyncio
    async def test_add_synonym(self, recorded_test, qna_authoring_creds):  # type: ignore[name-defined]
        client = QuestionAnsweringAuthoringClient(
            qna_authoring_creds["endpoint"], AzureKeyCredential(qna_authoring_creds["key"])
        )
        project_name = "IsaacNewton"
        async with client:
            await AuthoringAsyncTestHelper.create_test_project(
                client, project_name=project_name, polling_interval=0 if self.is_playback else None
            )
            synonyms_model = _models.SynonymAssets(
                value=[
                    _models.WordAlterations(alterations=["qnamaker", "qna maker"]),
                ]
            )
            await client.update_synonyms(
                project_name=project_name,
                synonyms=synonyms_model,
                content_type="application/json",
            )
            found = False
            async for s in client.list_synonyms(project_name=project_name):
                if "qnamaker" in s.get("alterations", []) and "qna maker" in s.get("alterations", []):
                    found = True
            assert found