Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_fragment_redirect():
client = Client(dispatch=MockDispatch())
response = await client.get("https://example.org/relative_redirect#fragment")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/#fragment")
assert len(response.history) == 1
def test_url():
url = URL("https://example.org:123/path/to/somewhere?abc=123#anchor")
assert url.scheme == "https"
assert url.host == "example.org"
assert url.port == 123
assert url.authority == "example.org:123"
assert url.path == "/path/to/somewhere"
assert url.query == "abc=123"
assert url.fragment == "anchor"
assert (
repr(url) == "URL('https://example.org:123/path/to/somewhere?abc=123#anchor')"
)
new = url.copy_with(scheme="http")
assert new == URL("http://example.org:123/path/to/somewhere?abc=123#anchor")
assert new.scheme == "http"
async def test_disallow_redirects():
client = Client(dispatch=MockDispatch())
response = await client.post(
"https://example.org/redirect_303", allow_redirects=False
)
assert response.status_code == codes.SEE_OTHER
assert response.url == URL("https://example.org/redirect_303")
assert response.is_redirect is True
assert len(response.history) == 0
response = await response.next()
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
assert response.is_redirect is False
assert len(response.history) == 1
async def test_cross_subdomain_redirect():
client = Client(dispatch=MockDispatch())
url = "https://example.com/cross_subdomain"
response = await client.get(url)
assert response.url == URL("https://www.example.org/cross_subdomain")
async def test_relative_redirect():
client = Client(dispatch=MockDispatch())
response = await client.get("https://example.org/relative_redirect")
assert response.status_code == codes.OK
assert response.url == URL("https://example.org/")
assert len(response.history) == 1
def test_url_eq_str():
url = URL("https://example.org:123/path/to/somewhere?abc=123#anchor")
assert url == "https://example.org:123/path/to/somewhere?abc=123#anchor"
assert str(url) == url
def test_url_full_path_setter():
url = URL("http://example.org")
url.full_path = "http://example.net"
assert url.full_path == "http://example.net"
def test_url_join():
"""
Some basic URL joining tests.
"""
url = URL("https://example.org:123/path/to/somewhere")
assert url.join("/somewhere-else") == "https://example.org:123/somewhere-else"
assert (
url.join("somewhere-else") == "https://example.org:123/path/to/somewhere-else"
)
assert (
url.join("../somewhere-else") == "https://example.org:123/path/somewhere-else"
)
assert url.join("../../somewhere-else") == "https://example.org:123/somewhere-else"
async def test_body_redirect():
"""
A 308 redirect should preserve the request body.
"""
client = Client(dispatch=MockDispatch())
url = "https://example.org/redirect_body"
data = b"Example request body"
response = await client.post(url, data=data)
assert response.url == URL("https://example.org/redirect_body_target")
assert response.json()["body"] == "Example request body"
assert "content-length" in response.json()["headers"]
async def auth_call(self, request, get_response, has_method=True):
content = await request.read()
if has_method:
url, headers, body = self.prepare(
request.method, str(request.url), request.headers, content)
else:
url, headers, body = self.prepare(
str(request.url), request.headers, content)
request.url = URL(url)
request.headers.update(headers)
if body:
body = to_bytes(body)
if body != content:
request.is_streaming = False
request.content = body
return await get_response(request)