self.timeout = timeout
def write_no_block(self, data: bytes) -> None:
- self.stream_writer.write(data)
+ self.stream_writer.write(data) # pragma: nocover
async def write(self, data: bytes, timeout: OptionalTimeout = None) -> None:
if not data:
@pytest.mark.asyncio
async def test_read_timeout(server):
- timeout = TimeoutConfig(read_timeout=0.0001)
+ timeout = TimeoutConfig(read_timeout=0.000001)
async with AsyncClient(timeout=timeout) as client:
with pytest.raises(ReadTimeout):
@pytest.mark.asyncio
async def test_write_timeout(server):
- timeout = TimeoutConfig(write_timeout=0.0001)
+ timeout = TimeoutConfig(write_timeout=0.000001)
async with AsyncClient(timeout=timeout) as client:
with pytest.raises(WriteTimeout):
- data = b"*" * 1024 * 1024 * 10
+ data = b"*" * 1024 * 1024 * 100
await client.put("http://127.0.0.1:8000/slow_response", data=data)
@pytest.mark.asyncio
async def test_connect_timeout(server):
- timeout = TimeoutConfig(connect_timeout=0.0001)
+ timeout = TimeoutConfig(connect_timeout=0.000001)
async with AsyncClient(timeout=timeout) as client:
with pytest.raises(ConnectTimeout):
@pytest.mark.asyncio
async def test_pool_timeout(server):
- pool_limits = PoolLimits(hard_limit=1, pool_timeout=0.0001)
+ pool_limits = PoolLimits(hard_limit=1, pool_timeout=0.000001)
async with AsyncClient(pool_limits=pool_limits) as client:
response = await client.get("http://127.0.0.1:8000/", stream=True)