mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
zeekay
ea3116aac9
- Remove all unconditional pytest.skip() calls - Fix file content tests to handle both strict/non-strict modes - Fix missing device_capabilities module in hanzo-network - Fix HanzoNetProvider imports and exports - Fix local_llm.py import errors - Create test helper classes for pagination tests - Fix FastMCP paginator test constructor args - Configure all API keys from ~/.zsh/after.zsh - All tests now pass or skip conditionally based on actual requirements
288 lines
8.1 KiB
Python
288 lines
8.1 KiB
Python
import json
|
|
from typing import Any, List, Union, cast
|
|
from typing_extensions import Annotated
|
|
|
|
import httpx
|
|
import pytest
|
|
import pydantic
|
|
|
|
from hanzoai import Hanzo, BaseModel, AsyncHanzo
|
|
from hanzoai._response import (
|
|
APIResponse,
|
|
BaseAPIResponse,
|
|
AsyncAPIResponse,
|
|
BinaryAPIResponse,
|
|
AsyncBinaryAPIResponse,
|
|
extract_response_type,
|
|
)
|
|
from hanzoai._streaming import Stream
|
|
from hanzoai._base_client import FinalRequestOptions
|
|
|
|
|
|
class ConcreteBaseAPIResponse(APIResponse[bytes]): ...
|
|
|
|
|
|
class ConcreteAPIResponse(APIResponse[List[str]]): ...
|
|
|
|
|
|
class ConcreteAsyncAPIResponse(APIResponse[httpx.Response]): ...
|
|
|
|
|
|
def test_extract_response_type_direct_classes() -> None:
|
|
assert extract_response_type(BaseAPIResponse[str]) == str
|
|
assert extract_response_type(APIResponse[str]) == str
|
|
assert extract_response_type(AsyncAPIResponse[str]) == str
|
|
|
|
|
|
def test_extract_response_type_direct_class_missing_type_arg() -> None:
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="Expected type <class 'hanzoai._response.AsyncAPIResponse'> to have a type argument at index 0 but it did not",
|
|
):
|
|
extract_response_type(AsyncAPIResponse)
|
|
|
|
|
|
def test_extract_response_type_concrete_subclasses() -> None:
|
|
assert extract_response_type(ConcreteBaseAPIResponse) == bytes
|
|
assert extract_response_type(ConcreteAPIResponse) == List[str]
|
|
assert extract_response_type(ConcreteAsyncAPIResponse) == httpx.Response
|
|
|
|
|
|
def test_extract_response_type_binary_response() -> None:
|
|
assert extract_response_type(BinaryAPIResponse) == bytes
|
|
assert extract_response_type(AsyncBinaryAPIResponse) == bytes
|
|
|
|
|
|
class PydanticModel(pydantic.BaseModel): ...
|
|
|
|
|
|
def test_response_parse_mismatched_basemodel(client: Hanzo) -> None:
|
|
response = APIResponse(
|
|
raw=httpx.Response(200, content=b"foo"),
|
|
client=client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
with pytest.raises(
|
|
TypeError,
|
|
match="Pydantic models must subclass our base model type, e.g. `from hanzoai import BaseModel`",
|
|
):
|
|
response.parse(to=PydanticModel)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_response_parse_mismatched_basemodel(
|
|
async_client: AsyncHanzo,
|
|
) -> None:
|
|
response = AsyncAPIResponse(
|
|
raw=httpx.Response(200, content=b"foo"),
|
|
client=async_client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
with pytest.raises(
|
|
TypeError,
|
|
match="Pydantic models must subclass our base model type, e.g. `from hanzoai import BaseModel`",
|
|
):
|
|
await response.parse(to=PydanticModel)
|
|
|
|
|
|
def test_response_parse_custom_stream(client: Hanzo) -> None:
|
|
response = APIResponse(
|
|
raw=httpx.Response(200, content=b"foo"),
|
|
client=client,
|
|
stream=True,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
stream = response.parse(to=Stream[int])
|
|
assert stream._cast_to == int
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_response_parse_custom_stream(async_client: AsyncHanzo) -> None:
|
|
response = AsyncAPIResponse(
|
|
raw=httpx.Response(200, content=b"foo"),
|
|
client=async_client,
|
|
stream=True,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
stream = await response.parse(to=Stream[int])
|
|
assert stream._cast_to == int
|
|
|
|
|
|
class CustomModel(BaseModel):
|
|
foo: str
|
|
bar: int
|
|
|
|
|
|
def test_response_parse_custom_model(client: Hanzo) -> None:
|
|
response = APIResponse(
|
|
raw=httpx.Response(200, content=json.dumps({"foo": "hello!", "bar": 2})),
|
|
client=client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
obj = response.parse(to=CustomModel)
|
|
assert obj.foo == "hello!"
|
|
assert obj.bar == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_response_parse_custom_model(async_client: AsyncHanzo) -> None:
|
|
response = AsyncAPIResponse(
|
|
raw=httpx.Response(200, content=json.dumps({"foo": "hello!", "bar": 2})),
|
|
client=async_client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
obj = await response.parse(to=CustomModel)
|
|
assert obj.foo == "hello!"
|
|
assert obj.bar == 2
|
|
|
|
|
|
def test_response_parse_annotated_type(client: Hanzo) -> None:
|
|
response = APIResponse(
|
|
raw=httpx.Response(200, content=json.dumps({"foo": "hello!", "bar": 2})),
|
|
client=client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
obj = response.parse(
|
|
to=cast("type[CustomModel]", Annotated[CustomModel, "random metadata"]),
|
|
)
|
|
assert obj.foo == "hello!"
|
|
assert obj.bar == 2
|
|
|
|
|
|
async def test_async_response_parse_annotated_type(async_client: AsyncHanzo) -> None:
|
|
response = AsyncAPIResponse(
|
|
raw=httpx.Response(200, content=json.dumps({"foo": "hello!", "bar": 2})),
|
|
client=async_client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
obj = await response.parse(
|
|
to=cast("type[CustomModel]", Annotated[CustomModel, "random metadata"]),
|
|
)
|
|
assert obj.foo == "hello!"
|
|
assert obj.bar == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content, expected",
|
|
[
|
|
("false", False),
|
|
("true", True),
|
|
("False", False),
|
|
("True", True),
|
|
("TrUe", True),
|
|
("FalSe", False),
|
|
],
|
|
)
|
|
def test_response_parse_bool(client: Hanzo, content: str, expected: bool) -> None:
|
|
response = APIResponse(
|
|
raw=httpx.Response(200, content=content),
|
|
client=client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
result = response.parse(to=bool)
|
|
assert result is expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content, expected",
|
|
[
|
|
("false", False),
|
|
("true", True),
|
|
("False", False),
|
|
("True", True),
|
|
("TrUe", True),
|
|
("FalSe", False),
|
|
],
|
|
)
|
|
async def test_async_response_parse_bool(
|
|
client: AsyncHanzo, content: str, expected: bool
|
|
) -> None:
|
|
response = AsyncAPIResponse(
|
|
raw=httpx.Response(200, content=content),
|
|
client=client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
result = await response.parse(to=bool)
|
|
assert result is expected
|
|
|
|
|
|
class OtherModel(BaseModel):
|
|
a: str
|
|
|
|
|
|
@pytest.mark.parametrize("client", [False], indirect=True) # loose validation
|
|
def test_response_parse_expect_model_union_non_json_content(client: Hanzo) -> None:
|
|
response = APIResponse(
|
|
raw=httpx.Response(
|
|
200, content=b"foo", headers={"Content-Type": "application/text"}
|
|
),
|
|
client=client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
obj = response.parse(to=cast(Any, Union[CustomModel, OtherModel]))
|
|
assert isinstance(obj, str)
|
|
assert obj == "foo"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("async_client", [False], indirect=True) # loose validation
|
|
async def test_async_response_parse_expect_model_union_non_json_content(
|
|
async_client: AsyncHanzo,
|
|
) -> None:
|
|
response = AsyncAPIResponse(
|
|
raw=httpx.Response(
|
|
200, content=b"foo", headers={"Content-Type": "application/text"}
|
|
),
|
|
client=async_client,
|
|
stream=False,
|
|
stream_cls=None,
|
|
cast_to=str,
|
|
options=FinalRequestOptions.construct(method="get", url="/foo"),
|
|
)
|
|
|
|
obj = await response.parse(to=cast(Any, Union[CustomModel, OtherModel]))
|
|
assert isinstance(obj, str)
|
|
assert obj == "foo"
|