mirror of
https://github.com/aphrody-labs/typesafe-sdk-python.git
synced 2026-10-14 04:19:06 +00:00
typesafe-public-bot[bot]andDaniel Gafni
c743d81416
Co-authored-by: Daniel Gafni <danielgafni16@gmail.com>
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import truststore
|
|
|
|
from typesafe_sdk._core.client.aio import client as aio_client
|
|
|
|
# Stresses `wrap_bio()` on the default async client's shared SSL context from many threads, as AnyIO does
|
|
# during concurrent TLS setup. Unlocked context configuration aborts the interpreter natively, so the
|
|
# stress runs in a subprocess. See https://github.com/typesafe-ai/typesafe-sdk-python/issues/20.
|
|
_STRESS = """
|
|
import ssl
|
|
import threading
|
|
|
|
from typesafe_sdk import AsyncTypeSafeClient
|
|
|
|
ctx = AsyncTypeSafeClient(api_key="test")._http_client._transport._pool._ssl_context
|
|
|
|
|
|
def worker():
|
|
for _ in range(300):
|
|
ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), server_hostname="api.typesafe.ai")
|
|
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(16)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
"""
|
|
|
|
|
|
def test_default_async_ssl_context_survives_concurrent_wrap_bio() -> None:
|
|
for _ in range(3):
|
|
result = subprocess.run( # noqa: S603 - Fixed interpreter and test-owned script.
|
|
[sys.executable, "-X", "faulthandler", "-c", _STRESS], capture_output=True, text=True, check=False
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
@pytest.mark.parametrize("truststore_version,locked", [("0.10.4", True), ("0.10.5", False), ("0.11.0", False)])
|
|
def test_wrap_bio_lock_only_on_affected_truststore(monkeypatch: pytest.MonkeyPatch, truststore_version: str, locked: bool) -> None:
|
|
monkeypatch.setattr(truststore, "__version__", truststore_version)
|
|
assert ("wrap_bio" in vars(aio_client._ssl_context())) is locked
|
|
|
|
|
|
def test_wrap_bio_lock_skipped_for_stdlib_context(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
monkeypatch.setenv("SSL_CERT_DIR", str(tmp_path))
|
|
monkeypatch.setattr(truststore, "__version__", "0.10.4")
|
|
assert "wrap_bio" not in vars(aio_client._ssl_context())
|