File: test_import_project.py

package info (click to toggle)
python-azure 20251104%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky
  • size: 770,224 kB
  • sloc: python: 6,357,217; ansic: 804; javascript: 287; makefile: 198; sh: 193; xml: 109
file content (120 lines) | stat: -rw-r--r-- 4,000 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
112
113
114
115
116
117
118
119
120
# pylint: disable=line-too-long,useless-suppression
import functools
import json

from devtools_testutils import AzureRecordedTestCase, EnvironmentVariableLoader, recorded_by_proxy
from azure.core.credentials import AzureKeyCredential
from azure.core.exceptions import HttpResponseError
from azure.ai.language.conversations.authoring import ConversationAuthoringClient
from azure.ai.language.conversations.authoring.models import (
    CreateProjectOptions,
    ProjectKind,
    ProjectSettings,
    ExportedProject,
    ExportedProjectFormat,
    ConversationExportedProjectAsset,
    ConversationExportedIntent,
    ConversationExportedEntity,
    ConversationExportedUtterance,
    ExportedUtteranceEntityLabel,
)

ConversationsPreparer = functools.partial(
    EnvironmentVariableLoader,
    "authoring",
    authoring_endpoint="https://Sanitized.cognitiveservices.azure.com/",
    authoring_key="fake_key",
)


class TestConversations(AzureRecordedTestCase):
    # Start with any helper functions you might need, for example a client creation method:
    def create_client(self, endpoint, key):
        credential = AzureKeyCredential(key)
        client = ConversationAuthoringClient(endpoint, credential)
        return client

    # ...


class TestConversationsImportCase(TestConversations):
    @ConversationsPreparer()
    @recorded_by_proxy
    def test_import_project(self, authoring_endpoint, authoring_key):
        authoring_client = self.create_client(authoring_endpoint, authoring_key)
        project_name = "PythonImportProject0820"
        project_client = authoring_client.get_project_client(project_name)

        # ----- Build assets using objects -----
        intents = [
            ConversationExportedIntent(category="None"),
            ConversationExportedIntent(category="Buy"),
        ]

        entities = [
            ConversationExportedEntity(
                category="product",
                composition_mode="combineComponents",
            )
        ]

        u1 = ConversationExportedUtterance(
            text="I want to buy a house",
            intent="Buy",
            language="en-us",
            dataset="Train",
            entities=[ExportedUtteranceEntityLabel(category="product", offset=16, length=5)],
        )

        u2 = ConversationExportedUtterance(
            text="I want to buy surface pro",
            intent="Buy",
            language="en-us",
            dataset="Train",
            entities=[ExportedUtteranceEntityLabel(category="product", offset=14, length=11)],
        )

        u3 = ConversationExportedUtterance(
            text="I want to buy xbox",
            intent="Buy",
            language="en-us",
            dataset="Train",
            entities=[ExportedUtteranceEntityLabel(category="product", offset=14, length=4)],
        )

        assets = ConversationExportedProjectAsset(
            intents=intents,
            entities=entities,
            utterances=[u1, u2, u3],
        )

        metadata = CreateProjectOptions(
            project_kind=ProjectKind.CONVERSATION,
            project_name=project_name,  # required
            language="en-us",  # required
            settings=ProjectSettings(confidence_threshold=0.0),
            multilingual=False,
            description="",
        )

        exported_project = ExportedProject(
            project_file_version="2025-05-15-preview",
            string_index_type="Utf16CodeUnit",
            metadata=metadata,
            assets=assets,
        )

        # ----- Act: begin import (LRO) -----
        poller = project_client.project.begin_import(
            body=exported_project,
            exported_project_format=ExportedProjectFormat.CONVERSATION,
        )

        try:
            poller.result()
        except HttpResponseError as e:
            print(f"Operation failed: {e.message}")
            print(e.error)
            raise

        print(f"Import completed. done={poller.done()} status={poller.status()}")