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
|
# pylint: disable=line-too-long,useless-suppression
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import AzureAISearchIndex, IndexType
from test_base import TestBase, servicePreparer
from devtools_testutils import recorded_by_proxy
class TestIndexes(TestBase):
# To run this test, use the following command in the \sdk\ai\azure-ai-projects folder:
# cls & pytest tests\test_indexes.py::TestIndexes::test_indexes -s
@servicePreparer()
@recorded_by_proxy
def test_indexes(self, **kwargs):
endpoint = kwargs.pop("azure_ai_projects_tests_project_endpoint")
print("\n=====> Endpoint:", endpoint)
index_name = self.test_indexes_params["index_name"]
index_version = self.test_indexes_params["index_version"]
ai_search_connection_name = self.test_indexes_params["ai_search_connection_name"]
ai_search_index_name = self.test_indexes_params["ai_search_index_name"]
with AIProjectClient(
endpoint=endpoint,
credential=self.get_credential(AIProjectClient, is_async=False),
) as project_client:
print(
f"[test_indexes] Create Index `{index_name}` with version `{index_version}`, referencing an existing AI Search resource:"
)
index = project_client.indexes.create_or_update(
name=index_name,
version=index_version,
index=AzureAISearchIndex(connection_name=ai_search_connection_name, index_name=ai_search_index_name),
)
print(index)
TestBase.validate_index(
index,
expected_index_type=IndexType.AZURE_SEARCH,
expected_index_name=index_name,
expected_index_version=index_version,
expected_ai_search_connection_name=ai_search_connection_name,
expected_ai_search_index_name=ai_search_index_name,
)
print(f"[test_indexes] Get Index `{index_name}` version `{index_version}`:")
index = project_client.indexes.get(name=index_name, version=index_version)
print(index)
TestBase.validate_index(
index,
expected_index_type=IndexType.AZURE_SEARCH,
expected_index_name=index_name,
expected_index_version=index_version,
expected_ai_search_connection_name=ai_search_connection_name,
expected_ai_search_index_name=ai_search_index_name,
)
print("[test_indexes] List latest versions of all Indexes:")
empty = True
for index in project_client.indexes.list():
empty = False
print(index)
TestBase.validate_index(index)
assert not empty
print(f"[test_indexes] Listing all versions of the Index named `{index_name}`:")
empty = True
for index in project_client.indexes.list_versions(name=index_name):
empty = False
print(index)
TestBase.validate_index(index)
assert not empty
print(f"[test_indexes] Delete Index `{index_name}` version `{index_version}`.")
project_client.indexes.delete(name=index_name, version=index_version)
print(
f"[test_indexes] Again delete Index `{index_name}` version `{index_version}`. Since it does not exist, the REST API should return 204 (No content). This call should NOT throw an exception."
)
project_client.indexes.delete(name=index_name, version=index_version)
|
