File: test_models_async.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 (81 lines) | stat: -rw-r--r-- 2,477 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
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------

import os
import pytest
import pathlib
import uuid
from devtools_testutils import AzureRecordedTestCase
from conftest import configure_async, ASST_AZURE, OPENAI, AZURE, PREVIEW, GA


@pytest.mark.live_test_only
class TestModelsAsync(AzureRecordedTestCase):

    @pytest.mark.skip("InternalServerError, opened issue")
    @configure_async
    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        "api_type, api_version",
        [(AZURE, GA), (AZURE, PREVIEW), (OPENAI, "v1")]
    )
    async def test_models_list(self, client_async, api_type, api_version, **kwargs):

        models = client_async.models.list()
        async for model in models:
            assert model.id

    @configure_async
    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        "api_type, api_version",
        [(AZURE, GA), (AZURE, PREVIEW), (OPENAI, "v1")]
    )
    async def test_models_retrieve(self, client_async, api_type, api_version, **kwargs):

        model = await client_async.models.retrieve(**kwargs)
        assert model.id

    @configure_async
    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        "api_type, api_version",
        [(OPENAI, "v1"), (ASST_AZURE, PREVIEW)]
    )
    async def test_files(self, client_async, api_type, api_version, **kwargs):
        file_name = f"test{uuid.uuid4()}.txt"
        with open(file_name, "w") as f:
            f.write("test")

        try:
            path = pathlib.Path(file_name)

            file1 = await client_async.files.create(
                file=open(path, "rb"),
                purpose="assistants"
            )

            files = client_async.files.list()
            async for file in files:
                assert file.id
                break

            files = client_async.files.list(purpose="assistants")
            async for file in files:
                assert file.id
                break

            retrieved_file = await client_async.files.retrieve(file1.id)
            assert retrieved_file.id
            assert retrieved_file.purpose == "assistants"
            assert retrieved_file.filename == file_name
            assert retrieved_file.created_at
            assert retrieved_file.bytes

        finally:
            await client_async.files.delete(
                file1.id
            )
            os.remove(file_name)