(self)
| 1805 | assert int(response.http_request.headers.get("x-stainless-retry-count")) == failures_before_success |
| 1806 | |
| 1807 | def test_get_platform(self) -> None: |
| 1808 | # A previous implementation of asyncify could leave threads unterminated when |
| 1809 | # used with nest_asyncio. |
| 1810 | # |
| 1811 | # Since nest_asyncio.apply() is global and cannot be un-applied, this |
| 1812 | # test is run in a separate process to avoid affecting other tests. |
| 1813 | test_code = dedent(""" |
| 1814 | import asyncio |
| 1815 | import nest_asyncio |
| 1816 | import threading |
| 1817 | |
| 1818 | from anthropic._utils import asyncify |
| 1819 | from anthropic._base_client import get_platform |
| 1820 | |
| 1821 | async def test_main() -> None: |
| 1822 | result = await asyncify(get_platform)() |
| 1823 | print(result) |
| 1824 | for thread in threading.enumerate(): |
| 1825 | print(thread.name) |
| 1826 | |
| 1827 | nest_asyncio.apply() |
| 1828 | asyncio.run(test_main()) |
| 1829 | """) |
| 1830 | with subprocess.Popen( |
| 1831 | [sys.executable, "-c", test_code], |
| 1832 | text=True, |
| 1833 | ) as process: |
| 1834 | timeout = 10 # seconds |
| 1835 | |
| 1836 | start_time = time.monotonic() |
| 1837 | while True: |
| 1838 | return_code = process.poll() |
| 1839 | if return_code is not None: |
| 1840 | if return_code != 0: |
| 1841 | raise AssertionError("calling get_platform using asyncify resulted in a non-zero exit code") |
| 1842 | |
| 1843 | # success |
| 1844 | break |
| 1845 | |
| 1846 | if time.monotonic() - start_time > timeout: |
| 1847 | process.kill() |
| 1848 | raise AssertionError("calling get_platform using asyncify resulted in a hung process") |
| 1849 | |
| 1850 | time.sleep(0.1) |
nothing calls this directly
no outgoing calls
no test coverage detected