File: test_image_analysis_async_client.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (106 lines) | stat: -rw-r--r-- 3,928 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
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
import inspect
import azure.ai.vision.imageanalysis as sdk

from image_analysis_test_base import ImageAnalysisTestBase, ServicePreparer
from devtools_testutils.aio import recorded_by_proxy_async


# The test class name needs to start with "Test" to get collected by pytest
class TestImageAnalysisAsyncClient(ImageAnalysisTestBase):

    # **********************************************************************************
    #
    #                            HAPPY PATH TESTS
    #
    # **********************************************************************************

    # Test all visual features from a local image, using default settings
    @ServicePreparer()
    @recorded_by_proxy_async
    async def test_analyze_async_all_features_from_file(self, **kwargs):

        self._create_client_for_standard_analysis(sync=False, **kwargs)

        await self._do_async_analysis(
            image_source=self.IMAGE_FILE,
            visual_features=[
                sdk.models.VisualFeatures.TAGS,
                sdk.models.VisualFeatures.OBJECTS,
                sdk.models.VisualFeatures.CAPTION,
                sdk.models.VisualFeatures.DENSE_CAPTIONS,
                sdk.models.VisualFeatures.READ,
                sdk.models.VisualFeatures.SMART_CROPS,
                sdk.models.VisualFeatures.PEOPLE,
            ],
            **kwargs
        )

        await self.async_client.close()

    # Test some visual features, one after the other, from image URL, with relevant settings specified
    @ServicePreparer()
    @recorded_by_proxy_async
    async def test_analyze_async_single_feature_from_url(self, **kwargs):

        self._create_client_for_standard_analysis(sync=False, **kwargs)

        await self._do_async_analysis(
            image_source=self.IMAGE_URL,
            visual_features=[sdk.models.VisualFeatures.DENSE_CAPTIONS],
            gender_neutral_caption=True,
            **kwargs
        )

        await self._do_async_analysis(
            image_source=self.IMAGE_URL,
            visual_features=[sdk.models.VisualFeatures.SMART_CROPS],
            smart_crops_aspect_ratios=[0.9, 1.33],
            **kwargs
        )

        await self._do_async_analysis(
            image_source=self.IMAGE_URL, visual_features=[sdk.models.VisualFeatures.TAGS], language="en", **kwargs
        )

        await self._do_async_analysis(
            image_source=self.IMAGE_URL, visual_features=[sdk.models.VisualFeatures.PEOPLE], **kwargs
        )

        await self.async_client.close()

    # Test a single visual feature from an image url, using Entra ID authentication
    @ServicePreparer()
    @recorded_by_proxy_async
    async def test_analyze_async_single_feature_from_file_entra_id_auth(self, **kwargs):

        self._create_client_for_standard_analysis_with_entra_id_auth(sync=False, **kwargs)

        await self._do_async_analysis(image_source=self.IMAGE_FILE,visual_features=[sdk.models.VisualFeatures.SMART_CROPS], **kwargs)

        await self.async_client.close()

    # **********************************************************************************
    #
    #                            ERROR TESTS
    #
    # **********************************************************************************

    @ServicePreparer()
    @recorded_by_proxy_async
    async def test_analyze_async_authentication_failure(self, **kwargs):

        self._create_client_for_authentication_failure(sync=False, **kwargs)

        await self._do_async_analysis_with_error(
            image_source=self.IMAGE_URL,
            visual_features=[sdk.models.VisualFeatures.TAGS],
            expected_status_code=401,
            expected_message_contains="Access denied",
            **kwargs
        )

        await self.async_client.close()