chore(tests): simplify get_platform test

`nest_asyncio` is archived and broken on some platforms so it's
not worth keeping in our test suite.
This commit is contained in:
stainless-app[bot] 2025-09-06 03:38:23 +00:00
parent 273dffa4ac
commit 0a47e601d1
3 changed files with 6 additions and 49 deletions

View file

@ -6,13 +6,10 @@ import gc
import os
import sys
import json
import time
import asyncio
import inspect
import subprocess
import tracemalloc
from typing import Any, Union, cast
from textwrap import dedent
from unittest import mock
from typing_extensions import Literal
@ -23,6 +20,7 @@ from pydantic import ValidationError
from opencode_ai import Opencode, AsyncOpencode, APIResponseValidationError
from opencode_ai._types import Omit
from opencode_ai._utils import asyncify
from opencode_ai._models import BaseModel, FinalRequestOptions
from opencode_ai._streaming import Stream, AsyncStream
from opencode_ai._exceptions import APIStatusError, APITimeoutError, APIResponseValidationError
@ -30,8 +28,10 @@ from opencode_ai._base_client import (
DEFAULT_TIMEOUT,
HTTPX_DEFAULT_TIMEOUT,
BaseClient,
OtherPlatform,
DefaultHttpxClient,
DefaultAsyncHttpxClient,
get_platform,
make_request_options,
)
@ -1566,50 +1566,9 @@ class TestAsyncOpencode:
assert response.http_request.headers.get("x-stainless-retry-count") == "42"
def test_get_platform(self) -> None:
# A previous implementation of asyncify could leave threads unterminated when
# used with nest_asyncio.
#
# Since nest_asyncio.apply() is global and cannot be un-applied, this
# test is run in a separate process to avoid affecting other tests.
test_code = dedent("""
import asyncio
import nest_asyncio
import threading
from opencode_ai._utils import asyncify
from opencode_ai._base_client import get_platform
async def test_main() -> None:
result = await asyncify(get_platform)()
print(result)
for thread in threading.enumerate():
print(thread.name)
nest_asyncio.apply()
asyncio.run(test_main())
""")
with subprocess.Popen(
[sys.executable, "-c", test_code],
text=True,
) as process:
timeout = 10 # seconds
start_time = time.monotonic()
while True:
return_code = process.poll()
if return_code is not None:
if return_code != 0:
raise AssertionError("calling get_platform using asyncify resulted in a non-zero exit code")
# success
break
if time.monotonic() - start_time > timeout:
process.kill()
raise AssertionError("calling get_platform using asyncify resulted in a hung process")
time.sleep(0.1)
async def test_get_platform(self) -> None:
platform = await asyncify(get_platform)()
assert isinstance(platform, (str, OtherPlatform))
async def test_proxy_environment_variables(self, monkeypatch: pytest.MonkeyPatch) -> None:
# Test that the proxy environment variables are set correctly