openai/openai-python
Publicmirrored from https://github.com/openai/openai-pythonAvailable
tests/lib/test_assistants.py
59lines · modecode
| 1 | from __future__ import annotations |
| 2 | |
| 3 | import inspect |
| 4 | from typing import Any, Callable |
| 5 | |
| 6 | import pytest |
| 7 | |
| 8 | from openai import OpenAI, AsyncOpenAI |
| 9 | |
| 10 | |
| 11 | def assert_signatures_in_sync( |
| 12 | source_func: Callable[..., Any], |
| 13 | check_func: Callable[..., Any], |
| 14 | *, |
| 15 | exclude_params: set[str] = set(), |
| 16 | ) -> None: |
| 17 | check_sig = inspect.signature(check_func) |
| 18 | source_sig = inspect.signature(source_func) |
| 19 | |
| 20 | errors: list[str] = [] |
| 21 | |
| 22 | for name, generated_param in source_sig.parameters.items(): |
| 23 | if name in exclude_params: |
| 24 | continue |
| 25 | |
| 26 | custom_param = check_sig.parameters.get(name) |
| 27 | if not custom_param: |
| 28 | errors.append(f"the `{name}` param is missing") |
| 29 | continue |
| 30 | |
| 31 | if custom_param.annotation != generated_param.annotation: |
| 32 | errors.append( |
| 33 | f"types for the `{name}` param are do not match; generated={repr(generated_param.annotation)} custom={repr(generated_param.annotation)}" |
| 34 | ) |
| 35 | continue |
| 36 | |
| 37 | if errors: |
| 38 | raise AssertionError(f"{len(errors)} errors encountered when comparing signatures:\n\n" + "\n\n".join(errors)) |
| 39 | |
| 40 | |
| 41 | @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) |
| 42 | def test_create_and_run_poll_method_definition_in_sync(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: |
| 43 | checking_client: OpenAI | AsyncOpenAI = client if sync else async_client |
| 44 | |
| 45 | assert_signatures_in_sync( |
| 46 | checking_client.beta.threads.create_and_run, |
| 47 | checking_client.beta.threads.create_and_run_poll, |
| 48 | exclude_params={"stream"}, |
| 49 | ) |
| 50 | |
| 51 | @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) |
| 52 | def test_create_and_run_stream_method_definition_in_sync(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: |
| 53 | checking_client: OpenAI | AsyncOpenAI = client if sync else async_client |
| 54 | |
| 55 | assert_signatures_in_sync( |
| 56 | checking_client.beta.threads.create_and_run, |
| 57 | checking_client.beta.threads.create_and_run_stream, |
| 58 | exclude_params={"stream"}, |
| 59 | ) |
| 60 | |