vrr/opencode-sdk-python
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode-sdk-python.git synced 2026-08-13 02:25:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(api): update via SDK Studio

Browse source
This commit is contained in:
stainless-app[bot] 2025-06-27 14:25:51 +00:00
parent 604017133e
commit ff05a4adf0
130 changed files with 17166 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

1
tests/__init__.py Normal file
View file

@ -0,0 +1 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.

1
tests/api_resources/__init__.py Normal file
View file

@ -0,0 +1 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.

136
tests/api_resources/test_app.py Normal file
View file

@ -0,0 +1,136 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from opencode import Opencode, AsyncOpencode
from tests.utils import assert_matches_type
from opencode.types import App, AppInitResponse
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestApp:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@pytest.mark.skip()
@parametrize
def test_method_get(self, client: Opencode) -> None:
app = client.app.get()
assert_matches_type(App, app, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_get(self, client: Opencode) -> None:
response = client.app.with_raw_response.get()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = response.parse()
assert_matches_type(App, app, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_get(self, client: Opencode) -> None:
with client.app.with_streaming_response.get() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = response.parse()
assert_matches_type(App, app, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_method_init(self, client: Opencode) -> None:
app = client.app.init()
assert_matches_type(AppInitResponse, app, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_init(self, client: Opencode) -> None:
response = client.app.with_raw_response.init()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = response.parse()
assert_matches_type(AppInitResponse, app, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_init(self, client: Opencode) -> None:
with client.app.with_streaming_response.init() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = response.parse()
assert_matches_type(AppInitResponse, app, path=["response"])
assert cast(Any, response.is_closed) is True
class TestAsyncApp:
parametrize = pytest.mark.parametrize(
"async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
)
@pytest.mark.skip()
@parametrize
async def test_method_get(self, async_client: AsyncOpencode) -> None:
app = await async_client.app.get()
assert_matches_type(App, app, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_get(self, async_client: AsyncOpencode) -> None:
response = await async_client.app.with_raw_response.get()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = await response.parse()
assert_matches_type(App, app, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_get(self, async_client: AsyncOpencode) -> None:
async with async_client.app.with_streaming_response.get() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = await response.parse()
assert_matches_type(App, app, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_method_init(self, async_client: AsyncOpencode) -> None:
app = await async_client.app.init()
assert_matches_type(AppInitResponse, app, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_init(self, async_client: AsyncOpencode) -> None:
response = await async_client.app.with_raw_response.init()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = await response.parse()
assert_matches_type(AppInitResponse, app, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_init(self, async_client: AsyncOpencode) -> None:
async with async_client.app.with_streaming_response.init() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
app = await response.parse()
assert_matches_type(AppInitResponse, app, path=["response"])
assert cast(Any, response.is_closed) is True

136
tests/api_resources/test_config.py Normal file
View file

@ -0,0 +1,136 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from opencode import Opencode, AsyncOpencode
from tests.utils import assert_matches_type
from opencode.types import Config, ConfigProvidersResponse
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestConfig:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@pytest.mark.skip()
@parametrize
def test_method_get(self, client: Opencode) -> None:
config = client.config.get()
assert_matches_type(Config, config, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_get(self, client: Opencode) -> None:
response = client.config.with_raw_response.get()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = response.parse()
assert_matches_type(Config, config, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_get(self, client: Opencode) -> None:
with client.config.with_streaming_response.get() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = response.parse()
assert_matches_type(Config, config, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_method_providers(self, client: Opencode) -> None:
config = client.config.providers()
assert_matches_type(ConfigProvidersResponse, config, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_providers(self, client: Opencode) -> None:
response = client.config.with_raw_response.providers()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = response.parse()
assert_matches_type(ConfigProvidersResponse, config, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_providers(self, client: Opencode) -> None:
with client.config.with_streaming_response.providers() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = response.parse()
assert_matches_type(ConfigProvidersResponse, config, path=["response"])
assert cast(Any, response.is_closed) is True
class TestAsyncConfig:
parametrize = pytest.mark.parametrize(
"async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
)
@pytest.mark.skip()
@parametrize
async def test_method_get(self, async_client: AsyncOpencode) -> None:
config = await async_client.config.get()
assert_matches_type(Config, config, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_get(self, async_client: AsyncOpencode) -> None:
response = await async_client.config.with_raw_response.get()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = await response.parse()
assert_matches_type(Config, config, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_get(self, async_client: AsyncOpencode) -> None:
async with async_client.config.with_streaming_response.get() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = await response.parse()
assert_matches_type(Config, config, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_method_providers(self, async_client: AsyncOpencode) -> None:
config = await async_client.config.providers()
assert_matches_type(ConfigProvidersResponse, config, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_providers(self, async_client: AsyncOpencode) -> None:
response = await async_client.config.with_raw_response.providers()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = await response.parse()
assert_matches_type(ConfigProvidersResponse, config, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_providers(self, async_client: AsyncOpencode) -> None:
async with async_client.config.with_streaming_response.providers() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
config = await response.parse()
assert_matches_type(ConfigProvidersResponse, config, path=["response"])
assert cast(Any, response.is_closed) is True

80
tests/api_resources/test_event.py Normal file
View file

@ -0,0 +1,80 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from opencode import Opencode, AsyncOpencode
from tests.utils import assert_matches_type
from opencode.types import EventListResponse
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestEvent:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@pytest.mark.skip()
@parametrize
def test_method_list(self, client: Opencode) -> None:
event = client.event.list()
assert_matches_type(EventListResponse, event, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_list(self, client: Opencode) -> None:
response = client.event.with_raw_response.list()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
event = response.parse()
assert_matches_type(EventListResponse, event, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_list(self, client: Opencode) -> None:
with client.event.with_streaming_response.list() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
event = response.parse()
assert_matches_type(EventListResponse, event, path=["response"])
assert cast(Any, response.is_closed) is True
class TestAsyncEvent:
parametrize = pytest.mark.parametrize(
"async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
)
@pytest.mark.skip()
@parametrize
async def test_method_list(self, async_client: AsyncOpencode) -> None:
event = await async_client.event.list()
assert_matches_type(EventListResponse, event, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_list(self, async_client: AsyncOpencode) -> None:
response = await async_client.event.with_raw_response.list()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
event = await response.parse()
assert_matches_type(EventListResponse, event, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_list(self, async_client: AsyncOpencode) -> None:
async with async_client.event.with_streaming_response.list() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
event = await response.parse()
assert_matches_type(EventListResponse, event, path=["response"])
assert cast(Any, response.is_closed) is True

92
tests/api_resources/test_file.py Normal file
View file

@ -0,0 +1,92 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from opencode import Opencode, AsyncOpencode
from tests.utils import assert_matches_type
from opencode.types import FileSearchResponse
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestFile:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@pytest.mark.skip()
@parametrize
def test_method_search(self, client: Opencode) -> None:
file = client.file.search(
query="query",
)
assert_matches_type(FileSearchResponse, file, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_search(self, client: Opencode) -> None:
response = client.file.with_raw_response.search(
query="query",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
file = response.parse()
assert_matches_type(FileSearchResponse, file, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_search(self, client: Opencode) -> None:
with client.file.with_streaming_response.search(
query="query",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
file = response.parse()
assert_matches_type(FileSearchResponse, file, path=["response"])
assert cast(Any, response.is_closed) is True
class TestAsyncFile:
parametrize = pytest.mark.parametrize(
"async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
)
@pytest.mark.skip()
@parametrize
async def test_method_search(self, async_client: AsyncOpencode) -> None:
file = await async_client.file.search(
query="query",
)
assert_matches_type(FileSearchResponse, file, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_search(self, async_client: AsyncOpencode) -> None:
response = await async_client.file.with_raw_response.search(
query="query",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
file = await response.parse()
assert_matches_type(FileSearchResponse, file, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_search(self, async_client: AsyncOpencode) -> None:
async with async_client.file.with_streaming_response.search(
query="query",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
file = await response.parse()
assert_matches_type(FileSearchResponse, file, path=["response"])
assert cast(Any, response.is_closed) is True

921
tests/api_resources/test_session.py Normal file
View file

@ -0,0 +1,921 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from opencode import Opencode, AsyncOpencode
from tests.utils import assert_matches_type
from opencode.types import (
Message,
Session,
SessionInitResponse,
SessionListResponse,
SessionAbortResponse,
SessionDeleteResponse,
SessionMessagesResponse,
SessionSummarizeResponse,
)
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestSession:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@pytest.mark.skip()
@parametrize
def test_method_create(self, client: Opencode) -> None:
session = client.session.create()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_create(self, client: Opencode) -> None:
response = client.session.with_raw_response.create()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_create(self, client: Opencode) -> None:
with client.session.with_streaming_response.create() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Session, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_method_list(self, client: Opencode) -> None:
session = client.session.list()
assert_matches_type(SessionListResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_list(self, client: Opencode) -> None:
response = client.session.with_raw_response.list()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionListResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_list(self, client: Opencode) -> None:
with client.session.with_streaming_response.list() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionListResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_method_delete(self, client: Opencode) -> None:
session = client.session.delete(
"id",
)
assert_matches_type(SessionDeleteResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_delete(self, client: Opencode) -> None:
response = client.session.with_raw_response.delete(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionDeleteResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_delete(self, client: Opencode) -> None:
with client.session.with_streaming_response.delete(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionDeleteResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_delete(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.delete(
"",
)
@pytest.mark.skip()
@parametrize
def test_method_abort(self, client: Opencode) -> None:
session = client.session.abort(
"id",
)
assert_matches_type(SessionAbortResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_abort(self, client: Opencode) -> None:
response = client.session.with_raw_response.abort(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionAbortResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_abort(self, client: Opencode) -> None:
with client.session.with_streaming_response.abort(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionAbortResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_abort(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.abort(
"",
)
@pytest.mark.skip()
@parametrize
def test_method_chat(self, client: Opencode) -> None:
session = client.session.chat(
id="id",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
)
assert_matches_type(Message, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_chat(self, client: Opencode) -> None:
response = client.session.with_raw_response.chat(
id="id",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Message, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_chat(self, client: Opencode) -> None:
with client.session.with_streaming_response.chat(
id="id",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Message, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_chat(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.chat(
id="",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
)
@pytest.mark.skip()
@parametrize
def test_method_init(self, client: Opencode) -> None:
session = client.session.init(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert_matches_type(SessionInitResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_init(self, client: Opencode) -> None:
response = client.session.with_raw_response.init(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionInitResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_init(self, client: Opencode) -> None:
with client.session.with_streaming_response.init(
id="id",
model_id="modelID",
provider_id="providerID",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionInitResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_init(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.init(
id="",
model_id="modelID",
provider_id="providerID",
)
@pytest.mark.skip()
@parametrize
def test_method_messages(self, client: Opencode) -> None:
session = client.session.messages(
"id",
)
assert_matches_type(SessionMessagesResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_messages(self, client: Opencode) -> None:
response = client.session.with_raw_response.messages(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionMessagesResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_messages(self, client: Opencode) -> None:
with client.session.with_streaming_response.messages(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionMessagesResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_messages(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.messages(
"",
)
@pytest.mark.skip()
@parametrize
def test_method_share(self, client: Opencode) -> None:
session = client.session.share(
"id",
)
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_share(self, client: Opencode) -> None:
response = client.session.with_raw_response.share(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_share(self, client: Opencode) -> None:
with client.session.with_streaming_response.share(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Session, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_share(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.share(
"",
)
@pytest.mark.skip()
@parametrize
def test_method_summarize(self, client: Opencode) -> None:
session = client.session.summarize(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert_matches_type(SessionSummarizeResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_summarize(self, client: Opencode) -> None:
response = client.session.with_raw_response.summarize(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionSummarizeResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_summarize(self, client: Opencode) -> None:
with client.session.with_streaming_response.summarize(
id="id",
model_id="modelID",
provider_id="providerID",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(SessionSummarizeResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_summarize(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.summarize(
id="",
model_id="modelID",
provider_id="providerID",
)
@pytest.mark.skip()
@parametrize
def test_method_unshare(self, client: Opencode) -> None:
session = client.session.unshare(
"id",
)
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_raw_response_unshare(self, client: Opencode) -> None:
response = client.session.with_raw_response.unshare(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
def test_streaming_response_unshare(self, client: Opencode) -> None:
with client.session.with_streaming_response.unshare(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = response.parse()
assert_matches_type(Session, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
def test_path_params_unshare(self, client: Opencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
client.session.with_raw_response.unshare(
"",
)
class TestAsyncSession:
parametrize = pytest.mark.parametrize(
"async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"]
)
@pytest.mark.skip()
@parametrize
async def test_method_create(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.create()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_create(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.create()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_create(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.create() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Session, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_method_list(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.list()
assert_matches_type(SessionListResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_list(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.list()
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionListResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_list(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.list() as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionListResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_method_delete(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.delete(
"id",
)
assert_matches_type(SessionDeleteResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_delete(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.delete(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionDeleteResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_delete(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.delete(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionDeleteResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_delete(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.delete(
"",
)
@pytest.mark.skip()
@parametrize
async def test_method_abort(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.abort(
"id",
)
assert_matches_type(SessionAbortResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_abort(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.abort(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionAbortResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_abort(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.abort(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionAbortResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_abort(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.abort(
"",
)
@pytest.mark.skip()
@parametrize
async def test_method_chat(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.chat(
id="id",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
)
assert_matches_type(Message, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_chat(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.chat(
id="id",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Message, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_chat(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.chat(
id="id",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Message, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_chat(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.chat(
id="",
model_id="modelID",
parts=[
{
"text": "text",
"type": "text",
}
],
provider_id="providerID",
session_id="sessionID",
)
@pytest.mark.skip()
@parametrize
async def test_method_init(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.init(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert_matches_type(SessionInitResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_init(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.init(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionInitResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_init(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.init(
id="id",
model_id="modelID",
provider_id="providerID",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionInitResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_init(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.init(
id="",
model_id="modelID",
provider_id="providerID",
)
@pytest.mark.skip()
@parametrize
async def test_method_messages(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.messages(
"id",
)
assert_matches_type(SessionMessagesResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_messages(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.messages(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionMessagesResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_messages(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.messages(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionMessagesResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_messages(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.messages(
"",
)
@pytest.mark.skip()
@parametrize
async def test_method_share(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.share(
"id",
)
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_share(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.share(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_share(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.share(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Session, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_share(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.share(
"",
)
@pytest.mark.skip()
@parametrize
async def test_method_summarize(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.summarize(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert_matches_type(SessionSummarizeResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_summarize(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.summarize(
id="id",
model_id="modelID",
provider_id="providerID",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionSummarizeResponse, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_summarize(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.summarize(
id="id",
model_id="modelID",
provider_id="providerID",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(SessionSummarizeResponse, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_summarize(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.summarize(
id="",
model_id="modelID",
provider_id="providerID",
)
@pytest.mark.skip()
@parametrize
async def test_method_unshare(self, async_client: AsyncOpencode) -> None:
session = await async_client.session.unshare(
"id",
)
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_raw_response_unshare(self, async_client: AsyncOpencode) -> None:
response = await async_client.session.with_raw_response.unshare(
"id",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Session, session, path=["response"])
@pytest.mark.skip()
@parametrize
async def test_streaming_response_unshare(self, async_client: AsyncOpencode) -> None:
async with async_client.session.with_streaming_response.unshare(
"id",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
session = await response.parse()
assert_matches_type(Session, session, path=["response"])
assert cast(Any, response.is_closed) is True
@pytest.mark.skip()
@parametrize
async def test_path_params_unshare(self, async_client: AsyncOpencode) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `id` but received ''"):
await async_client.session.with_raw_response.unshare(
"",
)

80
tests/conftest.py Normal file
View file

@ -0,0 +1,80 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
import logging
from typing import TYPE_CHECKING, Iterator, AsyncIterator
import httpx
import pytest
from pytest_asyncio import is_async_test
from opencode import Opencode, AsyncOpencode, DefaultAioHttpClient
from opencode._utils import is_dict
if TYPE_CHECKING:
from _pytest.fixtures import FixtureRequest # pyright: ignore[reportPrivateImportUsage]
pytest.register_assert_rewrite("tests.utils")
logging.getLogger("opencode").setLevel(logging.DEBUG)
# automatically add `pytest.mark.asyncio()` to all of our async tests
# so we don't have to add that boilerplate everywhere
def pytest_collection_modifyitems(items: list[pytest.Function]) -> None:
pytest_asyncio_tests = (item for item in items if is_async_test(item))
session_scope_marker = pytest.mark.asyncio(loop_scope="session")
for async_test in pytest_asyncio_tests:
async_test.add_marker(session_scope_marker, append=False)
# We skip tests that use both the aiohttp client and respx_mock as respx_mock
# doesn't support custom transports.
for item in items:
if "async_client" not in item.fixturenames or "respx_mock" not in item.fixturenames:
continue
if not hasattr(item, "callspec"):
continue
async_client_param = item.callspec.params.get("async_client")
if is_dict(async_client_param) and async_client_param.get("http_client") == "aiohttp":
item.add_marker(pytest.mark.skip(reason="aiohttp client is not compatible with respx_mock"))
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
@pytest.fixture(scope="session")
def client(request: FixtureRequest) -> Iterator[Opencode]:
strict = getattr(request, "param", True)
if not isinstance(strict, bool):
raise TypeError(f"Unexpected fixture parameter type {type(strict)}, expected {bool}")
with Opencode(base_url=base_url, _strict_response_validation=strict) as client:
yield client
@pytest.fixture(scope="session")
async def async_client(request: FixtureRequest) -> AsyncIterator[AsyncOpencode]:
param = getattr(request, "param", True)
# defaults
strict = True
http_client: None | httpx.AsyncClient = None
if isinstance(param, bool):
strict = param
elif is_dict(param):
strict = param.get("strict", True)
assert isinstance(strict, bool)
http_client_type = param.get("http_client", "httpx")
if http_client_type == "aiohttp":
http_client = DefaultAioHttpClient()
else:
raise TypeError(f"Unexpected fixture parameter type {type(param)}, expected bool or dict")
async with AsyncOpencode(base_url=base_url, _strict_response_validation=strict, http_client=http_client) as client:
yield client

1
tests/sample_file.txt Normal file
View file

@ -0,0 +1 @@
Hello, world!

1637
tests/test_client.py Normal file
View file

File diff suppressed because it is too large Load diff

58
tests/test_deepcopy.py Normal file
View file

@ -0,0 +1,58 @@
from opencode._utils import deepcopy_minimal
def assert_different_identities(obj1: object, obj2: object) -> None:
assert obj1 == obj2
assert id(obj1) != id(obj2)
def test_simple_dict() -> None:
obj1 = {"foo": "bar"}
obj2 = deepcopy_minimal(obj1)
assert_different_identities(obj1, obj2)
def test_nested_dict() -> None:
obj1 = {"foo": {"bar": True}}
obj2 = deepcopy_minimal(obj1)
assert_different_identities(obj1, obj2)
assert_different_identities(obj1["foo"], obj2["foo"])
def test_complex_nested_dict() -> None:
obj1 = {"foo": {"bar": [{"hello": "world"}]}}
obj2 = deepcopy_minimal(obj1)
assert_different_identities(obj1, obj2)
assert_different_identities(obj1["foo"], obj2["foo"])
assert_different_identities(obj1["foo"]["bar"], obj2["foo"]["bar"])
assert_different_identities(obj1["foo"]["bar"][0], obj2["foo"]["bar"][0])
def test_simple_list() -> None:
obj1 = ["a", "b", "c"]
obj2 = deepcopy_minimal(obj1)
assert_different_identities(obj1, obj2)
def test_nested_list() -> None:
obj1 = ["a", [1, 2, 3]]
obj2 = deepcopy_minimal(obj1)
assert_different_identities(obj1, obj2)
assert_different_identities(obj1[1], obj2[1])
class MyObject: ...
def test_ignores_other_types() -> None:
# custom classes
my_obj = MyObject()
obj1 = {"foo": my_obj}
obj2 = deepcopy_minimal(obj1)
assert_different_identities(obj1, obj2)
assert obj1["foo"] is my_obj
# tuples
obj3 = ("a", "b")
obj4 = deepcopy_minimal(obj3)
assert obj3 is obj4

64
tests/test_extract_files.py Normal file
View file

@ -0,0 +1,64 @@
from __future__ import annotations
from typing import Sequence
import pytest
from opencode._types import FileTypes
from opencode._utils import extract_files
def test_removes_files_from_input() -> None:
query = {"foo": "bar"}
assert extract_files(query, paths=[]) == []
assert query == {"foo": "bar"}
query2 = {"foo": b"Bar", "hello": "world"}
assert extract_files(query2, paths=[["foo"]]) == [("foo", b"Bar")]
assert query2 == {"hello": "world"}
query3 = {"foo": {"foo": {"bar": b"Bar"}}, "hello": "world"}
assert extract_files(query3, paths=[["foo", "foo", "bar"]]) == [("foo[foo][bar]", b"Bar")]
assert query3 == {"foo": {"foo": {}}, "hello": "world"}
query4 = {"foo": {"bar": b"Bar", "baz": "foo"}, "hello": "world"}
assert extract_files(query4, paths=[["foo", "bar"]]) == [("foo[bar]", b"Bar")]
assert query4 == {"hello": "world", "foo": {"baz": "foo"}}
def test_multiple_files() -> None:
query = {"documents": [{"file": b"My first file"}, {"file": b"My second file"}]}
assert extract_files(query, paths=[["documents", "<array>", "file"]]) == [
("documents[][file]", b"My first file"),
("documents[][file]", b"My second file"),
]
assert query == {"documents": [{}, {}]}
@pytest.mark.parametrize(
"query,paths,expected",
[
[
{"foo": {"bar": "baz"}},
[["foo", "<array>", "bar"]],
[],
],
[
{"foo": ["bar", "baz"]},
[["foo", "bar"]],
[],
],
[
{"foo": {"bar": "baz"}},
[["foo", "foo"]],
[],
],
],
ids=["dict expecting array", "array expecting dict", "unknown keys"],
)
def test_ignores_incorrect_paths(
query: dict[str, object],
paths: Sequence[Sequence[str]],
expected: list[tuple[str, FileTypes]],
) -> None:
assert extract_files(query, paths=paths) == expected

51
tests/test_files.py Normal file
View file

@ -0,0 +1,51 @@
from pathlib import Path
import anyio
import pytest
from dirty_equals import IsDict, IsList, IsBytes, IsTuple
from opencode._files import to_httpx_files, async_to_httpx_files
readme_path = Path(__file__).parent.parent.joinpath("README.md")
def test_pathlib_includes_file_name() -> None:
result = to_httpx_files({"file": readme_path})
print(result)
assert result == IsDict({"file": IsTuple("README.md", IsBytes())})
def test_tuple_input() -> None:
result = to_httpx_files([("file", readme_path)])
print(result)
assert result == IsList(IsTuple("file", IsTuple("README.md", IsBytes())))
@pytest.mark.asyncio
async def test_async_pathlib_includes_file_name() -> None:
result = await async_to_httpx_files({"file": readme_path})
print(result)
assert result == IsDict({"file": IsTuple("README.md", IsBytes())})
@pytest.mark.asyncio
async def test_async_supports_anyio_path() -> None:
result = await async_to_httpx_files({"file": anyio.Path(readme_path)})
print(result)
assert result == IsDict({"file": IsTuple("README.md", IsBytes())})
@pytest.mark.asyncio
async def test_async_tuple_input() -> None:
result = await async_to_httpx_files([("file", readme_path)])
print(result)
assert result == IsList(IsTuple("file", IsTuple("README.md", IsBytes())))
def test_string_not_allowed() -> None:
with pytest.raises(TypeError, match="Expected file types input to be a FileContent type or to be a tuple"):
to_httpx_files(
{
"file": "foo", # type: ignore
}
)

891
tests/test_models.py Normal file
View file

@ -0,0 +1,891 @@
import json
from typing import Any, Dict, List, Union, Optional, cast
from datetime import datetime, timezone
from typing_extensions import Literal, Annotated, TypeAliasType
import pytest
import pydantic
from pydantic import Field
from opencode._utils import PropertyInfo
from opencode._compat import PYDANTIC_V2, parse_obj, model_dump, model_json
from opencode._models import BaseModel, construct_type
class BasicModel(BaseModel):
foo: str
@pytest.mark.parametrize("value", ["hello", 1], ids=["correct type", "mismatched"])
def test_basic(value: object) -> None:
m = BasicModel.construct(foo=value)
assert m.foo == value
def test_directly_nested_model() -> None:
class NestedModel(BaseModel):
nested: BasicModel
m = NestedModel.construct(nested={"foo": "Foo!"})
assert m.nested.foo == "Foo!"
# mismatched types
m = NestedModel.construct(nested="hello!")
assert cast(Any, m.nested) == "hello!"
def test_optional_nested_model() -> None:
class NestedModel(BaseModel):
nested: Optional[BasicModel]
m1 = NestedModel.construct(nested=None)
assert m1.nested is None
m2 = NestedModel.construct(nested={"foo": "bar"})
assert m2.nested is not None
assert m2.nested.foo == "bar"
# mismatched types
m3 = NestedModel.construct(nested={"foo"})
assert isinstance(cast(Any, m3.nested), set)
assert cast(Any, m3.nested) == {"foo"}
def test_list_nested_model() -> None:
class NestedModel(BaseModel):
nested: List[BasicModel]
m = NestedModel.construct(nested=[{"foo": "bar"}, {"foo": "2"}])
assert m.nested is not None
assert isinstance(m.nested, list)
assert len(m.nested) == 2
assert m.nested[0].foo == "bar"
assert m.nested[1].foo == "2"
# mismatched types
m = NestedModel.construct(nested=True)
assert cast(Any, m.nested) is True
m = NestedModel.construct(nested=[False])
assert cast(Any, m.nested) == [False]
def test_optional_list_nested_model() -> None:
class NestedModel(BaseModel):
nested: Optional[List[BasicModel]]
m1 = NestedModel.construct(nested=[{"foo": "bar"}, {"foo": "2"}])
assert m1.nested is not None
assert isinstance(m1.nested, list)
assert len(m1.nested) == 2
assert m1.nested[0].foo == "bar"
assert m1.nested[1].foo == "2"
m2 = NestedModel.construct(nested=None)
assert m2.nested is None
# mismatched types
m3 = NestedModel.construct(nested={1})
assert cast(Any, m3.nested) == {1}
m4 = NestedModel.construct(nested=[False])
assert cast(Any, m4.nested) == [False]
def test_list_optional_items_nested_model() -> None:
class NestedModel(BaseModel):
nested: List[Optional[BasicModel]]
m = NestedModel.construct(nested=[None, {"foo": "bar"}])
assert m.nested is not None
assert isinstance(m.nested, list)
assert len(m.nested) == 2
assert m.nested[0] is None
assert m.nested[1] is not None
assert m.nested[1].foo == "bar"
# mismatched types
m3 = NestedModel.construct(nested="foo")
assert cast(Any, m3.nested) == "foo"
m4 = NestedModel.construct(nested=[False])
assert cast(Any, m4.nested) == [False]
def test_list_mismatched_type() -> None:
class NestedModel(BaseModel):
nested: List[str]
m = NestedModel.construct(nested=False)
assert cast(Any, m.nested) is False
def test_raw_dictionary() -> None:
class NestedModel(BaseModel):
nested: Dict[str, str]
m = NestedModel.construct(nested={"hello": "world"})
assert m.nested == {"hello": "world"}
# mismatched types
m = NestedModel.construct(nested=False)
assert cast(Any, m.nested) is False
def test_nested_dictionary_model() -> None:
class NestedModel(BaseModel):
nested: Dict[str, BasicModel]
m = NestedModel.construct(nested={"hello": {"foo": "bar"}})
assert isinstance(m.nested, dict)
assert m.nested["hello"].foo == "bar"
# mismatched types
m = NestedModel.construct(nested={"hello": False})
assert cast(Any, m.nested["hello"]) is False
def test_unknown_fields() -> None:
m1 = BasicModel.construct(foo="foo", unknown=1)
assert m1.foo == "foo"
assert cast(Any, m1).unknown == 1
m2 = BasicModel.construct(foo="foo", unknown={"foo_bar": True})
assert m2.foo == "foo"
assert cast(Any, m2).unknown == {"foo_bar": True}
assert model_dump(m2) == {"foo": "foo", "unknown": {"foo_bar": True}}
def test_strict_validation_unknown_fields() -> None:
class Model(BaseModel):
foo: str
model = parse_obj(Model, dict(foo="hello!", user="Robert"))
assert model.foo == "hello!"
assert cast(Any, model).user == "Robert"
assert model_dump(model) == {"foo": "hello!", "user": "Robert"}
def test_aliases() -> None:
class Model(BaseModel):
my_field: int = Field(alias="myField")
m = Model.construct(myField=1)
assert m.my_field == 1
# mismatched types
m = Model.construct(myField={"hello": False})
assert cast(Any, m.my_field) == {"hello": False}
def test_repr() -> None:
model = BasicModel(foo="bar")
assert str(model) == "BasicModel(foo='bar')"
assert repr(model) == "BasicModel(foo='bar')"
def test_repr_nested_model() -> None:
class Child(BaseModel):
name: str
age: int
class Parent(BaseModel):
name: str
child: Child
model = Parent(name="Robert", child=Child(name="Foo", age=5))
assert str(model) == "Parent(name='Robert', child=Child(name='Foo', age=5))"
assert repr(model) == "Parent(name='Robert', child=Child(name='Foo', age=5))"
def test_optional_list() -> None:
class Submodel(BaseModel):
name: str
class Model(BaseModel):
items: Optional[List[Submodel]]
m = Model.construct(items=None)
assert m.items is None
m = Model.construct(items=[])
assert m.items == []
m = Model.construct(items=[{"name": "Robert"}])
assert m.items is not None
assert len(m.items) == 1
assert m.items[0].name == "Robert"
def test_nested_union_of_models() -> None:
class Submodel1(BaseModel):
bar: bool
class Submodel2(BaseModel):
thing: str
class Model(BaseModel):
foo: Union[Submodel1, Submodel2]
m = Model.construct(foo={"thing": "hello"})
assert isinstance(m.foo, Submodel2)
assert m.foo.thing == "hello"
def test_nested_union_of_mixed_types() -> None:
class Submodel1(BaseModel):
bar: bool
class Model(BaseModel):
foo: Union[Submodel1, Literal[True], Literal["CARD_HOLDER"]]
m = Model.construct(foo=True)
assert m.foo is True
m = Model.construct(foo="CARD_HOLDER")
assert m.foo == "CARD_HOLDER"
m = Model.construct(foo={"bar": False})
assert isinstance(m.foo, Submodel1)
assert m.foo.bar is False
def test_nested_union_multiple_variants() -> None:
class Submodel1(BaseModel):
bar: bool
class Submodel2(BaseModel):
thing: str
class Submodel3(BaseModel):
foo: int
class Model(BaseModel):
foo: Union[Submodel1, Submodel2, None, Submodel3]
m = Model.construct(foo={"thing": "hello"})
assert isinstance(m.foo, Submodel2)
assert m.foo.thing == "hello"
m = Model.construct(foo=None)
assert m.foo is None
m = Model.construct()
assert m.foo is None
m = Model.construct(foo={"foo": "1"})
assert isinstance(m.foo, Submodel3)
assert m.foo.foo == 1
def test_nested_union_invalid_data() -> None:
class Submodel1(BaseModel):
level: int
class Submodel2(BaseModel):
name: str
class Model(BaseModel):
foo: Union[Submodel1, Submodel2]
m = Model.construct(foo=True)
assert cast(bool, m.foo) is True
m = Model.construct(foo={"name": 3})
if PYDANTIC_V2:
assert isinstance(m.foo, Submodel1)
assert m.foo.name == 3 # type: ignore
else:
assert isinstance(m.foo, Submodel2)
assert m.foo.name == "3"
def test_list_of_unions() -> None:
class Submodel1(BaseModel):
level: int
class Submodel2(BaseModel):
name: str
class Model(BaseModel):
items: List[Union[Submodel1, Submodel2]]
m = Model.construct(items=[{"level": 1}, {"name": "Robert"}])
assert len(m.items) == 2
assert isinstance(m.items[0], Submodel1)
assert m.items[0].level == 1
assert isinstance(m.items[1], Submodel2)
assert m.items[1].name == "Robert"
m = Model.construct(items=[{"level": -1}, 156])
assert len(m.items) == 2
assert isinstance(m.items[0], Submodel1)
assert m.items[0].level == -1
assert cast(Any, m.items[1]) == 156
def test_union_of_lists() -> None:
class SubModel1(BaseModel):
level: int
class SubModel2(BaseModel):
name: str
class Model(BaseModel):
items: Union[List[SubModel1], List[SubModel2]]
# with one valid entry
m = Model.construct(items=[{"name": "Robert"}])
assert len(m.items) == 1
assert isinstance(m.items[0], SubModel2)
assert m.items[0].name == "Robert"
# with two entries pointing to different types
m = Model.construct(items=[{"level": 1}, {"name": "Robert"}])
assert len(m.items) == 2
assert isinstance(m.items[0], SubModel1)
assert m.items[0].level == 1
assert isinstance(m.items[1], SubModel1)
assert cast(Any, m.items[1]).name == "Robert"
# with two entries pointing to *completely* different types
m = Model.construct(items=[{"level": -1}, 156])
assert len(m.items) == 2
assert isinstance(m.items[0], SubModel1)
assert m.items[0].level == -1
assert cast(Any, m.items[1]) == 156
def test_dict_of_union() -> None:
class SubModel1(BaseModel):
name: str
class SubModel2(BaseModel):
foo: str
class Model(BaseModel):
data: Dict[str, Union[SubModel1, SubModel2]]
m = Model.construct(data={"hello": {"name": "there"}, "foo": {"foo": "bar"}})
assert len(list(m.data.keys())) == 2
assert isinstance(m.data["hello"], SubModel1)
assert m.data["hello"].name == "there"
assert isinstance(m.data["foo"], SubModel2)
assert m.data["foo"].foo == "bar"
# TODO: test mismatched type
def test_double_nested_union() -> None:
class SubModel1(BaseModel):
name: str
class SubModel2(BaseModel):
bar: str
class Model(BaseModel):
data: Dict[str, List[Union[SubModel1, SubModel2]]]
m = Model.construct(data={"foo": [{"bar": "baz"}, {"name": "Robert"}]})
assert len(m.data["foo"]) == 2
entry1 = m.data["foo"][0]
assert isinstance(entry1, SubModel2)
assert entry1.bar == "baz"
entry2 = m.data["foo"][1]
assert isinstance(entry2, SubModel1)
assert entry2.name == "Robert"
# TODO: test mismatched type
def test_union_of_dict() -> None:
class SubModel1(BaseModel):
name: str
class SubModel2(BaseModel):
foo: str
class Model(BaseModel):
data: Union[Dict[str, SubModel1], Dict[str, SubModel2]]
m = Model.construct(data={"hello": {"name": "there"}, "foo": {"foo": "bar"}})
assert len(list(m.data.keys())) == 2
assert isinstance(m.data["hello"], SubModel1)
assert m.data["hello"].name == "there"
assert isinstance(m.data["foo"], SubModel1)
assert cast(Any, m.data["foo"]).foo == "bar"
def test_iso8601_datetime() -> None:
class Model(BaseModel):
created_at: datetime
expected = datetime(2019, 12, 27, 18, 11, 19, 117000, tzinfo=timezone.utc)
if PYDANTIC_V2:
expected_json = '{"created_at":"2019-12-27T18:11:19.117000Z"}'
else:
expected_json = '{"created_at": "2019-12-27T18:11:19.117000+00:00"}'
model = Model.construct(created_at="2019-12-27T18:11:19.117Z")
assert model.created_at == expected
assert model_json(model) == expected_json
model = parse_obj(Model, dict(created_at="2019-12-27T18:11:19.117Z"))
assert model.created_at == expected
assert model_json(model) == expected_json
def test_does_not_coerce_int() -> None:
class Model(BaseModel):
bar: int
assert Model.construct(bar=1).bar == 1
assert Model.construct(bar=10.9).bar == 10.9
assert Model.construct(bar="19").bar == "19" # type: ignore[comparison-overlap]
assert Model.construct(bar=False).bar is False
def test_int_to_float_safe_conversion() -> None:
class Model(BaseModel):
float_field: float
m = Model.construct(float_field=10)
assert m.float_field == 10.0
assert isinstance(m.float_field, float)
m = Model.construct(float_field=10.12)
assert m.float_field == 10.12
assert isinstance(m.float_field, float)
# number too big
m = Model.construct(float_field=2**53 + 1)
assert m.float_field == 2**53 + 1
assert isinstance(m.float_field, int)
def test_deprecated_alias() -> None:
class Model(BaseModel):
resource_id: str = Field(alias="model_id")
@property
def model_id(self) -> str:
return self.resource_id
m = Model.construct(model_id="id")
assert m.model_id == "id"
assert m.resource_id == "id"
assert m.resource_id is m.model_id
m = parse_obj(Model, {"model_id": "id"})
assert m.model_id == "id"
assert m.resource_id == "id"
assert m.resource_id is m.model_id
def test_omitted_fields() -> None:
class Model(BaseModel):
resource_id: Optional[str] = None
m = Model.construct()
assert m.resource_id is None
assert "resource_id" not in m.model_fields_set
m = Model.construct(resource_id=None)
assert m.resource_id is None
assert "resource_id" in m.model_fields_set
m = Model.construct(resource_id="foo")
assert m.resource_id == "foo"
assert "resource_id" in m.model_fields_set
def test_to_dict() -> None:
class Model(BaseModel):
foo: Optional[str] = Field(alias="FOO", default=None)
m = Model(FOO="hello")
assert m.to_dict() == {"FOO": "hello"}
assert m.to_dict(use_api_names=False) == {"foo": "hello"}
m2 = Model()
assert m2.to_dict() == {}
assert m2.to_dict(exclude_unset=False) == {"FOO": None}
assert m2.to_dict(exclude_unset=False, exclude_none=True) == {}
assert m2.to_dict(exclude_unset=False, exclude_defaults=True) == {}
m3 = Model(FOO=None)
assert m3.to_dict() == {"FOO": None}
assert m3.to_dict(exclude_none=True) == {}
assert m3.to_dict(exclude_defaults=True) == {}
class Model2(BaseModel):
created_at: datetime
time_str = "2024-03-21T11:39:01.275859"
m4 = Model2.construct(created_at=time_str)
assert m4.to_dict(mode="python") == {"created_at": datetime.fromisoformat(time_str)}
assert m4.to_dict(mode="json") == {"created_at": time_str}
if not PYDANTIC_V2:
with pytest.raises(ValueError, match="warnings is only supported in Pydantic v2"):
m.to_dict(warnings=False)
def test_forwards_compat_model_dump_method() -> None:
class Model(BaseModel):
foo: Optional[str] = Field(alias="FOO", default=None)
m = Model(FOO="hello")
assert m.model_dump() == {"foo": "hello"}
assert m.model_dump(include={"bar"}) == {}
assert m.model_dump(exclude={"foo"}) == {}
assert m.model_dump(by_alias=True) == {"FOO": "hello"}
m2 = Model()
assert m2.model_dump() == {"foo": None}
assert m2.model_dump(exclude_unset=True) == {}
assert m2.model_dump(exclude_none=True) == {}
assert m2.model_dump(exclude_defaults=True) == {}
m3 = Model(FOO=None)
assert m3.model_dump() == {"foo": None}
assert m3.model_dump(exclude_none=True) == {}
if not PYDANTIC_V2:
with pytest.raises(ValueError, match="round_trip is only supported in Pydantic v2"):
m.model_dump(round_trip=True)
with pytest.raises(ValueError, match="warnings is only supported in Pydantic v2"):
m.model_dump(warnings=False)
def test_compat_method_no_error_for_warnings() -> None:
class Model(BaseModel):
foo: Optional[str]
m = Model(foo="hello")
assert isinstance(model_dump(m, warnings=False), dict)
def test_to_json() -> None:
class Model(BaseModel):
foo: Optional[str] = Field(alias="FOO", default=None)
m = Model(FOO="hello")
assert json.loads(m.to_json()) == {"FOO": "hello"}
assert json.loads(m.to_json(use_api_names=False)) == {"foo": "hello"}
if PYDANTIC_V2:
assert m.to_json(indent=None) == '{"FOO":"hello"}'
else:
assert m.to_json(indent=None) == '{"FOO": "hello"}'
m2 = Model()
assert json.loads(m2.to_json()) == {}
assert json.loads(m2.to_json(exclude_unset=False)) == {"FOO": None}
assert json.loads(m2.to_json(exclude_unset=False, exclude_none=True)) == {}
assert json.loads(m2.to_json(exclude_unset=False, exclude_defaults=True)) == {}
m3 = Model(FOO=None)
assert json.loads(m3.to_json()) == {"FOO": None}
assert json.loads(m3.to_json(exclude_none=True)) == {}
if not PYDANTIC_V2:
with pytest.raises(ValueError, match="warnings is only supported in Pydantic v2"):
m.to_json(warnings=False)
def test_forwards_compat_model_dump_json_method() -> None:
class Model(BaseModel):
foo: Optional[str] = Field(alias="FOO", default=None)
m = Model(FOO="hello")
assert json.loads(m.model_dump_json()) == {"foo": "hello"}
assert json.loads(m.model_dump_json(include={"bar"})) == {}
assert json.loads(m.model_dump_json(include={"foo"})) == {"foo": "hello"}
assert json.loads(m.model_dump_json(by_alias=True)) == {"FOO": "hello"}
assert m.model_dump_json(indent=2) == '{\n "foo": "hello"\n}'
m2 = Model()
assert json.loads(m2.model_dump_json()) == {"foo": None}
assert json.loads(m2.model_dump_json(exclude_unset=True)) == {}
assert json.loads(m2.model_dump_json(exclude_none=True)) == {}
assert json.loads(m2.model_dump_json(exclude_defaults=True)) == {}
m3 = Model(FOO=None)
assert json.loads(m3.model_dump_json()) == {"foo": None}
assert json.loads(m3.model_dump_json(exclude_none=True)) == {}
if not PYDANTIC_V2:
with pytest.raises(ValueError, match="round_trip is only supported in Pydantic v2"):
m.model_dump_json(round_trip=True)
with pytest.raises(ValueError, match="warnings is only supported in Pydantic v2"):
m.model_dump_json(warnings=False)
def test_type_compat() -> None:
# our model type can be assigned to Pydantic's model type
def takes_pydantic(model: pydantic.BaseModel) -> None: # noqa: ARG001
...
class OurModel(BaseModel):
foo: Optional[str] = None
takes_pydantic(OurModel())
def test_annotated_types() -> None:
class Model(BaseModel):
value: str
m = construct_type(
value={"value": "foo"},
type_=cast(Any, Annotated[Model, "random metadata"]),
)
assert isinstance(m, Model)
assert m.value == "foo"
def test_discriminated_unions_invalid_data() -> None:
class A(BaseModel):
type: Literal["a"]
data: str
class B(BaseModel):
type: Literal["b"]
data: int
m = construct_type(
value={"type": "b", "data": "foo"},
type_=cast(Any, Annotated[Union[A, B], PropertyInfo(discriminator="type")]),
)
assert isinstance(m, B)
assert m.type == "b"
assert m.data == "foo" # type: ignore[comparison-overlap]
m = construct_type(
value={"type": "a", "data": 100},
type_=cast(Any, Annotated[Union[A, B], PropertyInfo(discriminator="type")]),
)
assert isinstance(m, A)
assert m.type == "a"
if PYDANTIC_V2:
assert m.data == 100 # type: ignore[comparison-overlap]
else:
# pydantic v1 automatically converts inputs to strings
# if the expected type is a str
assert m.data == "100"
def test_discriminated_unions_unknown_variant() -> None:
class A(BaseModel):
type: Literal["a"]
data: str
class B(BaseModel):
type: Literal["b"]
data: int
m = construct_type(
value={"type": "c", "data": None, "new_thing": "bar"},
type_=cast(Any, Annotated[Union[A, B], PropertyInfo(discriminator="type")]),
)
# just chooses the first variant
assert isinstance(m, A)
assert m.type == "c" # type: ignore[comparison-overlap]
assert m.data == None # type: ignore[unreachable]
assert m.new_thing == "bar"
def test_discriminated_unions_invalid_data_nested_unions() -> None:
class A(BaseModel):
type: Literal["a"]
data: str
class B(BaseModel):
type: Literal["b"]
data: int
class C(BaseModel):
type: Literal["c"]
data: bool
m = construct_type(
value={"type": "b", "data": "foo"},
type_=cast(Any, Annotated[Union[Union[A, B], C], PropertyInfo(discriminator="type")]),
)
assert isinstance(m, B)
assert m.type == "b"
assert m.data == "foo" # type: ignore[comparison-overlap]
m = construct_type(
value={"type": "c", "data": "foo"},
type_=cast(Any, Annotated[Union[Union[A, B], C], PropertyInfo(discriminator="type")]),
)
assert isinstance(m, C)
assert m.type == "c"
assert m.data == "foo" # type: ignore[comparison-overlap]
def test_discriminated_unions_with_aliases_invalid_data() -> None:
class A(BaseModel):
foo_type: Literal["a"] = Field(alias="type")
data: str
class B(BaseModel):
foo_type: Literal["b"] = Field(alias="type")
data: int
m = construct_type(
value={"type": "b", "data": "foo"},
type_=cast(Any, Annotated[Union[A, B], PropertyInfo(discriminator="foo_type")]),
)
assert isinstance(m, B)
assert m.foo_type == "b"
assert m.data == "foo" # type: ignore[comparison-overlap]
m = construct_type(
value={"type": "a", "data": 100},
type_=cast(Any, Annotated[Union[A, B], PropertyInfo(discriminator="foo_type")]),
)
assert isinstance(m, A)
assert m.foo_type == "a"
if PYDANTIC_V2:
assert m.data == 100 # type: ignore[comparison-overlap]
else:
# pydantic v1 automatically converts inputs to strings
# if the expected type is a str
assert m.data == "100"
def test_discriminated_unions_overlapping_discriminators_invalid_data() -> None:
class A(BaseModel):
type: Literal["a"]
data: bool
class B(BaseModel):
type: Literal["a"]
data: int
m = construct_type(
value={"type": "a", "data": "foo"},
type_=cast(Any, Annotated[Union[A, B], PropertyInfo(discriminator="type")]),
)
assert isinstance(m, B)
assert m.type == "a"
assert m.data == "foo" # type: ignore[comparison-overlap]
def test_discriminated_unions_invalid_data_uses_cache() -> None:
class A(BaseModel):
type: Literal["a"]
data: str
class B(BaseModel):
type: Literal["b"]
data: int
UnionType = cast(Any, Union[A, B])
assert not hasattr(UnionType, "__discriminator__")
m = construct_type(
value={"type": "b", "data": "foo"}, type_=cast(Any, Annotated[UnionType, PropertyInfo(discriminator="type")])
)
assert isinstance(m, B)
assert m.type == "b"
assert m.data == "foo" # type: ignore[comparison-overlap]
discriminator = UnionType.__discriminator__
assert discriminator is not None
m = construct_type(
value={"type": "b", "data": "foo"}, type_=cast(Any, Annotated[UnionType, PropertyInfo(discriminator="type")])
)
assert isinstance(m, B)
assert m.type == "b"
assert m.data == "foo" # type: ignore[comparison-overlap]
# if the discriminator details object stays the same between invocations then
# we hit the cache
assert UnionType.__discriminator__ is discriminator
@pytest.mark.skipif(not PYDANTIC_V2, reason="TypeAliasType is not supported in Pydantic v1")
def test_type_alias_type() -> None:
Alias = TypeAliasType("Alias", str) # pyright: ignore
class Model(BaseModel):
alias: Alias
union: Union[int, Alias]
m = construct_type(value={"alias": "foo", "union": "bar"}, type_=Model)
assert isinstance(m, Model)
assert isinstance(m.alias, str)
assert m.alias == "foo"
assert isinstance(m.union, str)
assert m.union == "bar"
@pytest.mark.skipif(not PYDANTIC_V2, reason="TypeAliasType is not supported in Pydantic v1")
def test_field_named_cls() -> None:
class Model(BaseModel):
cls: str
m = construct_type(value={"cls": "foo"}, type_=Model)
assert isinstance(m, Model)
assert isinstance(m.cls, str)
def test_discriminated_union_case() -> None:
class A(BaseModel):
type: Literal["a"]
data: bool
class B(BaseModel):
type: Literal["b"]
data: List[Union[A, object]]
class ModelA(BaseModel):
type: Literal["modelA"]
data: int
class ModelB(BaseModel):
type: Literal["modelB"]
required: str
data: Union[A, B]
# when constructing ModelA | ModelB, value data doesn't match ModelB exactly - missing `required`
m = construct_type(
value={"type": "modelB", "data": {"type": "a", "data": True}},
type_=cast(Any, Annotated[Union[ModelA, ModelB], PropertyInfo(discriminator="type")]),
)
assert isinstance(m, ModelB)

78
tests/test_qs.py Normal file
View file

@ -0,0 +1,78 @@
from typing import Any, cast
from functools import partial
from urllib.parse import unquote
import pytest
from opencode._qs import Querystring, stringify
def test_empty() -> None:
assert stringify({}) == ""
assert stringify({"a": {}}) == ""
assert stringify({"a": {"b": {"c": {}}}}) == ""
def test_basic() -> None:
assert stringify({"a": 1}) == "a=1"
assert stringify({"a": "b"}) == "a=b"
assert stringify({"a": True}) == "a=true"
assert stringify({"a": False}) == "a=false"
assert stringify({"a": 1.23456}) == "a=1.23456"
assert stringify({"a": None}) == ""
@pytest.mark.parametrize("method", ["class", "function"])
def test_nested_dotted(method: str) -> None:
if method == "class":
serialise = Querystring(nested_format="dots").stringify
else:
serialise = partial(stringify, nested_format="dots")
assert unquote(serialise({"a": {"b": "c"}})) == "a.b=c"
assert unquote(serialise({"a": {"b": "c", "d": "e", "f": "g"}})) == "a.b=c&a.d=e&a.f=g"
assert unquote(serialise({"a": {"b": {"c": {"d": "e"}}}})) == "a.b.c.d=e"
assert unquote(serialise({"a": {"b": True}})) == "a.b=true"
def test_nested_brackets() -> None:
assert unquote(stringify({"a": {"b": "c"}})) == "a[b]=c"
assert unquote(stringify({"a": {"b": "c", "d": "e", "f": "g"}})) == "a[b]=c&a[d]=e&a[f]=g"
assert unquote(stringify({"a": {"b": {"c": {"d": "e"}}}})) == "a[b][c][d]=e"
assert unquote(stringify({"a": {"b": True}})) == "a[b]=true"
@pytest.mark.parametrize("method", ["class", "function"])
def test_array_comma(method: str) -> None:
if method == "class":
serialise = Querystring(array_format="comma").stringify
else:
serialise = partial(stringify, array_format="comma")
assert unquote(serialise({"in": ["foo", "bar"]})) == "in=foo,bar"
assert unquote(serialise({"a": {"b": [True, False]}})) == "a[b]=true,false"
assert unquote(serialise({"a": {"b": [True, False, None, True]}})) == "a[b]=true,false,true"
def test_array_repeat() -> None:
assert unquote(stringify({"in": ["foo", "bar"]})) == "in=foo&in=bar"
assert unquote(stringify({"a": {"b": [True, False]}})) == "a[b]=true&a[b]=false"
assert unquote(stringify({"a": {"b": [True, False, None, True]}})) == "a[b]=true&a[b]=false&a[b]=true"
assert unquote(stringify({"in": ["foo", {"b": {"c": ["d", "e"]}}]})) == "in=foo&in[b][c]=d&in[b][c]=e"
@pytest.mark.parametrize("method", ["class", "function"])
def test_array_brackets(method: str) -> None:
if method == "class":
serialise = Querystring(array_format="brackets").stringify
else:
serialise = partial(stringify, array_format="brackets")
assert unquote(serialise({"in": ["foo", "bar"]})) == "in[]=foo&in[]=bar"
assert unquote(serialise({"a": {"b": [True, False]}})) == "a[b][]=true&a[b][]=false"
assert unquote(serialise({"a": {"b": [True, False, None, True]}})) == "a[b][]=true&a[b][]=false&a[b][]=true"
def test_unknown_array_format() -> None:
with pytest.raises(NotImplementedError, match="Unknown array_format value: foo, choose from comma, repeat"):
stringify({"a": ["foo", "bar"]}, array_format=cast(Any, "foo"))

111
tests/test_required_args.py Normal file
View file

@ -0,0 +1,111 @@
from __future__ import annotations
import pytest
from opencode._utils import required_args
def test_too_many_positional_params() -> None:
@required_args(["a"])
def foo(a: str | None = None) -> str | None:
return a
with pytest.raises(TypeError, match=r"foo\(\) takes 1 argument\(s\) but 2 were given"):
foo("a", "b") # type: ignore
def test_positional_param() -> None:
@required_args(["a"])
def foo(a: str | None = None) -> str | None:
return a
assert foo("a") == "a"
assert foo(None) is None
assert foo(a="b") == "b"
with pytest.raises(TypeError, match="Missing required argument: 'a'"):
foo()
def test_keyword_only_param() -> None:
@required_args(["a"])
def foo(*, a: str | None = None) -> str | None:
return a
assert foo(a="a") == "a"
assert foo(a=None) is None
assert foo(a="b") == "b"
with pytest.raises(TypeError, match="Missing required argument: 'a'"):
foo()
def test_multiple_params() -> None:
@required_args(["a", "b", "c"])
def foo(a: str = "", *, b: str = "", c: str = "") -> str | None:
return f"{a} {b} {c}"
assert foo(a="a", b="b", c="c") == "a b c"
error_message = r"Missing required arguments.*"
with pytest.raises(TypeError, match=error_message):
foo()
with pytest.raises(TypeError, match=error_message):
foo(a="a")
with pytest.raises(TypeError, match=error_message):
foo(b="b")
with pytest.raises(TypeError, match=error_message):
foo(c="c")
with pytest.raises(TypeError, match=r"Missing required argument: 'a'"):
foo(b="a", c="c")
with pytest.raises(TypeError, match=r"Missing required argument: 'b'"):
foo("a", c="c")
def test_multiple_variants() -> None:
@required_args(["a"], ["b"])
def foo(*, a: str | None = None, b: str | None = None) -> str | None:
return a if a is not None else b
assert foo(a="foo") == "foo"
assert foo(b="bar") == "bar"
assert foo(a=None) is None
assert foo(b=None) is None
# TODO: this error message could probably be improved
with pytest.raises(
TypeError,
match=r"Missing required arguments; Expected either \('a'\) or \('b'\) arguments to be given",
):
foo()
def test_multiple_params_multiple_variants() -> None:
@required_args(["a", "b"], ["c"])
def foo(*, a: str | None = None, b: str | None = None, c: str | None = None) -> str | None:
if a is not None:
return a
if b is not None:
return b
return c
error_message = r"Missing required arguments; Expected either \('a' and 'b'\) or \('c'\) arguments to be given"
with pytest.raises(TypeError, match=error_message):
foo(a="foo")
with pytest.raises(TypeError, match=error_message):
foo(b="bar")
with pytest.raises(TypeError, match=error_message):
foo()
assert foo(a=None, b="bar") == "bar"
assert foo(c=None) is None
assert foo(c="foo") == "foo"

277
tests/test_response.py Normal file
View file

@ -0,0 +1,277 @@
import json
from typing import Any, List, Union, cast
from typing_extensions import Annotated
import httpx
import pytest
import pydantic
from opencode import Opencode, BaseModel, AsyncOpencode
from opencode._response import (
APIResponse,
BaseAPIResponse,
AsyncAPIResponse,
BinaryAPIResponse,
AsyncBinaryAPIResponse,
extract_response_type,
)
from opencode._streaming import Stream
from opencode._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 'opencode._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: Opencode) -> 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 opencode import BaseModel`",
):
response.parse(to=PydanticModel)
@pytest.mark.asyncio
async def test_async_response_parse_mismatched_basemodel(async_client: AsyncOpencode) -> 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 opencode import BaseModel`",
):
await response.parse(to=PydanticModel)
def test_response_parse_custom_stream(client: Opencode) -> 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: AsyncOpencode) -> 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: Opencode) -> 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: AsyncOpencode) -> 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: Opencode) -> 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: AsyncOpencode) -> 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: Opencode, 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: AsyncOpencode, 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: Opencode) -> 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: AsyncOpencode) -> 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"

248
tests/test_streaming.py Normal file
View file

@ -0,0 +1,248 @@
from __future__ import annotations
from typing import Iterator, AsyncIterator
import httpx
import pytest
from opencode import Opencode, AsyncOpencode
from opencode._streaming import Stream, AsyncStream, ServerSentEvent
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_basic(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b"event: completion\n"
yield b'data: {"foo":true}\n'
yield b"\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event == "completion"
assert sse.json() == {"foo": True}
await assert_empty_iter(iterator)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_data_missing_event(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b'data: {"foo":true}\n'
yield b"\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event is None
assert sse.json() == {"foo": True}
await assert_empty_iter(iterator)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_event_missing_data(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b"event: ping\n"
yield b"\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event == "ping"
assert sse.data == ""
await assert_empty_iter(iterator)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_multiple_events(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b"event: ping\n"
yield b"\n"
yield b"event: completion\n"
yield b"\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event == "ping"
assert sse.data == ""
sse = await iter_next(iterator)
assert sse.event == "completion"
assert sse.data == ""
await assert_empty_iter(iterator)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_multiple_events_with_data(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b"event: ping\n"
yield b'data: {"foo":true}\n'
yield b"\n"
yield b"event: completion\n"
yield b'data: {"bar":false}\n'
yield b"\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event == "ping"
assert sse.json() == {"foo": True}
sse = await iter_next(iterator)
assert sse.event == "completion"
assert sse.json() == {"bar": False}
await assert_empty_iter(iterator)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_multiple_data_lines_with_empty_line(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b"event: ping\n"
yield b"data: {\n"
yield b'data: "foo":\n'
yield b"data: \n"
yield b"data:\n"
yield b"data: true}\n"
yield b"\n\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event == "ping"
assert sse.json() == {"foo": True}
assert sse.data == '{\n"foo":\n\n\ntrue}'
await assert_empty_iter(iterator)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_data_json_escaped_double_new_line(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b"event: ping\n"
yield b'data: {"foo": "my long\\n\\ncontent"}'
yield b"\n\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event == "ping"
assert sse.json() == {"foo": "my long\n\ncontent"}
await assert_empty_iter(iterator)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_multiple_data_lines(sync: bool, client: Opencode, async_client: AsyncOpencode) -> None:
def body() -> Iterator[bytes]:
yield b"event: ping\n"
yield b"data: {\n"
yield b'data: "foo":\n'
yield b"data: true}\n"
yield b"\n\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event == "ping"
assert sse.json() == {"foo": True}
await assert_empty_iter(iterator)
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_special_new_line_character(
sync: bool,
client: Opencode,
async_client: AsyncOpencode,
) -> None:
def body() -> Iterator[bytes]:
yield b'data: {"content":" culpa"}\n'
yield b"\n"
yield b'data: {"content":" \xe2\x80\xa8"}\n'
yield b"\n"
yield b'data: {"content":"foo"}\n'
yield b"\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event is None
assert sse.json() == {"content": " culpa"}
sse = await iter_next(iterator)
assert sse.event is None
assert sse.json() == {"content": " 
"}
sse = await iter_next(iterator)
assert sse.event is None
assert sse.json() == {"content": "foo"}
await assert_empty_iter(iterator)
@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
async def test_multi_byte_character_multiple_chunks(
sync: bool,
client: Opencode,
async_client: AsyncOpencode,
) -> None:
def body() -> Iterator[bytes]:
yield b'data: {"content":"'
# bytes taken from the string 'известни' and arbitrarily split
# so that some multi-byte characters span multiple chunks
yield b"\xd0"
yield b"\xb8\xd0\xb7\xd0"
yield b"\xb2\xd0\xb5\xd1\x81\xd1\x82\xd0\xbd\xd0\xb8"
yield b'"}\n'
yield b"\n"
iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
sse = await iter_next(iterator)
assert sse.event is None
assert sse.json() == {"content": "известни"}
async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
for chunk in iter:
yield chunk
async def iter_next(iter: Iterator[ServerSentEvent] | AsyncIterator[ServerSentEvent]) -> ServerSentEvent:
if isinstance(iter, AsyncIterator):
return await iter.__anext__()
return next(iter)
async def assert_empty_iter(iter: Iterator[ServerSentEvent] | AsyncIterator[ServerSentEvent]) -> None:
with pytest.raises((StopAsyncIteration, RuntimeError)):
await iter_next(iter)
def make_event_iterator(
content: Iterator[bytes],
*,
sync: bool,
client: Opencode,
async_client: AsyncOpencode,
) -> Iterator[ServerSentEvent] | AsyncIterator[ServerSentEvent]:
if sync:
return Stream(cast_to=object, client=client, response=httpx.Response(200, content=content))._iter_events()
return AsyncStream(
cast_to=object, client=async_client, response=httpx.Response(200, content=to_aiter(content))
)._iter_events()

453
tests/test_transform.py Normal file
View file

@ -0,0 +1,453 @@
from __future__ import annotations
import io
import pathlib
from typing import Any, Dict, List, Union, TypeVar, Iterable, Optional, cast
from datetime import date, datetime
from typing_extensions import Required, Annotated, TypedDict
import pytest
from opencode._types import NOT_GIVEN, Base64FileInput
from opencode._utils import (
PropertyInfo,
transform as _transform,
parse_datetime,
async_transform as _async_transform,
)
from opencode._compat import PYDANTIC_V2
from opencode._models import BaseModel
_T = TypeVar("_T")
SAMPLE_FILE_PATH = pathlib.Path(__file__).parent.joinpath("sample_file.txt")
async def transform(
data: _T,
expected_type: object,
use_async: bool,
) -> _T:
if use_async:
return await _async_transform(data, expected_type=expected_type)
return _transform(data, expected_type=expected_type)
parametrize = pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"])
class Foo1(TypedDict):
foo_bar: Annotated[str, PropertyInfo(alias="fooBar")]
@parametrize
@pytest.mark.asyncio
async def test_top_level_alias(use_async: bool) -> None:
assert await transform({"foo_bar": "hello"}, expected_type=Foo1, use_async=use_async) == {"fooBar": "hello"}
class Foo2(TypedDict):
bar: Bar2
class Bar2(TypedDict):
this_thing: Annotated[int, PropertyInfo(alias="this__thing")]
baz: Annotated[Baz2, PropertyInfo(alias="Baz")]
class Baz2(TypedDict):
my_baz: Annotated[str, PropertyInfo(alias="myBaz")]
@parametrize
@pytest.mark.asyncio
async def test_recursive_typeddict(use_async: bool) -> None:
assert await transform({"bar": {"this_thing": 1}}, Foo2, use_async) == {"bar": {"this__thing": 1}}
assert await transform({"bar": {"baz": {"my_baz": "foo"}}}, Foo2, use_async) == {"bar": {"Baz": {"myBaz": "foo"}}}
class Foo3(TypedDict):
things: List[Bar3]
class Bar3(TypedDict):
my_field: Annotated[str, PropertyInfo(alias="myField")]
@parametrize
@pytest.mark.asyncio
async def test_list_of_typeddict(use_async: bool) -> None:
result = await transform({"things": [{"my_field": "foo"}, {"my_field": "foo2"}]}, Foo3, use_async)
assert result == {"things": [{"myField": "foo"}, {"myField": "foo2"}]}
class Foo4(TypedDict):
foo: Union[Bar4, Baz4]
class Bar4(TypedDict):
foo_bar: Annotated[str, PropertyInfo(alias="fooBar")]
class Baz4(TypedDict):
foo_baz: Annotated[str, PropertyInfo(alias="fooBaz")]
@parametrize
@pytest.mark.asyncio
async def test_union_of_typeddict(use_async: bool) -> None:
assert await transform({"foo": {"foo_bar": "bar"}}, Foo4, use_async) == {"foo": {"fooBar": "bar"}}
assert await transform({"foo": {"foo_baz": "baz"}}, Foo4, use_async) == {"foo": {"fooBaz": "baz"}}
assert await transform({"foo": {"foo_baz": "baz", "foo_bar": "bar"}}, Foo4, use_async) == {
"foo": {"fooBaz": "baz", "fooBar": "bar"}
}
class Foo5(TypedDict):
foo: Annotated[Union[Bar4, List[Baz4]], PropertyInfo(alias="FOO")]
class Bar5(TypedDict):
foo_bar: Annotated[str, PropertyInfo(alias="fooBar")]
class Baz5(TypedDict):
foo_baz: Annotated[str, PropertyInfo(alias="fooBaz")]
@parametrize
@pytest.mark.asyncio
async def test_union_of_list(use_async: bool) -> None:
assert await transform({"foo": {"foo_bar": "bar"}}, Foo5, use_async) == {"FOO": {"fooBar": "bar"}}
assert await transform(
{
"foo": [
{"foo_baz": "baz"},
{"foo_baz": "baz"},
]
},
Foo5,
use_async,
) == {"FOO": [{"fooBaz": "baz"}, {"fooBaz": "baz"}]}
class Foo6(TypedDict):
bar: Annotated[str, PropertyInfo(alias="Bar")]
@parametrize
@pytest.mark.asyncio
async def test_includes_unknown_keys(use_async: bool) -> None:
assert await transform({"bar": "bar", "baz_": {"FOO": 1}}, Foo6, use_async) == {
"Bar": "bar",
"baz_": {"FOO": 1},
}
class Foo7(TypedDict):
bar: Annotated[List[Bar7], PropertyInfo(alias="bAr")]
foo: Bar7
class Bar7(TypedDict):
foo: str
@parametrize
@pytest.mark.asyncio
async def test_ignores_invalid_input(use_async: bool) -> None:
assert await transform({"bar": "<foo>"}, Foo7, use_async) == {"bAr": "<foo>"}
assert await transform({"foo": "<foo>"}, Foo7, use_async) == {"foo": "<foo>"}
class DatetimeDict(TypedDict, total=False):
foo: Annotated[datetime, PropertyInfo(format="iso8601")]
bar: Annotated[Optional[datetime], PropertyInfo(format="iso8601")]
required: Required[Annotated[Optional[datetime], PropertyInfo(format="iso8601")]]
list_: Required[Annotated[Optional[List[datetime]], PropertyInfo(format="iso8601")]]
union: Annotated[Union[int, datetime], PropertyInfo(format="iso8601")]
class DateDict(TypedDict, total=False):
foo: Annotated[date, PropertyInfo(format="iso8601")]
class DatetimeModel(BaseModel):
foo: datetime
class DateModel(BaseModel):
foo: Optional[date]
@parametrize
@pytest.mark.asyncio
async def test_iso8601_format(use_async: bool) -> None:
dt = datetime.fromisoformat("2023-02-23T14:16:36.337692+00:00")
tz = "Z" if PYDANTIC_V2 else "+00:00"
assert await transform({"foo": dt}, DatetimeDict, use_async) == {"foo": "2023-02-23T14:16:36.337692+00:00"} # type: ignore[comparison-overlap]
assert await transform(DatetimeModel(foo=dt), Any, use_async) == {"foo": "2023-02-23T14:16:36.337692" + tz} # type: ignore[comparison-overlap]
dt = dt.replace(tzinfo=None)
assert await transform({"foo": dt}, DatetimeDict, use_async) == {"foo": "2023-02-23T14:16:36.337692"} # type: ignore[comparison-overlap]
assert await transform(DatetimeModel(foo=dt), Any, use_async) == {"foo": "2023-02-23T14:16:36.337692"} # type: ignore[comparison-overlap]
assert await transform({"foo": None}, DateDict, use_async) == {"foo": None} # type: ignore[comparison-overlap]
assert await transform(DateModel(foo=None), Any, use_async) == {"foo": None} # type: ignore
assert await transform({"foo": date.fromisoformat("2023-02-23")}, DateDict, use_async) == {"foo": "2023-02-23"} # type: ignore[comparison-overlap]
assert await transform(DateModel(foo=date.fromisoformat("2023-02-23")), DateDict, use_async) == {
"foo": "2023-02-23"
} # type: ignore[comparison-overlap]
@parametrize
@pytest.mark.asyncio
async def test_optional_iso8601_format(use_async: bool) -> None:
dt = datetime.fromisoformat("2023-02-23T14:16:36.337692+00:00")
assert await transform({"bar": dt}, DatetimeDict, use_async) == {"bar": "2023-02-23T14:16:36.337692+00:00"} # type: ignore[comparison-overlap]
assert await transform({"bar": None}, DatetimeDict, use_async) == {"bar": None}
@parametrize
@pytest.mark.asyncio
async def test_required_iso8601_format(use_async: bool) -> None:
dt = datetime.fromisoformat("2023-02-23T14:16:36.337692+00:00")
assert await transform({"required": dt}, DatetimeDict, use_async) == {
"required": "2023-02-23T14:16:36.337692+00:00"
} # type: ignore[comparison-overlap]
assert await transform({"required": None}, DatetimeDict, use_async) == {"required": None}
@parametrize
@pytest.mark.asyncio
async def test_union_datetime(use_async: bool) -> None:
dt = datetime.fromisoformat("2023-02-23T14:16:36.337692+00:00")
assert await transform({"union": dt}, DatetimeDict, use_async) == { # type: ignore[comparison-overlap]
"union": "2023-02-23T14:16:36.337692+00:00"
}
assert await transform({"union": "foo"}, DatetimeDict, use_async) == {"union": "foo"}
@parametrize
@pytest.mark.asyncio
async def test_nested_list_iso6801_format(use_async: bool) -> None:
dt1 = datetime.fromisoformat("2023-02-23T14:16:36.337692+00:00")
dt2 = parse_datetime("2022-01-15T06:34:23Z")
assert await transform({"list_": [dt1, dt2]}, DatetimeDict, use_async) == { # type: ignore[comparison-overlap]
"list_": ["2023-02-23T14:16:36.337692+00:00", "2022-01-15T06:34:23+00:00"]
}
@parametrize
@pytest.mark.asyncio
async def test_datetime_custom_format(use_async: bool) -> None:
dt = parse_datetime("2022-01-15T06:34:23Z")
result = await transform(dt, Annotated[datetime, PropertyInfo(format="custom", format_template="%H")], use_async)
assert result == "06" # type: ignore[comparison-overlap]
class DateDictWithRequiredAlias(TypedDict, total=False):
required_prop: Required[Annotated[date, PropertyInfo(format="iso8601", alias="prop")]]
@parametrize
@pytest.mark.asyncio
async def test_datetime_with_alias(use_async: bool) -> None:
assert await transform({"required_prop": None}, DateDictWithRequiredAlias, use_async) == {"prop": None} # type: ignore[comparison-overlap]
assert await transform(
{"required_prop": date.fromisoformat("2023-02-23")}, DateDictWithRequiredAlias, use_async
) == {"prop": "2023-02-23"} # type: ignore[comparison-overlap]
class MyModel(BaseModel):
foo: str
@parametrize
@pytest.mark.asyncio
async def test_pydantic_model_to_dictionary(use_async: bool) -> None:
assert cast(Any, await transform(MyModel(foo="hi!"), Any, use_async)) == {"foo": "hi!"}
assert cast(Any, await transform(MyModel.construct(foo="hi!"), Any, use_async)) == {"foo": "hi!"}
@parametrize
@pytest.mark.asyncio
async def test_pydantic_empty_model(use_async: bool) -> None:
assert cast(Any, await transform(MyModel.construct(), Any, use_async)) == {}
@parametrize
@pytest.mark.asyncio
async def test_pydantic_unknown_field(use_async: bool) -> None:
assert cast(Any, await transform(MyModel.construct(my_untyped_field=True), Any, use_async)) == {
"my_untyped_field": True
}
@parametrize
@pytest.mark.asyncio
async def test_pydantic_mismatched_types(use_async: bool) -> None:
model = MyModel.construct(foo=True)
if PYDANTIC_V2:
with pytest.warns(UserWarning):
params = await transform(model, Any, use_async)
else:
params = await transform(model, Any, use_async)
assert cast(Any, params) == {"foo": True}
@parametrize
@pytest.mark.asyncio
async def test_pydantic_mismatched_object_type(use_async: bool) -> None:
model = MyModel.construct(foo=MyModel.construct(hello="world"))
if PYDANTIC_V2:
with pytest.warns(UserWarning):
params = await transform(model, Any, use_async)
else:
params = await transform(model, Any, use_async)
assert cast(Any, params) == {"foo": {"hello": "world"}}
class ModelNestedObjects(BaseModel):
nested: MyModel
@parametrize
@pytest.mark.asyncio
async def test_pydantic_nested_objects(use_async: bool) -> None:
model = ModelNestedObjects.construct(nested={"foo": "stainless"})
assert isinstance(model.nested, MyModel)
assert cast(Any, await transform(model, Any, use_async)) == {"nested": {"foo": "stainless"}}
class ModelWithDefaultField(BaseModel):
foo: str
with_none_default: Union[str, None] = None
with_str_default: str = "foo"
@parametrize
@pytest.mark.asyncio
async def test_pydantic_default_field(use_async: bool) -> None:
# should be excluded when defaults are used
model = ModelWithDefaultField.construct()
assert model.with_none_default is None
assert model.with_str_default == "foo"
assert cast(Any, await transform(model, Any, use_async)) == {}
# should be included when the default value is explicitly given
model = ModelWithDefaultField.construct(with_none_default=None, with_str_default="foo")
assert model.with_none_default is None
assert model.with_str_default == "foo"
assert cast(Any, await transform(model, Any, use_async)) == {"with_none_default": None, "with_str_default": "foo"}
# should be included when a non-default value is explicitly given
model = ModelWithDefaultField.construct(with_none_default="bar", with_str_default="baz")
assert model.with_none_default == "bar"
assert model.with_str_default == "baz"
assert cast(Any, await transform(model, Any, use_async)) == {"with_none_default": "bar", "with_str_default": "baz"}
class TypedDictIterableUnion(TypedDict):
foo: Annotated[Union[Bar8, Iterable[Baz8]], PropertyInfo(alias="FOO")]
class Bar8(TypedDict):
foo_bar: Annotated[str, PropertyInfo(alias="fooBar")]
class Baz8(TypedDict):
foo_baz: Annotated[str, PropertyInfo(alias="fooBaz")]
@parametrize
@pytest.mark.asyncio
async def test_iterable_of_dictionaries(use_async: bool) -> None:
assert await transform({"foo": [{"foo_baz": "bar"}]}, TypedDictIterableUnion, use_async) == {
"FOO": [{"fooBaz": "bar"}]
}
assert cast(Any, await transform({"foo": ({"foo_baz": "bar"},)}, TypedDictIterableUnion, use_async)) == {
"FOO": [{"fooBaz": "bar"}]
}
def my_iter() -> Iterable[Baz8]:
yield {"foo_baz": "hello"}
yield {"foo_baz": "world"}
assert await transform({"foo": my_iter()}, TypedDictIterableUnion, use_async) == {
"FOO": [{"fooBaz": "hello"}, {"fooBaz": "world"}]
}
@parametrize
@pytest.mark.asyncio
async def test_dictionary_items(use_async: bool) -> None:
class DictItems(TypedDict):
foo_baz: Annotated[str, PropertyInfo(alias="fooBaz")]
assert await transform({"foo": {"foo_baz": "bar"}}, Dict[str, DictItems], use_async) == {"foo": {"fooBaz": "bar"}}
class TypedDictIterableUnionStr(TypedDict):
foo: Annotated[Union[str, Iterable[Baz8]], PropertyInfo(alias="FOO")]
@parametrize
@pytest.mark.asyncio
async def test_iterable_union_str(use_async: bool) -> None:
assert await transform({"foo": "bar"}, TypedDictIterableUnionStr, use_async) == {"FOO": "bar"}
assert cast(Any, await transform(iter([{"foo_baz": "bar"}]), Union[str, Iterable[Baz8]], use_async)) == [
{"fooBaz": "bar"}
]
class TypedDictBase64Input(TypedDict):
foo: Annotated[Union[str, Base64FileInput], PropertyInfo(format="base64")]
@parametrize
@pytest.mark.asyncio
async def test_base64_file_input(use_async: bool) -> None:
# strings are left as-is
assert await transform({"foo": "bar"}, TypedDictBase64Input, use_async) == {"foo": "bar"}
# pathlib.Path is automatically converted to base64
assert await transform({"foo": SAMPLE_FILE_PATH}, TypedDictBase64Input, use_async) == {
"foo": "SGVsbG8sIHdvcmxkIQo="
} # type: ignore[comparison-overlap]
# io instances are automatically converted to base64
assert await transform({"foo": io.StringIO("Hello, world!")}, TypedDictBase64Input, use_async) == {
"foo": "SGVsbG8sIHdvcmxkIQ=="
} # type: ignore[comparison-overlap]
assert await transform({"foo": io.BytesIO(b"Hello, world!")}, TypedDictBase64Input, use_async) == {
"foo": "SGVsbG8sIHdvcmxkIQ=="
} # type: ignore[comparison-overlap]
@parametrize
@pytest.mark.asyncio
async def test_transform_skipping(use_async: bool) -> None:
# lists of ints are left as-is
data = [1, 2, 3]
assert await transform(data, List[int], use_async) is data
# iterables of ints are converted to a list
data = iter([1, 2, 3])
assert await transform(data, Iterable[int], use_async) == [1, 2, 3]
@parametrize
@pytest.mark.asyncio
async def test_strips_notgiven(use_async: bool) -> None:
assert await transform({"foo_bar": "bar"}, Foo1, use_async) == {"fooBar": "bar"}
assert await transform({"foo_bar": NOT_GIVEN}, Foo1, use_async) == {}

34
tests/test_utils/test_proxy.py Normal file
View file

@ -0,0 +1,34 @@
import operator
from typing import Any
from typing_extensions import override
from opencode._utils import LazyProxy
class RecursiveLazyProxy(LazyProxy[Any]):
@override
def __load__(self) -> Any:
return self
def __call__(self, *_args: Any, **_kwds: Any) -> Any:
raise RuntimeError("This should never be called!")
def test_recursive_proxy() -> None:
proxy = RecursiveLazyProxy()
assert repr(proxy) == "RecursiveLazyProxy"
assert str(proxy) == "RecursiveLazyProxy"
assert dir(proxy) == []
assert type(proxy).__name__ == "RecursiveLazyProxy"
assert type(operator.attrgetter("name.foo.bar.baz")(proxy)).__name__ == "RecursiveLazyProxy"
def test_isinstance_does_not_error() -> None:
class AlwaysErrorProxy(LazyProxy[Any]):
@override
def __load__(self) -> Any:
raise RuntimeError("Mocking missing dependency")
proxy = AlwaysErrorProxy()
assert not isinstance(proxy, dict)
assert isinstance(proxy, LazyProxy)

73
tests/test_utils/test_typing.py Normal file
View file

@ -0,0 +1,73 @@
from __future__ import annotations
from typing import Generic, TypeVar, cast
from opencode._utils import extract_type_var_from_base
_T = TypeVar("_T")
_T2 = TypeVar("_T2")
_T3 = TypeVar("_T3")
class BaseGeneric(Generic[_T]): ...
class SubclassGeneric(BaseGeneric[_T]): ...
class BaseGenericMultipleTypeArgs(Generic[_T, _T2, _T3]): ...
class SubclassGenericMultipleTypeArgs(BaseGenericMultipleTypeArgs[_T, _T2, _T3]): ...
class SubclassDifferentOrderGenericMultipleTypeArgs(BaseGenericMultipleTypeArgs[_T2, _T, _T3]): ...
def test_extract_type_var() -> None:
assert (
extract_type_var_from_base(
BaseGeneric[int],
index=0,
generic_bases=cast("tuple[type, ...]", (BaseGeneric,)),
)
== int
)
def test_extract_type_var_generic_subclass() -> None:
assert (
extract_type_var_from_base(
SubclassGeneric[int],
index=0,
generic_bases=cast("tuple[type, ...]", (BaseGeneric,)),
)
== int
)
def test_extract_type_var_multiple() -> None:
typ = BaseGenericMultipleTypeArgs[int, str, None]
generic_bases = cast("tuple[type, ...]", (BaseGenericMultipleTypeArgs,))
assert extract_type_var_from_base(typ, index=0, generic_bases=generic_bases) == int
assert extract_type_var_from_base(typ, index=1, generic_bases=generic_bases) == str
assert extract_type_var_from_base(typ, index=2, generic_bases=generic_bases) == type(None)
def test_extract_type_var_generic_subclass_multiple() -> None:
typ = SubclassGenericMultipleTypeArgs[int, str, None]
generic_bases = cast("tuple[type, ...]", (BaseGenericMultipleTypeArgs,))
assert extract_type_var_from_base(typ, index=0, generic_bases=generic_bases) == int
assert extract_type_var_from_base(typ, index=1, generic_bases=generic_bases) == str
assert extract_type_var_from_base(typ, index=2, generic_bases=generic_bases) == type(None)
def test_extract_type_var_generic_subclass_different_ordering_multiple() -> None:
typ = SubclassDifferentOrderGenericMultipleTypeArgs[int, str, None]
generic_bases = cast("tuple[type, ...]", (BaseGenericMultipleTypeArgs,))
assert extract_type_var_from_base(typ, index=0, generic_bases=generic_bases) == int
assert extract_type_var_from_base(typ, index=1, generic_bases=generic_bases) == str
assert extract_type_var_from_base(typ, index=2, generic_bases=generic_bases) == type(None)

159
tests/utils.py Normal file
View file

@ -0,0 +1,159 @@
from __future__ import annotations
import os
import inspect
import traceback
import contextlib
from typing import Any, TypeVar, Iterator, cast
from datetime import date, datetime
from typing_extensions import Literal, get_args, get_origin, assert_type
from opencode._types import Omit, NoneType
from opencode._utils import (
is_dict,
is_list,
is_list_type,
is_union_type,
extract_type_arg,
is_annotated_type,
is_type_alias_type,
)
from opencode._compat import PYDANTIC_V2, field_outer_type, get_model_fields
from opencode._models import BaseModel
BaseModelT = TypeVar("BaseModelT", bound=BaseModel)
def assert_matches_model(model: type[BaseModelT], value: BaseModelT, *, path: list[str]) -> bool:
for name, field in get_model_fields(model).items():
field_value = getattr(value, name)
if PYDANTIC_V2:
allow_none = False
else:
# in v1 nullability was structured differently
# https://docs.pydantic.dev/2.0/migration/#required-optional-and-nullable-fields
allow_none = getattr(field, "allow_none", False)
assert_matches_type(
field_outer_type(field),
field_value,
path=[*path, name],
allow_none=allow_none,
)
return True
# Note: the `path` argument is only used to improve error messages when `--showlocals` is used
def assert_matches_type(
type_: Any,
value: object,
*,
path: list[str],
allow_none: bool = False,
) -> None:
if is_type_alias_type(type_):
type_ = type_.__value__
# unwrap `Annotated[T, ...]` -> `T`
if is_annotated_type(type_):
type_ = extract_type_arg(type_, 0)
if allow_none and value is None:
return
if type_ is None or type_ is NoneType:
assert value is None
return
origin = get_origin(type_) or type_
if is_list_type(type_):
return _assert_list_type(type_, value)
if origin == str:
assert isinstance(value, str)
elif origin == int:
assert isinstance(value, int)
elif origin == bool:
assert isinstance(value, bool)
elif origin == float:
assert isinstance(value, float)
elif origin == bytes:
assert isinstance(value, bytes)
elif origin == datetime:
assert isinstance(value, datetime)
elif origin == date:
assert isinstance(value, date)
elif origin == object:
# nothing to do here, the expected type is unknown
pass
elif origin == Literal:
assert value in get_args(type_)
elif origin == dict:
assert is_dict(value)
args = get_args(type_)
key_type = args[0]
items_type = args[1]
for key, item in value.items():
assert_matches_type(key_type, key, path=[*path, "<dict key>"])
assert_matches_type(items_type, item, path=[*path, "<dict item>"])
elif is_union_type(type_):
variants = get_args(type_)
try:
none_index = variants.index(type(None))
except ValueError:
pass
else:
# special case Optional[T] for better error messages
if len(variants) == 2:
if value is None:
# valid
return
return assert_matches_type(type_=variants[not none_index], value=value, path=path)
for i, variant in enumerate(variants):
try:
assert_matches_type(variant, value, path=[*path, f"variant {i}"])
return
except AssertionError:
traceback.print_exc()
continue
raise AssertionError("Did not match any variants")
elif issubclass(origin, BaseModel):
assert isinstance(value, type_)
assert assert_matches_model(type_, cast(Any, value), path=path)
elif inspect.isclass(origin) and origin.__name__ == "HttpxBinaryResponseContent":
assert value.__class__.__name__ == "HttpxBinaryResponseContent"
else:
assert None, f"Unhandled field type: {type_}"
def _assert_list_type(type_: type[object], value: object) -> None:
assert is_list(value)
inner_type = get_args(type_)[0]
for entry in value:
assert_type(inner_type, entry) # type: ignore
@contextlib.contextmanager
def update_env(**new_env: str | Omit) -> Iterator[None]:
old = os.environ.copy()
try:
for name, value in new_env.items():
if isinstance(value, Omit):
os.environ.pop(name, None)
else:
os.environ[name] = value
yield None
finally:
os.environ.clear()
os.environ.update(old)