-# HTTPCore
+# HTTP3
-<a href="https://travis-ci.org/encode/httpcore">
- <img src="https://travis-ci.org/encode/httpcore.svg?branch=master" alt="Build Status">
+<a href="https://travis-ci.org/encode/http3">
+ <img src="https://travis-ci.org/encode/http3.svg?branch=master" alt="Build Status">
</a>
-<a href="https://codecov.io/gh/encode/httpcore">
- <img src="https://codecov.io/gh/encode/httpcore/branch/master/graph/badge.svg" alt="Coverage">
+<a href="https://codecov.io/gh/encode/http3">
+ <img src="https://codecov.io/gh/encode/http3/branch/master/graph/badge.svg" alt="Coverage">
</a>
-<a href="https://pypi.org/project/httpcore/">
- <img src="https://badge.fury.io/py/httpcore.svg" alt="Package version">
+<a href="https://pypi.org/project/http3/">
+ <img src="https://badge.fury.io/py/http3.svg" alt="Package version">
</a>
## Feature support
Making a request:
```python
->>> import httpcore
->>> client = httpcore.Client()
+>>> import http3
+>>> client = http3.Client()
>>> response = client.get('https://example.com')
>>> response.status_code
<HTTPStatus.OK: 200>
**Note**: Use `ipython` to try this from the console, since it supports `await`.
```python
->>> import httpcore
->>> client = httpcore.AsyncClient()
+>>> import http3
+>>> client = http3.AsyncClient()
>>> response = await client.get('https://example.com')
>>> response.status_code
<StatusCode.OK: 200>
*An HTTP client, with connection pooling, redirects, cookie persistence, etc.*
```python
->>> client = Client()
+>>> client = http3.Client()
>>> response = client.get('https://example.org')
```
what gets sent over the wire.*
```python
->>> request = Request("GET", "https://example.org", headers={'host': 'example.org'})
+>>> request = http3.Request("GET", "https://example.org", headers={'host': 'example.org'})
>>> response = client.send(request)
```
--- /dev/null
+# Developer Interface
+
+## Main Interface
+
+* `get(url, [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `options(url, [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `head(url, [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `post(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `put(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `patch(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `delete(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `request(method, url, [data], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+
+## `Client`
+
+*An HTTP client, with connection pooling, redirects, cookie persistence, etc.*
+
+```python
+>>> client = http3.Client()
+>>> response = client.get('https://example.org')
+```
+
+* `def __init__([auth], [cookies], [verify], [cert], [timeout], [pool_limits], [max_redirects], [dispatch])`
+* `def .get(url, [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .options(url, [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .head(url, [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .post(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .put(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .patch(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .delete(url, [data], [json], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .request(method, url, [data], [params], [headers], [cookies], [auth], [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .send(request, [stream], [allow_redirects], [verify], [cert], [timeout])`
+* `def .close()`
+
+## `Response`
+
+*An HTTP response.*
+
+* `def __init__(...)`
+* `.status_code` - **int** *(Typically a `StatusCode` IntEnum.)*
+* `.reason_phrase` - **str**
+* `.protocol` - `"HTTP/2"` or `"HTTP/1.1"`
+* `.url` - **URL**
+* `.headers` - **Headers**
+* `.content` - **bytes**
+* `.text` - **str**
+* `.encoding` - **str**
+* `.is_redirect` - **bool**
+* `.request` - **Request**
+* `.cookies` - **Cookies**
+* `.history` - **List[Response]**
+* `def .raise_for_status()` - **None**
+* `def .json()` - **Any**
+* `def .read()` - **bytes**
+* `def .stream()` - **bytes iterator**
+* `def .raw()` - **bytes iterator**
+* `def .close()` - **None**
+* `def .next()` - **Response**
+
+## `Request`
+
+*An HTTP request. Can be constructed explicitly for more control over exactly
+what gets sent over the wire.*
+
+```python
+>>> request = http3.Request("GET", "https://example.org", headers={'host': 'example.org'})
+>>> response = client.send(request)
+```
+
+* `def __init__(method, url, [params], [data], [json], [headers], [cookies])`
+* `.method` - **str**
+* `.url` - **URL**
+* `.content` - **byte** or **byte async iterator**
+* `.headers` - **Headers**
+* `.cookies` - **Cookies**
+
+## `URL`
+
+*A normalized, IDNA supporting URL.*
+
+```python
+>>> url = URL("https://example.org/")
+>>> url.host
+'example.org'
+```
+
+* `def __init__(url, allow_relative=False, params=None)`
+* `.scheme` - **str**
+* `.authority` - **str**
+* `.host` - **str**
+* `.port` - **int**
+* `.path` - **str**
+* `.query` - **str**
+* `.full_path` - **str**
+* `.fragment` - **str**
+* `.is_ssl` - **bool**
+* `.origin` - **Origin**
+* `.is_absolute_url` - **bool**
+* `.is_relative_url` - **bool**
+* `def .copy_with([scheme], [authority], [path], [query], [fragment])` - **URL**
+* `def .resolve_with(url)` - **URL**
+
+## `Origin`
+
+*A normalized, IDNA supporting set of scheme/host/port info.*
+
+```python
+>>> Origin('https://example.org') == Origin('HTTPS://EXAMPLE.ORG:443')
+True
+```
+
+* `def __init__(url)`
+* `.is_ssl` - **bool**
+* `.host` - **str**
+* `.port` - **int**
+
+## `Headers`
+
+*A case-insensitive multi-dict.*
+
+```python
+>>> headers = Headers({'Content-Type': 'application/json'})
+>>> headers['content-type']
+'application/json'
+```
+
+* `def __init__(self, headers)`
+
+## `Cookies`
+
+*A dict-like cookie store.*
+
+```python
+>>> cookies = Cookies()
+>>> cookies.set("name", "value", domain="example.org")
+```
+
+* `def __init__(cookies: [dict, Cookies, CookieJar])`
+* `.jar` - **CookieJar**
+* `def extract_cookies(response)`
+* `def set_cookie_header(request)`
+* `def set(name, value, [domain], [path])`
+* `def get(name, [domain], [path])`
+* `def delete(name, [domain], [path])`
+* `def clear([domain], [path])`
+* *Standard mutable mapping interface*
--- /dev/null
+# Async Client
+
+HTTP3 offers a standard synchronous API by default, but also gives you
+the option of an async client if you need it.
+
+Async is a concurrency model that is far more efficient than multi-threading,
+and can provide significant performance benefits and enable the use of
+long-lived network connections such as WebSockets.
+
+If you're working with an async web framework such as Sanic, Starlette, FastAPI,
+Responder or Bocadillo, then you'll also want to use an async client for sending
+outgoing HTTP requests.
+
+## Making Async requests
+
+To make asynchronous requests, you'll need an `AsyncClient`.
+
+```python
+>>> client = http3.AsyncClient()
+>>> r = await client.get('https://www.example.com/')
+```
+
+## API Differences
+
+If you're using streaming responses then there are a few bits of API that
+use async methods:
+
+```python
+>>> client = http3.AsyncClient()
+>>> r = await client.get('https://www.example.com/', stream=True)
+>>> try:
+>>> async for chunk in r.stream():
+>>> ...
+>>> finally:
+>>> await r.close()
+```
+
+The async response methods are:
+
+* `.read()`
+* `.stream()`
+* `.raw()`
+* `.close()`
+
+If you're making parallel requests, then you'll also need to use an async API:
+
+```python
+>>> client = http3.AsyncClient()
+>>> async with client.parallel() as parallel:
+>>> pending_one = parallel.get('https://example.com/1')
+>>> pending_two = parallel.get('https://example.com/2')
+>>> response_one = await pending_one.get_response()
+>>> response_two = await pending_two.get_response()
+```
+
+The async parallel methods are:
+
+* `.parallel()` *Used as an "async with" context manager.*
+* `.get_response()`
+* `.next_response()`
--- /dev/null
+# Requests Compatibility Guide
+
+HTTP3 aims to be compatible with the `requests` API wherever possible.
+
+This documentation outlines places where the API differs...
+
+**TODO**
--- /dev/null
+# HTTP3
+
+<a href="https://travis-ci.org/encode/http3">
+ <img src="https://travis-ci.org/encode/http3.svg?branch=master" alt="Build Status">
+</a>
+<a href="https://codecov.io/gh/encode/http3">
+ <img src="https://codecov.io/gh/encode/http3/branch/master/graph/badge.svg" alt="Coverage">
+</a>
+<a href="https://pypi.org/project/http3/">
+ <img src="https://badge.fury.io/py/http3.svg" alt="Package version">
+</a>
+
+HTTP3 is a next-generation HTTP client for Python.
+
+!!! warning
+ This project should be considered as an "alpha" release. It is substantially
+ API complete, but there are still some areas that need more work.
+
+---
+
+Let's get started...
+
+```python
+>>> import http3
+>>> r = http3.get('https://www.example.org/')
+>>> r.status_code
+<StatusCode.OK: 200>
+>>> r.protocol
+'HTTP/2'
+>>> r.headers['content-type']
+'text/html; charset=UTF-8'
+>>> r.text
+'<!doctype html>\n<html>\n<head>\n<title>Example Domain</title>...'
+```
+
+## Features
+
+HTTP3 builds on the well-established usability of `requests`, and gives you:
+
+* A requests-compatible API.
+* HTTP/2 and HTTP/1.1 support.
+* Support for issuing HTTP requests in parallel.
+* Standard synchronous interface, but with `async`/`await` support if you need it.
+* Strict timeouts everywhere.
+* Fully type annotated.
+* 100% test coverage.
+
+Plus all the standard features of `requests`...
+
+* International Domains and URLs
+* Keep-Alive & Connection Pooling
+* Sessions with Cookie Persistence
+* Browser-style SSL Verification
+* Basic/Digest Authentication *Digest is still TODO*
+* Elegant Key/Value Cookies
+* Automatic Decompression
+* Automatic Content Decoding
+* Unicode Response Bodies
+* Multipart File Uploads *TODO*
+* HTTP(S) Proxy Support *TODO*
+* Connection Timeouts
+* Streaming Downloads
+* .netrc Support *TODO*
+* Chunked Requests
+
+## Documentation
+
+For a run-through of all the basics, head over to the [QuickStart](quickstart.md).
+
+For more advanced topics, see the [Parallel Requests](parallel.md) or [Async Client](async.md) documentation.
+
+The [Developer Interface](api.md) provides a comprehensive API reference.
+
+## Dependencies
+
+The HTTP3 project relies on these excellent libraries:
+
+* `h2` - HTTP/2 support.
+* `h11` - HTTP/1.1 support.
+* `certifi` - SSL certificates.
+* `chardet` - Fallback auto-detection for response encoding.
+* `idna` - Internationalized domain name support.
+* `rfc3986` - URL parsing & normalization.
+* `brotlipy` - Decoding for "brotli" compressed responses. *(Optional)*
+
+A huge amount of credit is due to `requests` for the API layout that
+much of this work follows, as well as to `urllib3` for plenty of design
+inspiration around the lower level networking details.
--- /dev/null
+# Parallel Requests
+
+HTTP3 allows you to make HTTP requests in parallel in a highly efficient way,
+using async under the hood, while still presenting a standard threaded interface.
+
+This has the huge benefit of allowing you to efficiently make parallel HTTP
+requests without having to switch out to using async all the way through.
+
+## Making Parallel Requests
+
+Let's make two outgoing HTTP requests in parallel:
+
+```python
+>>> with http3.parallel() as parallel:
+>>> pending_one = parallel.get('https://example.com/1')
+>>> pending_two = parallel.get('https://example.com/2')
+>>> response_one = pending_one.get_response()
+>>> response_two = pending_two.get_response()
+```
+
+If we're making lots of outgoing requests, we might not want to deal with the
+responses sequentially, but rather deal with each response that comes back
+as soon as it's available:
+
+```python
+>>> with http3.parallel() as parallel:
+>>> for counter in range(1, 10):
+>>> parallel.get(f'https://example.com/{counter}')
+>>> while parallel.has_pending_responses:
+>>> r = parallel.next_response()
+```
+
+## Exceptions and Cancellations
+
+The style of using `parallel` blocks ensures that you'll always have well
+defined exception and cancellation behaviours. Request exceptions are only ever
+raised when calling either `get_response` or `next_response`, and any pending
+requests are cancelled on exiting the block.
+
+## Parallel requests with a Client
+
+You can also call `parallel()` from a client instance, which allows you to
+control the authentication or dispatch behaviour for all requests within the
+block.
+
+```python
+>>> client = http3.Client()
+>>> with client.parallel() as parallel:
+>>> ...
+```
+
+## Async parallel requests
+
+If you're working within an async framework, then you'll want to use a fully
+async API for making requests.
+
+```python
+>>> client = http3.AsyncClient()
+>>> async with client.parallel() as parallel:
+>>> pending_one = await parallel.get('https://example.com/1')
+>>> pending_two = await parallel.get('https://example.com/2')
+>>> response_one = await pending_one.get_response()
+>>> response_two = await pending_two.get_response()
+```
+
+See [the Async Client documentation](async.md) for more details.
--- /dev/null
+# QuickStart
+
+!!! note
+ This page closely follows the layout of the `requests` QuickStart documentation.
+ The `http3` library is designed to be API compatible with `requests` wherever
+ possible.
+
+First start by importing HTTP3:
+
+```
+>>> import http3
+```
+
+Now, let’s try to get a webpage. For this example, let’s get GitHub’s public timeline:
+
+```python
+>>> r = http3.get('https://api.github.com/events')
+```
+
+Similarly, to make an HTTP POST request:
+
+```python
+>>> r = http3.post('https://httpbin.org/post', data={'key': 'value'})
+```
+
+The PUT, DELETE, HEAD, and OPTIONS requests all follow the same style:
+
+```python
+>>> r = http3.put('https://httpbin.org/put', data={'key': 'value'})
+>>> r = http3.delete('https://httpbin.org/delete')
+>>> r = http3.head('https://httpbin.org/get')
+>>> r = http3.options('https://httpbin.org/get')
+```
+
+## Passing Parameters in URLs
+
+To include URL query parameters in the request, use the `params` keyword:
+
+```python
+>>> params = {'key1': 'value1', 'key2': 'value2'}
+>>> r = http3.get('https://httpbin.org/get', params=params)
+```
+
+To see how the values get encoding into the URL string, we can inspect the
+resulting URL that was used to make the request:
+
+```python
+>>> r.url
+URL('https://httpbin.org/get?key2=value2&key1=value1')
+```
+
+You can also pass a list of items as a value:
+
+```python
+>>> params = {'key1': 'value1', 'key2': ['value2', 'value3']}
+>>> r = http3.get('https://httpbin.org/get', params=params)
+>>> r.url
+URL('https://httpbin.org/get?key1=value1&key2=value2&key2=value3')
+```
+
+## Response Content
+
+HTTP3 will automatically handle decoding the response content into unicode text.
+
+```python
+>>> r = http3.get('https://www.example.org/')
+>>> r.text
+'<!doctype html>\n<html>\n<head>\n<title>Example Domain</title>...'
+```
+
+You can inspect what encoding has been used to decode the response.
+
+```python
+>>> r.encoding
+'UTF-8'
+```
+
+If you need to override the standard behavior and explicitly set the encoding to
+use, then you can do that too.
+
+```python
+>>> r.encoding = 'ISO-8859-1'
+```
+
+## Binary Response Content
+
+The response content can also be accessed as bytes, for non-text responses:
+
+```python
+>>> r.content
+b'<!doctype html>\n<html>\n<head>\n<title>Example Domain</title>...'
+```
+
+Any `gzip` and `deflate` HTTP response encodings will automatically
+be decoded for you. If `brotlipy` is installed, then the `brotli` response
+encoding will also be supported.
+
+For example, to create an image from binary data returned by a request, you can use the following code:
+
+```python
+>>> from PIL import Image
+>>> from io import BytesIO
+>>> i = Image.open(BytesIO(r.content))
+```
+
+## JSON Response Content
+
+Often Web API responses will be encoded as JSON.
+
+```python
+>>> r = http3.get('https://api.github.com/events')
+>>> r.json()
+[{u'repository': {u'open_issues': 0, u'url': 'https://github.com/...' ... }}]
+```
+
+## Custom Headers
+
+To include additional headers in the outgoing request, use the `headers` keyword argument:
+
+```python
+>>> url = 'http://httpbin.org/headers'
+>>> headers = {'user-agent': 'my-app/0.0.1'}
+>>> r = http3.get(url, headers=headers)
+```
+
+## Sending Form Encoded Data
+
+Some types of HTTP requests, such as `POST` and `PUT` requests, can include data
+in the request body. One common way of including that is as form encoded data,
+which is used for HTML forms.
+
+```python
+>>> data = {'key1': 'value1', 'key2': 'value2'}
+>>> r = http3.post("https://httpbin.org/post", data=data)
+>>> print(r.text)
+{
+ ...
+ "form": {
+ "key2": "value2",
+ "key1": "value1"
+ },
+ ...
+}
+```
+
+Form encoded data can also include multiple values form a given key.
+
+```python
+>>> data = {'key1': ['value1', 'value2']}
+>>> r = http3.post("https://httpbin.org/post", data=data)
+>>> print(r.text)
+{
+ ...
+ "form": {
+ "key1": [
+ "value1",
+ "value2"
+ ]
+ },
+ ...
+}
+```
+
+## Sending JSON Encoded Data
+
+Form encoded data is okay if all you need is simple key-value data structure.
+For more complicated data structures you'll often want to use JSON encoding instead.
+
+```python
+>>> data = {'integer': 123, 'boolean': True, 'list': ['a', 'b', 'c']}
+>>> r = http3.post("https://httpbin.org/post", json=data)
+>>> print(r.text)
+{
+ ...
+ "json": {
+ "boolean": true,
+ "integer": 123,
+ "list": [
+ "a",
+ "b",
+ "c"
+ ]
+ },
+ ...
+}
+```
+
+## Sending Binary Request Data
+
+For other encodings you should use either a `bytes` type, or a generator
+that yields `bytes`.
+
+You'll probably also want to set a custom `Content-Type` header when uploading
+binary data.
+
+## Response Status Codes
+
+We can inspect the HTTP status code of the response:
+
+```python
+>>> r = http3.get('https://httpbin.org/get')
+>>> r.status_code
+<StatusCode.OK: 200>
+```
+
+The status code is an integer enum, meaning that the Python representation gives
+use some descriptive information, but the value itself can be used as a regular integer.
+
+```python
+>>> r.status_code == 200
+True
+```
+
+HTTP3 also includes an easy shortcut for accessing status codes by their text phrase.
+
+```python
+>>> r.status_code == requests.codes.OK
+True
+```
+
+We can raise an exception for any Client or Server error responses (4xx or 5xx status codes):
+
+```python
+>>> not_found = http3.get('https://httpbin.org/status/404')
+>>> not_found.status_code
+<StatusCode.NOT_FOUND: 404>
+>>> not_found.raise_for_status()
+Traceback (most recent call last):
+ File "/Users/tomchristie/GitHub/encode/httpcore/http3/models.py", line 776, in raise_for_status
+ raise HttpError(message)
+http3.exceptions.HttpError: 404 Not Found
+```
+
+Any successful response codes will simply return `None` rather than raising an exception.
+
+``` python
+>>> r.raise_for_status()
+```
+
+## Response Headers
+
+The response headers are available as a dictionary-like interface.
+
+```python
+>>> r.headers
+Headers({
+ 'content-encoding': 'gzip',
+ 'transfer-encoding': 'chunked',
+ 'connection': 'close',
+ 'server': 'nginx/1.0.4',
+ 'x-runtime': '148ms',
+ 'etag': '"e1ca502697e5c9317743dc078f67693f"',
+ 'content-type': 'application/json'
+})
+```
+
+The `Headers` data type is case-insensitive, so you can use any capitalization.
+
+```python
+>>> r.headers['Content-Type']
+'application/json'
+
+>>> r.headers.get('content-type')
+'application/json'
+```
+
+Multiple values for a single response header are represented as a single comma separated
+value, as per [RFC 7230](https://tools.ietf.org/html/rfc7230#section-3.2):
+
+> A recipient MAY combine multiple header fields with the same field name into one “field-name: field-value” pair, without changing the semantics of the message, by appending each subsequent field value to the combined field value in order, separated by a comma.
)
from .status_codes import StatusCode, codes
-__version__ = "0.4.0"
+__version__ = "0.0.1"
import typing
import zlib
-import httpcore.exceptions
+from .exceptions import DecodingError
try:
import brotli
try:
return self.decompressor.decompress(data)
except zlib.error as exc:
- raise httpcore.exceptions.DecodingError from exc
+ raise DecodingError from exc
def flush(self) -> bytes:
try:
return self.decompressor.flush()
except zlib.error as exc: # pragma: nocover
- raise httpcore.exceptions.DecodingError from exc
+ raise DecodingError from exc
class GZipDecoder(Decoder):
try:
return self.decompressor.decompress(data)
except zlib.error as exc:
- raise httpcore.exceptions.DecodingError from exc
+ raise DecodingError from exc
def flush(self) -> bytes:
try:
return self.decompressor.flush()
except zlib.error as exc: # pragma: nocover
- raise httpcore.exceptions.DecodingError from exc
+ raise DecodingError from exc
class BrotliDecoder(Decoder):
try:
return self.decompressor.decompress(data)
except brotli.Error as exc:
- raise httpcore.exceptions.DecodingError from exc
+ raise DecodingError from exc
def flush(self) -> bytes:
try:
self.decompressor.finish()
return b""
except brotli.Error as exc: # pragma: nocover
- raise httpcore.exceptions.DecodingError from exc
+ raise DecodingError from exc
class MultiDecoder(Decoder):
has_accept_encoding = "accept-encoding" in self.headers
if not has_user_agent:
- auto_headers.append((b"user-agent", b"httpcore"))
+ auto_headers.append((b"user-agent", b"http3"))
if not has_accept:
auto_headers.append((b"accept", b"*/*"))
if not has_content_length:
codes = StatusCode
+
+# Include lower-case styles for `requests` compatability.
+for code in codes:
+ setattr(codes, code._name_.lower(), int(code))
--- /dev/null
+site_name: HTTP3
+site_description: The next generation HTTP client.
+
+theme:
+ name: 'material'
+
+repo_name: encode/http3
+repo_url: https://github.com/encode/http3
+edit_uri: ""
+
+nav:
+ - Introduction: 'index.md'
+ - QuickStart: 'quickstart.md'
+ - Parallel Requests: 'parallel.md'
+ - Async Client: 'async.md'
+ - Requests Compatibility: 'compatibility.md'
+ - Developer Interface: 'api.md'
+
+markdown_extensions:
+ - admonition
+ - codehilite
if [ -d 'htmlcov' ] ; then
rm -r htmlcov
fi
-if [ -d 'httpcore.egg-info' ] ; then
- rm -r httpcore.egg-info
+if [ -d 'http3.egg-info' ] ; then
+ rm -r http3.egg-info
fi
set -x
-${PREFIX}autoflake --in-place --recursive httpcore tests
-${PREFIX}black httpcore tests
-${PREFIX}isort --multi-line=3 --trailing-comma --force-grid-wrap=0 --combine-as --line-width 88 --recursive --apply httpcore tests
-${PREFIX}mypy httpcore --ignore-missing-imports --disallow-untyped-defs
+${PREFIX}autoflake --in-place --recursive http3 tests
+${PREFIX}black http3 tests
+${PREFIX}isort --multi-line=3 --trailing-comma --force-grid-wrap=0 --combine-as --line-width 88 --recursive --apply http3 tests
+${PREFIX}mypy http3 --ignore-missing-imports --disallow-untyped-defs
scripts/clean
#!/bin/sh -e
-export PACKAGE="httpcore"
+export PACKAGE="http3"
export VERSION=`cat ${PACKAGE}/__init__.py | grep __version__ | sed "s/__version__ = //" | sed "s/'//g"`
export PREFIX=""
if [ -d 'venv' ] ; then
#!/bin/sh -e
-export PACKAGE="httpcore"
+export PACKAGE="http3"
export PREFIX=""
if [ -d 'venv' ] ; then
export PREFIX="venv/bin/"
setup(
- name="httpcore",
+ name="http3",
python_requires=">=3.6",
- version=get_version("httpcore"),
- url="https://github.com/encode/httpcore",
+ version=get_version("http3"),
+ url="https://github.com/encode/http3",
license="BSD",
- description="...",
+ description="The next generation HTTP client.",
long_description=get_long_description(),
long_description_content_type="text/markdown",
author="Tom Christie",
author_email="tom@tomchristie.com",
- packages=get_packages("httpcore"),
+ packages=get_packages("http3"),
data_files=[("", ["LICENSE.md"])],
install_requires=[
"certifi",
import pytest
-import httpcore
+import http3
@pytest.mark.asyncio
async def test_get(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.get(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
@pytest.mark.asyncio
async def test_post(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.post(url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_post_json(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.post(url, json={"text": "Hello, world!"})
assert response.status_code == 200
@pytest.mark.asyncio
async def test_stream_response(server):
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
body = await response.read()
@pytest.mark.asyncio
async def test_access_content_stream_response(server):
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.request("GET", "http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
- with pytest.raises(httpcore.ResponseNotRead):
+ with pytest.raises(http3.ResponseNotRead):
response.content
yield b"Hello, "
yield b"world!"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.request(
"POST", "http://127.0.0.1:8000/", data=hello_world()
)
@pytest.mark.asyncio
async def test_raise_for_status(server):
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
for status_code in (200, 400, 404, 500, 505):
response = await client.request(
"GET", f"http://127.0.0.1:8000/status/{status_code}"
)
if 400 <= status_code < 600:
- with pytest.raises(httpcore.exceptions.HttpError):
+ with pytest.raises(http3.exceptions.HttpError):
response.raise_for_status()
else:
assert response.raise_for_status() is None
@pytest.mark.asyncio
async def test_options(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.options(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
@pytest.mark.asyncio
async def test_head(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.head(url)
assert response.status_code == 200
assert response.text == ""
@pytest.mark.asyncio
async def test_put(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.put(url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_patch(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.patch(url, data=b"Hello, world!")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_delete(server):
url = "http://127.0.0.1:8000/"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.delete(url)
assert response.status_code == 200
assert response.text == "Hello, world!"
headers = {"Expect": "100-continue"}
data = b"Echo request body"
- async with httpcore.AsyncClient() as client:
+ async with http3.AsyncClient() as client:
response = await client.post(url, headers=headers, data=data)
assert response.status_code == 200
import pytest
-from httpcore import (
+from http3 import (
URL,
AsyncDispatcher,
AsyncRequest,
import pytest
-import httpcore
+import http3
def threadpool(func):
@threadpool
def test_get(server):
url = "http://127.0.0.1:8000/"
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.get(url)
assert response.status_code == 200
- assert response.url == httpcore.URL(url)
+ assert response.url == http3.URL(url)
assert response.content == b"Hello, world!"
assert response.text == "Hello, world!"
assert response.protocol == "HTTP/1.1"
assert response.encoding == "iso-8859-1"
- assert response.request.url == httpcore.URL(url)
+ assert response.request.url == http3.URL(url)
assert response.headers
assert response.is_redirect is False
assert repr(response) == "<Response(200, 'OK')>"
@threadpool
def test_post(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.post("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_post_json(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.post("http://127.0.0.1:8000/", json={"text": "Hello, world!"})
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_stream_response(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.get("http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
content = response.read()
@threadpool
def test_stream_iterator(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.get("http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
body = b""
@threadpool
def test_raw_iterator(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.get("http://127.0.0.1:8000/", stream=True)
assert response.status_code == 200
body = b""
@threadpool
def test_raise_for_status(server):
- with httpcore.Client() as client:
+ with http3.Client() as client:
for status_code in (200, 400, 404, 500, 505):
response = client.request(
"GET", "http://127.0.0.1:8000/status/{}".format(status_code)
)
if 400 <= status_code < 600:
- with pytest.raises(httpcore.exceptions.HttpError):
+ with pytest.raises(http3.exceptions.HttpError):
response.raise_for_status()
else:
assert response.raise_for_status() is None
@threadpool
def test_options(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.options("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_head(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.head("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_put(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.put("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_patch(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_delete(server):
- with httpcore.Client() as http:
+ with http3.Client() as http:
response = http.delete("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
import pytest
-from httpcore import (
+from http3 import (
URL,
AsyncDispatcher,
AsyncRequest,
import pytest
-from httpcore import (
+from http3 import (
URL,
AsyncClient,
AsyncDispatcher,
import pytest
-import httpcore
+import http3
@pytest.mark.asyncio
"""
Connections should default to staying in a keep-alive state.
"""
- async with httpcore.ConnectionPool() as http:
+ async with http3.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
assert len(http.active_connections) == 0
"""
Connnections to differing connection keys should result in multiple connections.
"""
- async with httpcore.ConnectionPool() as http:
+ async with http3.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
assert len(http.active_connections) == 0
"""
The soft_limit config should limit the maximum number of keep-alive connections.
"""
- pool_limits = httpcore.PoolLimits(soft_limit=1)
+ pool_limits = http3.PoolLimits(soft_limit=1)
- async with httpcore.ConnectionPool(pool_limits=pool_limits) as http:
+ async with http3.ConnectionPool(pool_limits=pool_limits) as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
assert len(http.active_connections) == 0
"""
A streaming request should hold the connection open until the response is read.
"""
- async with httpcore.ConnectionPool() as http:
+ async with http3.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
assert len(http.active_connections) == 1
assert len(http.keepalive_connections) == 0
"""
Multiple conncurrent requests should open multiple conncurrent connections.
"""
- async with httpcore.ConnectionPool() as http:
+ async with http3.ConnectionPool() as http:
response_a = await http.request("GET", "http://127.0.0.1:8000/")
assert len(http.active_connections) == 1
assert len(http.keepalive_connections) == 0
Using a `Connection: close` header should close the connection.
"""
headers = [(b"connection", b"close")]
- async with httpcore.ConnectionPool() as http:
+ async with http3.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/", headers=headers)
await response.read()
assert len(http.active_connections) == 0
"""
A standard close should keep the connection open.
"""
- async with httpcore.ConnectionPool() as http:
+ async with http3.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.read()
await response.close()
"""
A premature close should close the connection.
"""
- async with httpcore.ConnectionPool() as http:
+ async with http3.ConnectionPool() as http:
response = await http.request("GET", "http://127.0.0.1:8000/")
await response.close()
assert len(http.active_connections) == 0
import pytest
-from httpcore import HTTPConnection, Request, SSLConfig
+from http3 import HTTPConnection, Request, SSLConfig
@pytest.mark.asyncio
import pytest
-from httpcore import Client, Response
+from http3 import Client, Response
from .utils import MockHTTP2Backend
import pytest
-from httpcore import (
+from http3 import (
CertTypes,
Client,
Dispatcher,
import h2.connection
import h2.events
-from httpcore import (
+from http3 import (
AsyncioBackend,
BaseReader,
BaseWriter,
import pytest
-from httpcore import CookieConflict, Cookies
+from http3 import CookieConflict, Cookies
def test_cookies():
-import httpcore
+import http3
def test_headers():
- h = httpcore.Headers([("a", "123"), ("a", "456"), ("b", "789")])
+ h = http3.Headers([("a", "123"), ("a", "456"), ("b", "789")])
assert "a" in h
assert "A" in h
assert "b" in h
assert list(h) == ["a", "a", "b"]
assert dict(h) == {"a": "123, 456", "b": "789"}
assert repr(h) == "Headers([('a', '123'), ('a', '456'), ('b', '789')])"
- assert h == httpcore.Headers([("a", "123"), ("b", "789"), ("a", "456")])
+ assert h == http3.Headers([("a", "123"), ("b", "789"), ("a", "456")])
assert h != [("a", "123"), ("A", "456"), ("b", "789")]
- h = httpcore.Headers({"a": "123", "b": "789"})
+ h = http3.Headers({"a": "123", "b": "789"})
assert h["A"] == "123"
assert h["B"] == "789"
assert h.raw == [(b"a", b"123"), (b"b", b"789")]
def test_header_mutations():
- h = httpcore.Headers()
+ h = http3.Headers()
assert dict(h) == {}
h["a"] = "1"
assert dict(h) == {"a": "1"}
def test_copy_headers():
- headers = httpcore.Headers({"custom": "example"})
- headers_copy = httpcore.Headers(headers)
+ headers = http3.Headers({"custom": "example"})
+ headers_copy = http3.Headers(headers)
assert headers == headers_copy
def test_headers_insert_retains_ordering():
- headers = httpcore.Headers({"a": "a", "b": "b", "c": "c"})
+ headers = http3.Headers({"a": "a", "b": "b", "c": "c"})
headers["b"] = "123"
assert list(headers.values()) == ["a", "123", "c"]
def test_headers_insert_appends_if_new():
- headers = httpcore.Headers({"a": "a", "b": "b", "c": "c"})
+ headers = http3.Headers({"a": "a", "b": "b", "c": "c"})
headers["d"] = "123"
assert list(headers.values()) == ["a", "b", "c", "123"]
def test_headers_insert_removes_all_existing():
- headers = httpcore.Headers([("a", "123"), ("a", "456")])
+ headers = http3.Headers([("a", "123"), ("a", "456")])
headers["a"] = "789"
assert dict(headers) == {"a": "789"}
def test_headers_delete_removes_all_existing():
- headers = httpcore.Headers([("a", "123"), ("a", "456")])
+ headers = http3.Headers([("a", "123"), ("a", "456")])
del headers["a"]
assert dict(headers) == {}
"""
Headers should display with a dict repr by default.
"""
- headers = httpcore.Headers({"custom": "example"})
+ headers = http3.Headers({"custom": "example"})
assert repr(headers) == "Headers({'custom': 'example'})"
"""
Headers should display an encoding in the repr if required.
"""
- headers = httpcore.Headers({b"custom": "example ☃".encode("utf-8")})
+ headers = http3.Headers({b"custom": "example ☃".encode("utf-8")})
assert repr(headers) == "Headers({'custom': 'example ☃'}, encoding='utf-8')"
"""
Headers should display with a list repr if they include multiple identical keys.
"""
- headers = httpcore.Headers([("custom", "example 1"), ("custom", "example 2")])
+ headers = http3.Headers([("custom", "example 1"), ("custom", "example 2")])
assert (
repr(headers) == "Headers([('custom', 'example 1'), ('custom', 'example 2')])"
)
Headers should decode as ascii by default.
"""
raw_headers = [(b"Custom", b"Example")]
- headers = httpcore.Headers(raw_headers)
+ headers = http3.Headers(raw_headers)
assert dict(headers) == {"custom": "Example"}
assert headers.encoding == "ascii"
Headers containing non-ascii codepoints should default to decoding as utf-8.
"""
raw_headers = [(b"Custom", "Code point: ☃".encode("utf-8"))]
- headers = httpcore.Headers(raw_headers)
+ headers = http3.Headers(raw_headers)
assert dict(headers) == {"custom": "Code point: ☃"}
assert headers.encoding == "utf-8"
Headers containing non-UTF-8 codepoints should default to decoding as iso-8859-1.
"""
raw_headers = [(b"Custom", "Code point: ÿ".encode("iso-8859-1"))]
- headers = httpcore.Headers(raw_headers)
+ headers = http3.Headers(raw_headers)
assert dict(headers) == {"custom": "Code point: ÿ"}
assert headers.encoding == "iso-8859-1"
particular decoding.
"""
raw_headers = [(b"Custom", "Code point: ☃".encode("utf-8"))]
- headers = httpcore.Headers(raw_headers)
+ headers = http3.Headers(raw_headers)
headers.encoding = "iso-8859-1"
assert dict(headers) == {"custom": "Code point: â\x98\x83"}
assert headers.encoding == "iso-8859-1"
"""
Most headers should split by commas for `getlist`, except 'Set-Cookie'.
"""
- h = httpcore.Headers([("set-cookie", "a, b"), ("set-cookie", "c")])
+ h = http3.Headers([("set-cookie", "a, b"), ("set-cookie", "c")])
h.getlist("Set-Cookie") == ["a, b", "b"]
- h = httpcore.Headers([("vary", "a, b"), ("vary", "c")])
+ h = http3.Headers([("vary", "a, b"), ("vary", "c")])
h.getlist("Vary") == ["a", "b", "c"]
-from httpcore import QueryParams
+from http3 import QueryParams
def test_queryparams():
import pytest
-import httpcore
+import http3
def test_request_repr():
- request = httpcore.Request("GET", "http://example.org")
+ request = http3.Request("GET", "http://example.org")
assert repr(request) == "<Request('GET', 'http://example.org')>"
def test_no_content():
- request = httpcore.Request("GET", "http://example.org")
+ request = http3.Request("GET", "http://example.org")
assert "Content-Length" not in request.headers
def test_content_length_header():
- request = httpcore.Request("POST", "http://example.org", data=b"test 123")
+ request = http3.Request("POST", "http://example.org", data=b"test 123")
assert request.headers["Content-Length"] == "8"
def test_url_encoded_data():
- for RequestClass in (httpcore.Request, httpcore.AsyncRequest):
+ for RequestClass in (http3.Request, http3.AsyncRequest):
request = RequestClass("POST", "http://example.org", data={"test": "123"})
assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"
assert request.content == b"test=123"
def test_json_encoded_data():
- for RequestClass in (httpcore.Request, httpcore.AsyncRequest):
+ for RequestClass in (http3.Request, http3.AsyncRequest):
request = RequestClass("POST", "http://example.org", json={"test": 123})
assert request.headers["Content-Type"] == "application/json"
assert request.content == b'{"test": 123}'
data = streaming_body(b"test 123")
- request = httpcore.Request("POST", "http://example.org", data=data)
+ request = http3.Request("POST", "http://example.org", data=data)
assert "Content-Length" not in request.headers
assert request.headers["Transfer-Encoding"] == "chunked"
def test_override_host_header():
headers = {"host": "1.2.3.4:80"}
- request = httpcore.Request("GET", "http://example.org", headers=headers)
+ request = http3.Request("GET", "http://example.org", headers=headers)
assert request.headers["Host"] == "1.2.3.4:80"
def test_override_accept_encoding_header():
headers = {"Accept-Encoding": "identity"}
- request = httpcore.Request("GET", "http://example.org", headers=headers)
+ request = http3.Request("GET", "http://example.org", headers=headers)
assert request.headers["Accept-Encoding"] == "identity"
data = streaming_body(b"test 123")
headers = {"Content-Length": "8"}
- request = httpcore.Request("POST", "http://example.org", data=data, headers=headers)
+ request = http3.Request("POST", "http://example.org", data=data, headers=headers)
assert request.headers["Content-Length"] == "8"
def test_url():
url = "http://example.org"
- request = httpcore.Request("GET", url)
+ request = http3.Request("GET", url)
assert request.url.scheme == "http"
assert request.url.port == 80
assert request.url.full_path == "/"
url = "https://example.org/abc?foo=bar"
- request = httpcore.Request("GET", url)
+ request = http3.Request("GET", url)
assert request.url.scheme == "https"
assert request.url.port == 443
assert request.url.full_path == "/abc?foo=bar"
def test_invalid_urls():
- with pytest.raises(httpcore.InvalidURL):
- httpcore.Request("GET", "example.org")
+ with pytest.raises(http3.InvalidURL):
+ http3.Request("GET", "example.org")
- with pytest.raises(httpcore.InvalidURL):
- httpcore.Request("GET", "invalid://example.org")
+ with pytest.raises(http3.InvalidURL):
+ http3.Request("GET", "invalid://example.org")
- with pytest.raises(httpcore.InvalidURL):
- httpcore.Request("GET", "http:///foo")
+ with pytest.raises(http3.InvalidURL):
+ http3.Request("GET", "http:///foo")
import pytest
-import httpcore
+import http3
def streaming_body():
def test_response():
- response = httpcore.Response(200, content=b"Hello, world!")
+ response = http3.Response(200, content=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
assert response.text == "Hello, world!"
def test_response_repr():
- response = httpcore.Response(200, content=b"Hello, world!")
+ response = http3.Response(200, content=b"Hello, world!")
assert repr(response) == "<Response(200, 'OK')>"
"""
headers = {"Content-Type": "text-plain; charset=latin-1"}
content = "Latin 1: ÿ".encode("latin-1")
- response = httpcore.Response(200, content=content, headers=headers)
+ response = http3.Response(200, content=content, headers=headers)
assert response.text == "Latin 1: ÿ"
assert response.encoding == "latin-1"
Autodetect encoding if there is no charset info in a Content-Type header.
"""
content = "おはようございます。".encode("EUC-JP")
- response = httpcore.Response(200, content=content)
+ response = http3.Response(200, content=content)
assert response.text == "おはようございます。"
assert response.encoding == "EUC-JP"
"""
headers = {"Content-Type": "text-plain; charset=invalid-codec-name"}
content = "おはようございます。".encode("EUC-JP")
- response = httpcore.Response(200, content=content, headers=headers)
+ response = http3.Response(200, content=content, headers=headers)
assert response.text == "おはようございます。"
assert response.encoding == "EUC-JP"
"""
content = b"Hello, world!"
headers = {"Content-Type": "text/plain"}
- response = httpcore.Response(200, content=content, headers=headers)
+ response = http3.Response(200, content=content, headers=headers)
assert response.status_code == 200
assert response.encoding == "iso-8859-1"
assert response.text == "Hello, world!"
"""
Default to utf-8 if all else fails.
"""
- response = httpcore.Response(200, content=b"")
+ response = http3.Response(200, content=b"")
assert response.text == ""
assert response.encoding == "utf-8"
Default to apparent encoding for non-text content-type headers.
"""
headers = {"Content-Type": "image/png"}
- response = httpcore.Response(200, content=b"xyz", headers=headers)
+ response = http3.Response(200, content=b"xyz", headers=headers)
assert response.text == "xyz"
assert response.encoding == "ascii"
headers = {
"Content-Type": "text-plain; charset=utf-8"
} # Deliberately incorrect charset
- response = httpcore.Response(
+ response = http3.Response(
200, content="Latin 1: ÿ".encode("latin-1"), headers=headers
)
response.encoding = "latin-1"
def test_response_force_encoding():
- response = httpcore.Response(200, content="Snowman: ☃".encode("utf-8"))
+ response = http3.Response(200, content="Snowman: ☃".encode("utf-8"))
response.encoding = "iso-8859-1"
assert response.status_code == 200
assert response.reason_phrase == "OK"
def test_read_response():
- response = httpcore.Response(200, content=b"Hello, world!")
+ response = http3.Response(200, content=b"Hello, world!")
assert response.status_code == 200
assert response.text == "Hello, world!"
def test_raw_interface():
- response = httpcore.Response(200, content=b"Hello, world!")
+ response = http3.Response(200, content=b"Hello, world!")
raw = b""
for part in response.raw():
def test_stream_interface():
- response = httpcore.Response(200, content=b"Hello, world!")
+ response = http3.Response(200, content=b"Hello, world!")
content = b""
for part in response.stream():
@pytest.mark.asyncio
async def test_async_stream_interface():
- response = httpcore.AsyncResponse(200, content=b"Hello, world!")
+ response = http3.AsyncResponse(200, content=b"Hello, world!")
content = b""
async for part in response.stream():
def test_stream_interface_after_read():
- response = httpcore.Response(200, content=b"Hello, world!")
+ response = http3.Response(200, content=b"Hello, world!")
response.read()
@pytest.mark.asyncio
async def test_async_stream_interface_after_read():
- response = httpcore.AsyncResponse(200, content=b"Hello, world!")
+ response = http3.AsyncResponse(200, content=b"Hello, world!")
await response.read()
def test_streaming_response():
- response = httpcore.Response(200, content=streaming_body())
+ response = http3.Response(200, content=streaming_body())
assert response.status_code == 200
assert not response.is_closed
@pytest.mark.asyncio
async def test_async_streaming_response():
- response = httpcore.AsyncResponse(200, content=async_streaming_body())
+ response = http3.AsyncResponse(200, content=async_streaming_body())
assert response.status_code == 200
assert not response.is_closed
def test_cannot_read_after_stream_consumed():
- response = httpcore.Response(200, content=streaming_body())
+ response = http3.Response(200, content=streaming_body())
content = b""
for part in response.stream():
content += part
- with pytest.raises(httpcore.StreamConsumed):
+ with pytest.raises(http3.StreamConsumed):
response.read()
@pytest.mark.asyncio
async def test_async_cannot_read_after_stream_consumed():
- response = httpcore.AsyncResponse(200, content=async_streaming_body())
+ response = http3.AsyncResponse(200, content=async_streaming_body())
content = b""
async for part in response.stream():
content += part
- with pytest.raises(httpcore.StreamConsumed):
+ with pytest.raises(http3.StreamConsumed):
await response.read()
def test_cannot_read_after_response_closed():
- response = httpcore.Response(200, content=streaming_body())
+ response = http3.Response(200, content=streaming_body())
response.close()
- with pytest.raises(httpcore.ResponseClosed):
+ with pytest.raises(http3.ResponseClosed):
response.read()
@pytest.mark.asyncio
async def test_async_cannot_read_after_response_closed():
- response = httpcore.AsyncResponse(200, content=async_streaming_body())
+ response = http3.AsyncResponse(200, content=async_streaming_body())
await response.close()
- with pytest.raises(httpcore.ResponseClosed):
+ with pytest.raises(http3.ResponseClosed):
await response.read()
def test_unknown_status_code():
- response = httpcore.Response(600)
+ response = http3.Response(600)
assert response.status_code == 600
assert response.reason_phrase == ""
assert response.text == ""
-from httpcore import URL
+from http3 import URL
def test_idna_url():
import pytest
-import httpcore
+import http3
def threadpool(func):
@threadpool
def test_get(server):
- response = httpcore.get("http://127.0.0.1:8000/")
+ response = http3.get("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
assert response.text == "Hello, world!"
@threadpool
def test_post(server):
- response = httpcore.post("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = http3.post("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
yield b", "
yield b"world!"
- response = httpcore.post("http://127.0.0.1:8000/", data=data())
+ response = http3.post("http://127.0.0.1:8000/", data=data())
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_options(server):
- response = httpcore.options("http://127.0.0.1:8000/")
+ response = http3.options("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_head(server):
- response = httpcore.head("http://127.0.0.1:8000/")
+ response = http3.head("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_put(server):
- response = httpcore.put("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = http3.put("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_patch(server):
- response = httpcore.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
+ response = http3.patch("http://127.0.0.1:8000/", data=b"Hello, world!")
assert response.status_code == 200
assert response.reason_phrase == "OK"
@threadpool
def test_delete(server):
- response = httpcore.delete("http://127.0.0.1:8000/")
+ response = http3.delete("http://127.0.0.1:8000/")
assert response.status_code == 200
assert response.reason_phrase == "OK"
import pytest
-import httpcore
+import http3
@pytest.mark.asyncio
async def test_load_ssl_config():
- ssl_config = httpcore.SSLConfig()
+ ssl_config = http3.SSLConfig()
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
@pytest.mark.asyncio
async def test_load_ssl_config_verify_non_existing_path():
- ssl_config = httpcore.SSLConfig(verify="/path/to/nowhere")
+ ssl_config = http3.SSLConfig(verify="/path/to/nowhere")
with pytest.raises(IOError):
await ssl_config.load_ssl_context()
@pytest.mark.asyncio
async def test_load_ssl_config_verify_existing_file():
- ssl_config = httpcore.SSLConfig(verify=httpcore.config.DEFAULT_CA_BUNDLE_PATH)
+ ssl_config = http3.SSLConfig(verify=http3.config.DEFAULT_CA_BUNDLE_PATH)
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
@pytest.mark.asyncio
async def test_load_ssl_config_verify_directory():
- path = os.path.dirname(httpcore.config.DEFAULT_CA_BUNDLE_PATH)
- ssl_config = httpcore.SSLConfig(verify=path)
+ path = os.path.dirname(http3.config.DEFAULT_CA_BUNDLE_PATH)
+ ssl_config = http3.SSLConfig(verify=path)
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
@pytest.mark.asyncio
async def test_load_ssl_config_cert_and_key(cert_and_key_paths):
cert_path, key_path = cert_and_key_paths
- ssl_config = httpcore.SSLConfig(cert=(cert_path, key_path))
+ ssl_config = http3.SSLConfig(cert=(cert_path, key_path))
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_REQUIRED
@pytest.mark.asyncio
async def test_load_ssl_config_cert_without_key_raises(cert_and_key_paths):
cert_path, _ = cert_and_key_paths
- ssl_config = httpcore.SSLConfig(cert=cert_path)
+ ssl_config = http3.SSLConfig(cert=cert_path)
with pytest.raises(ssl.SSLError):
await ssl_config.load_ssl_context()
@pytest.mark.asyncio
async def test_load_ssl_config_no_verify(verify=False):
- ssl_config = httpcore.SSLConfig(verify=False)
+ ssl_config = http3.SSLConfig(verify=False)
context = await ssl_config.load_ssl_context()
assert context.verify_mode == ssl.VerifyMode.CERT_NONE
def test_ssl_repr():
- ssl = httpcore.SSLConfig(verify=False)
+ ssl = http3.SSLConfig(verify=False)
assert repr(ssl) == "SSLConfig(cert=None, verify=False)"
def test_timeout_repr():
- timeout = httpcore.TimeoutConfig(timeout=5.0)
+ timeout = http3.TimeoutConfig(timeout=5.0)
assert repr(timeout) == "TimeoutConfig(timeout=5.0)"
- timeout = httpcore.TimeoutConfig(read_timeout=5.0)
+ timeout = http3.TimeoutConfig(read_timeout=5.0)
assert (
repr(timeout)
== "TimeoutConfig(connect_timeout=None, read_timeout=5.0, write_timeout=None)"
def test_limits_repr():
- limits = httpcore.PoolLimits(hard_limit=100)
+ limits = http3.PoolLimits(hard_limit=100)
assert (
repr(limits) == "PoolLimits(soft_limit=None, hard_limit=100, pool_timeout=None)"
)
def test_ssl_eq():
- ssl = httpcore.SSLConfig(verify=False)
- assert ssl == httpcore.SSLConfig(verify=False)
+ ssl = http3.SSLConfig(verify=False)
+ assert ssl == http3.SSLConfig(verify=False)
def test_timeout_eq():
- timeout = httpcore.TimeoutConfig(timeout=5.0)
- assert timeout == httpcore.TimeoutConfig(timeout=5.0)
+ timeout = http3.TimeoutConfig(timeout=5.0)
+ assert timeout == http3.TimeoutConfig(timeout=5.0)
def test_limits_eq():
- limits = httpcore.PoolLimits(hard_limit=100)
- assert limits == httpcore.PoolLimits(hard_limit=100)
+ limits = http3.PoolLimits(hard_limit=100)
+ assert limits == http3.PoolLimits(hard_limit=100)
def test_timeout_from_tuple():
- timeout = httpcore.TimeoutConfig(timeout=(5.0, 5.0, 5.0))
- assert timeout == httpcore.TimeoutConfig(timeout=5.0)
+ timeout = http3.TimeoutConfig(timeout=(5.0, 5.0, 5.0))
+ assert timeout == http3.TimeoutConfig(timeout=5.0)
def test_timeout_from_config_instance():
- timeout = httpcore.TimeoutConfig(timeout=(5.0))
- assert httpcore.TimeoutConfig(timeout) == httpcore.TimeoutConfig(timeout=5.0)
+ timeout = http3.TimeoutConfig(timeout=(5.0))
+ assert http3.TimeoutConfig(timeout) == http3.TimeoutConfig(timeout=5.0)
import brotli
import pytest
-import httpcore
+import http3
def test_deflate():
compressed_body = compressor.compress(body) + compressor.flush()
headers = [(b"Content-Encoding", b"deflate")]
- response = httpcore.Response(200, headers=headers, content=compressed_body)
+ response = http3.Response(200, headers=headers, content=compressed_body)
assert response.content == body
compressed_body = compressor.compress(body) + compressor.flush()
headers = [(b"Content-Encoding", b"gzip")]
- response = httpcore.Response(200, headers=headers, content=compressed_body)
+ response = http3.Response(200, headers=headers, content=compressed_body)
assert response.content == body
compressed_body = brotli.compress(body)
headers = [(b"Content-Encoding", b"br")]
- response = httpcore.Response(200, headers=headers, content=compressed_body)
+ response = http3.Response(200, headers=headers, content=compressed_body)
assert response.content == body
)
headers = [(b"Content-Encoding", b"deflate, gzip")]
- response = httpcore.Response(200, headers=headers, content=compressed_body)
+ response = http3.Response(200, headers=headers, content=compressed_body)
assert response.content == body
compressed_body = brotli.compress(body)
headers = [(b"Content-Encoding", b"br, identity")]
- response = httpcore.Response(200, headers=headers, content=compressed_body)
+ response = http3.Response(200, headers=headers, content=compressed_body)
assert response.content == body
headers = [(b"Content-Encoding", b"identity, br")]
- response = httpcore.Response(200, headers=headers, content=compressed_body)
+ response = http3.Response(200, headers=headers, content=compressed_body)
assert response.content == body
yield compressor.flush()
headers = [(b"Content-Encoding", b"gzip")]
- response = httpcore.Response(200, headers=headers, content=compress(body))
+ response = http3.Response(200, headers=headers, content=compress(body))
assert not hasattr(response, "body")
assert response.read() == body
headers = [(b"Content-Encoding", header_value)]
body = b"test 123"
compressed_body = brotli.compress(body)[3:]
- with pytest.raises(httpcore.exceptions.DecodingError):
- response = httpcore.Response(200, headers=headers, content=compressed_body)
+ with pytest.raises(http3.exceptions.DecodingError):
+ response = http3.Response(200, headers=headers, content=compressed_body)
response.content
import pytest
-from httpcore import (
+from http3 import (
AsyncClient,
ConnectTimeout,
PoolLimits,
projects / thirdparty / httpx.git / commitdiff
