Gecko [ www.smanegeri1koba.sch.id ]

Name	Size	Permission
certs	[ DIR ]	drwxr-xr-x
testserver	[ DIR ]	drwxr-xr-x
__init__.py	469 B	-rw-r--r--
compat.py	480 B	-rw-r--r--
conftest.py	1.54 KB	-rw-r--r--
test_adapters.py	311 B	-rw-r--r--
test_help.py	839 B	-rw-r--r--
test_hooks.py	424 B	-rw-r--r--
test_lowlevel.py	14.98 KB	-rw-r--r--
test_packages.py	229 B	-rw-r--r--
test_requests.py	102.05 KB	-rw-r--r--
test_structures.py	2.23 KB	-rw-r--r--
test_testserver.py	5.42 KB	-rw-r--r--
test_utils.py	28.67 KB	-rw-r--r--
utils.py	366 B	-rw-r--r--

Code Editor : test_requests.py

"""Tests for Requests."""

import collections
import contextlib
import io
import json
import os
import pickle
import re
import threading
import warnings
from unittest import mock

import pytest
import urllib3
from urllib3.util import Timeout as Urllib3Timeout

import requests
from requests.adapters import HTTPAdapter
from requests.auth import HTTPDigestAuth, _basic_auth_str
from requests.compat import (
    JSONDecodeError,
    Morsel,
    MutableMapping,
    builtin_str,
    cookielib,
    getproxies,
    urlparse,
)
from requests.cookies import cookiejar_from_dict, morsel_to_cookie
from requests.exceptions import (
    ChunkedEncodingError,
    ConnectionError,
    ConnectTimeout,
    ContentDecodingError,
    InvalidHeader,
    InvalidProxyURL,
    InvalidSchema,
    InvalidURL,
    MissingSchema,
    ProxyError,
    ReadTimeout,
    RequestException,
    RetryError,
)
from requests.exceptions import SSLError as RequestsSSLError
from requests.exceptions import Timeout, TooManyRedirects, UnrewindableBodyError
from requests.hooks import default_hooks
from requests.models import PreparedRequest, urlencode
from requests.sessions import SessionRedirectMixin
from requests.structures import CaseInsensitiveDict

from . import SNIMissingWarning
from .compat import StringIO
from .testserver.server import TLSServer, consume_socket_content
from .utils import override_environ

# Requests to this URL should always fail with a connection timeout (nothing
# listening on that port)
TARPIT = "http://10.255.255.1"

# This is to avoid waiting the timeout of using TARPIT
INVALID_PROXY = "http://localhost:1"

try:
    from ssl import SSLContext

del SSLContext
    HAS_MODERN_SSL = True
except ImportError:
    HAS_MODERN_SSL = False

try:
    requests.pyopenssl
    HAS_PYOPENSSL = True
except AttributeError:
    HAS_PYOPENSSL = False

class TestRequests:
    digest_auth_algo = ("MD5", "SHA-256", "SHA-512")

def test_entry_points(self):
        requests.session
        requests.session().get
        requests.session().head
        requests.get
        requests.head
        requests.put
        requests.patch
        requests.post
        # Not really an entry point, but people rely on it.
        from requests.packages.urllib3.poolmanager import PoolManager  # noqa:F401

@pytest.mark.parametrize(
        "exception, url",
        (
            (MissingSchema, "hiwpefhipowhefopw"),
            (InvalidSchema, "localhost:3128"),
            (InvalidSchema, "localhost.localdomain:3128/"),
            (InvalidSchema, "10.122.1.1:3128/"),
            (InvalidURL, "http://"),
            (InvalidURL, "http://*example.com"),
            (InvalidURL, "http://.example.com"),
        ),
    )
    def test_invalid_url(self, exception, url):
        with pytest.raises(exception):
            requests.get(url)

def test_basic_building(self):
        req = requests.Request()
        req.url = "http://kennethreitz.org/"
        req.data = {"life": "42"}

pr = req.prepare()
        assert pr.url == req.url
        assert pr.body == "life=42"

@pytest.mark.parametrize("method", ("GET", "HEAD"))
    def test_no_content_length(self, httpbin, method):
        req = requests.Request(method, httpbin(method.lower())).prepare()
        assert "Content-Length" not in req.headers

@pytest.mark.parametrize("method", ("POST", "PUT", "PATCH", "OPTIONS"))
    def test_no_body_content_length(self, httpbin, method):
        req = requests.Request(method, httpbin(method.lower())).prepare()
        assert req.headers["Content-Length"] == "0"

@pytest.mark.parametrize("method", ("POST", "PUT", "PATCH", "OPTIONS"))
    def test_empty_content_length(self, httpbin, method):
        req = requests.Request(method, httpbin(method.lower()), data="").prepare()
        assert req.headers["Content-Length"] == "0"

def test_override_content_length(self, httpbin):
        headers = {"Content-Length": "not zero"}
        r = requests.Request("POST", httpbin("post"), headers=headers).prepare()
        assert "Content-Length" in r.headers
        assert r.headers["Content-Length"] == "not zero"

def test_path_is_not_double_encoded(self):
        request = requests.Request("GET", "http://0.0.0.0/get/test case").prepare()

assert request.path_url == "/get/test%20case"

@pytest.mark.parametrize(
        "url, expected",
        (
            (
                "http://example.com/path#fragment",
                "http://example.com/path?a=b#fragment",
            ),
            (
                "http://example.com/path?key=value#fragment",
                "http://example.com/path?key=value&a=b#fragment",
            ),
        ),
    )
    def test_params_are_added_before_fragment(self, url, expected):
        request = requests.Request("GET", url, params={"a": "b"}).prepare()
        assert request.url == expected

def test_params_original_order_is_preserved_by_default(self):
        param_ordered_dict = collections.OrderedDict(
            (("z", 1), ("a", 1), ("k", 1), ("d", 1))
        )
        session = requests.Session()
        request = requests.Request(
            "GET", "http://example.com/", params=param_ordered_dict
        )
        prep = session.prepare_request(request)
        assert prep.url == "http://example.com/?z=1&a=1&k=1&d=1"

def test_params_bytes_are_encoded(self):
        request = requests.Request(
            "GET", "http://example.com", params=b"test=foo"
        ).prepare()
        assert request.url == "http://example.com/?test=foo"

def test_binary_put(self):
        request = requests.Request(
            "PUT", "http://example.com", data="ööö".encode()
        ).prepare()
        assert isinstance(request.body, bytes)

def test_whitespaces_are_removed_from_url(self):
        # Test for issue #3696
        request = requests.Request("GET", " http://example.com").prepare()
        assert request.url == "http://example.com/"

@pytest.mark.parametrize("scheme", ("http://", "HTTP://", "hTTp://", "HttP://"))
    def test_mixed_case_scheme_acceptable(self, httpbin, scheme):
        s = requests.Session()
        s.proxies = getproxies()
        parts = urlparse(httpbin("get"))
        url = scheme + parts.netloc + parts.path
        r = requests.Request("GET", url)
        r = s.send(r.prepare())
        assert r.status_code == 200, f"failed for scheme {scheme}"

def test_HTTP_200_OK_GET_ALTERNATIVE(self, httpbin):
        r = requests.Request("GET", httpbin("get"))
        s = requests.Session()
        s.proxies = getproxies()

r = s.send(r.prepare())

assert r.status_code == 200

def test_HTTP_302_ALLOW_REDIRECT_GET(self, httpbin):
        r = requests.get(httpbin("redirect", "1"))
        assert r.status_code == 200
        assert r.history[0].status_code == 302
        assert r.history[0].is_redirect

def test_HTTP_307_ALLOW_REDIRECT_POST(self, httpbin):
        r = requests.post(
            httpbin("redirect-to"),
            data="test",
            params={"url": "post", "status_code": 307},
        )
        assert r.status_code == 200
        assert r.history[0].status_code == 307
        assert r.history[0].is_redirect
        assert r.json()["data"] == "test"

def test_HTTP_307_ALLOW_REDIRECT_POST_WITH_SEEKABLE(self, httpbin):
        byte_str = b"test"
        r = requests.post(
            httpbin("redirect-to"),
            data=io.BytesIO(byte_str),
            params={"url": "post", "status_code": 307},
        )
        assert r.status_code == 200
        assert r.history[0].status_code == 307
        assert r.history[0].is_redirect
        assert r.json()["data"] == byte_str.decode("utf-8")

def test_HTTP_302_TOO_MANY_REDIRECTS(self, httpbin):
        try:
            requests.get(httpbin("relative-redirect", "50"))
        except TooManyRedirects as e:
            url = httpbin("relative-redirect", "20")
            assert e.request.url == url
            assert e.response.url == url
            assert len(e.response.history) == 30
        else:
            pytest.fail("Expected redirect to raise TooManyRedirects but it did not")

def test_HTTP_302_TOO_MANY_REDIRECTS_WITH_PARAMS(self, httpbin):
        s = requests.session()
        s.max_redirects = 5
        try:
            s.get(httpbin("relative-redirect", "50"))
        except TooManyRedirects as e:
            url = httpbin("relative-redirect", "45")
            assert e.request.url == url
            assert e.response.url == url
            assert len(e.response.history) == 5
        else:
            pytest.fail(
                "Expected custom max number of redirects to be respected but was not"
            )

def test_http_301_changes_post_to_get(self, httpbin):
        r = requests.post(httpbin("status", "301"))
        assert r.status_code == 200
        assert r.request.method == "GET"
        assert r.history[0].status_code == 301
        assert r.history[0].is_redirect

def test_http_301_doesnt_change_head_to_get(self, httpbin):
        r = requests.head(httpbin("status", "301"), allow_redirects=True)
        print(r.content)
        assert r.status_code == 200
        assert r.request.method == "HEAD"
        assert r.history[0].status_code == 301
        assert r.history[0].is_redirect

def test_http_302_changes_post_to_get(self, httpbin):
        r = requests.post(httpbin("status", "302"))
        assert r.status_code == 200
        assert r.request.method == "GET"
        assert r.history[0].status_code == 302
        assert r.history[0].is_redirect

def test_http_302_doesnt_change_head_to_get(self, httpbin):
        r = requests.head(httpbin("status", "302"), allow_redirects=True)
        assert r.status_code == 200
        assert r.request.method == "HEAD"
        assert r.history[0].status_code == 302
        assert r.history[0].is_redirect

def test_http_303_changes_post_to_get(self, httpbin):
        r = requests.post(httpbin("status", "303"))
        assert r.status_code == 200
        assert r.request.method == "GET"
        assert r.history[0].status_code == 303
        assert r.history[0].is_redirect

def test_http_303_doesnt_change_head_to_get(self, httpbin):
        r = requests.head(httpbin("status", "303"), allow_redirects=True)
        assert r.status_code == 200
        assert r.request.method == "HEAD"
        assert r.history[0].status_code == 303
        assert r.history[0].is_redirect

def test_header_and_body_removal_on_redirect(self, httpbin):
        purged_headers = ("Content-Length", "Content-Type")
        ses = requests.Session()
        req = requests.Request("POST", httpbin("post"), data={"test": "data"})
        prep = ses.prepare_request(req)
        resp = ses.send(prep)

# Mimic a redirect response
        resp.status_code = 302
        resp.headers["location"] = "get"

# Run request through resolve_redirects
        next_resp = next(ses.resolve_redirects(resp, prep))
        assert next_resp.request.body is None
        for header in purged_headers:
            assert header not in next_resp.request.headers

def test_transfer_enc_removal_on_redirect(self, httpbin):
        purged_headers = ("Transfer-Encoding", "Content-Type")
        ses = requests.Session()
        req = requests.Request("POST", httpbin("post"), data=(b"x" for x in range(1)))
        prep = ses.prepare_request(req)
        assert "Transfer-Encoding" in prep.headers

# Create Response to avoid https://github.com/kevin1024/pytest-httpbin/issues/33
        resp = requests.Response()
        resp.raw = io.BytesIO(b"the content")
        resp.request = prep
        setattr(resp.raw, "release_conn", lambda *args: args)

# Mimic a redirect response
        resp.status_code = 302
        resp.headers["location"] = httpbin("get")

# Run request through resolve_redirect
        next_resp = next(ses.resolve_redirects(resp, prep))
        assert next_resp.request.body is None
        for header in purged_headers:
            assert header not in next_resp.request.headers

def test_fragment_maintained_on_redirect(self, httpbin):
        fragment = "#view=edit&token=hunter2"
        r = requests.get(httpbin("redirect-to?url=get") + fragment)

assert len(r.history) > 0
        assert r.history[0].request.url == httpbin("redirect-to?url=get") + fragment
        assert r.url == httpbin("get") + fragment

def test_HTTP_200_OK_GET_WITH_PARAMS(self, httpbin):
        heads = {"User-agent": "Mozilla/5.0"}

r = requests.get(httpbin("user-agent"), headers=heads)

assert heads["User-agent"] in r.text
        assert r.status_code == 200

def test_HTTP_200_OK_GET_WITH_MIXED_PARAMS(self, httpbin):
        heads = {"User-agent": "Mozilla/5.0"}

r = requests.get(
            httpbin("get") + "?test=true", params={"q": "test"}, headers=heads
        )
        assert r.status_code == 200

def test_set_cookie_on_301(self, httpbin):
        s = requests.session()
        url = httpbin("cookies/set?foo=bar")
        s.get(url)
        assert s.cookies["foo"] == "bar"

def test_cookie_sent_on_redirect(self, httpbin):
        s = requests.session()
        s.get(httpbin("cookies/set?foo=bar"))
        r = s.get(httpbin("redirect/1"))  # redirects to httpbin('get')
        assert "Cookie" in r.json()["headers"]

def test_cookie_removed_on_expire(self, httpbin):
        s = requests.session()
        s.get(httpbin("cookies/set?foo=bar"))
        assert s.cookies["foo"] == "bar"
        s.get(
            httpbin("response-headers"),
            params={"Set-Cookie": "foo=deleted; expires=Thu, 01-Jan-1970 00:00:01 GMT"},
        )
        assert "foo" not in s.cookies

def test_cookie_quote_wrapped(self, httpbin):
        s = requests.session()
        s.get(httpbin('cookies/set?foo="bar:baz"'))
        assert s.cookies["foo"] == '"bar:baz"'

def test_cookie_persists_via_api(self, httpbin):
        s = requests.session()
        r = s.get(httpbin("redirect/1"), cookies={"foo": "bar"})
        assert "foo" in r.request.headers["Cookie"]
        assert "foo" in r.history[0].request.headers["Cookie"]

def test_request_cookie_overrides_session_cookie(self, httpbin):
        s = requests.session()
        s.cookies["foo"] = "bar"
        r = s.get(httpbin("cookies"), cookies={"foo": "baz"})
        assert r.json()["cookies"]["foo"] == "baz"
        # Session cookie should not be modified
        assert s.cookies["foo"] == "bar"

def test_request_cookies_not_persisted(self, httpbin):
        s = requests.session()
        s.get(httpbin("cookies"), cookies={"foo": "baz"})
        # Sending a request with cookies should not add cookies to the session
        assert not s.cookies

def test_generic_cookiejar_works(self, httpbin):
        cj = cookielib.CookieJar()
        cookiejar_from_dict({"foo": "bar"}, cj)
        s = requests.session()
        s.cookies = cj
        r = s.get(httpbin("cookies"))
        # Make sure the cookie was sent
        assert r.json()["cookies"]["foo"] == "bar"
        # Make sure the session cj is still the custom one
        assert s.cookies is cj

def test_param_cookiejar_works(self, httpbin):
        cj = cookielib.CookieJar()
        cookiejar_from_dict({"foo": "bar"}, cj)
        s = requests.session()
        r = s.get(httpbin("cookies"), cookies=cj)
        # Make sure the cookie was sent
        assert r.json()["cookies"]["foo"] == "bar"

def test_cookielib_cookiejar_on_redirect(self, httpbin):
        """Tests resolve_redirect doesn't fail when merging cookies
        with non-RequestsCookieJar cookiejar.

See GH #3579
        """
        cj = cookiejar_from_dict({"foo": "bar"}, cookielib.CookieJar())
        s = requests.Session()
        s.cookies = cookiejar_from_dict({"cookie": "tasty"})

# Prepare request without using Session
        req = requests.Request("GET", httpbin("headers"), cookies=cj)
        prep_req = req.prepare()

# Send request and simulate redirect
        resp = s.send(prep_req)
        resp.status_code = 302
        resp.headers["location"] = httpbin("get")
        redirects = s.resolve_redirects(resp, prep_req)
        resp = next(redirects)

# Verify CookieJar isn't being converted to RequestsCookieJar
        assert isinstance(prep_req._cookies, cookielib.CookieJar)
        assert isinstance(resp.request._cookies, cookielib.CookieJar)
        assert not isinstance(resp.request._cookies, requests.cookies.RequestsCookieJar)

cookies = {}
        for c in resp.request._cookies:
            cookies[c.name] = c.value
        assert cookies["foo"] == "bar"
        assert cookies["cookie"] == "tasty"

def test_requests_in_history_are_not_overridden(self, httpbin):
        resp = requests.get(httpbin("redirect/3"))
        urls = [r.url for r in resp.history]
        req_urls = [r.request.url for r in resp.history]
        assert urls == req_urls

def test_history_is_always_a_list(self, httpbin):
        """Show that even with redirects, Response.history is always a list."""
        resp = requests.get(httpbin("get"))
        assert isinstance(resp.history, list)
        resp = requests.get(httpbin("redirect/1"))
        assert isinstance(resp.history, list)
        assert not isinstance(resp.history, tuple)

def test_headers_on_session_with_None_are_not_sent(self, httpbin):
        """Do not send headers in Session.headers with None values."""
        ses = requests.Session()
        ses.headers["Accept-Encoding"] = None
        req = requests.Request("GET", httpbin("get"))
        prep = ses.prepare_request(req)
        assert "Accept-Encoding" not in prep.headers

def test_headers_preserve_order(self, httpbin):
        """Preserve order when headers provided as OrderedDict."""
        ses = requests.Session()
        ses.headers = collections.OrderedDict()
        ses.headers["Accept-Encoding"] = "identity"
        ses.headers["First"] = "1"
        ses.headers["Second"] = "2"
        headers = collections.OrderedDict([("Third", "3"), ("Fourth", "4")])
        headers["Fifth"] = "5"
        headers["Second"] = "222"
        req = requests.Request("GET", httpbin("get"), headers=headers)
        prep = ses.prepare_request(req)
        items = list(prep.headers.items())
        assert items[0] == ("Accept-Encoding", "identity")
        assert items[1] == ("First", "1")
        assert items[2] == ("Second", "222")
        assert items[3] == ("Third", "3")
        assert items[4] == ("Fourth", "4")
        assert items[5] == ("Fifth", "5")

@pytest.mark.parametrize("key", ("User-agent", "user-agent"))
    def test_user_agent_transfers(self, httpbin, key):
        heads = {key: "Mozilla/5.0 (github.com/psf/requests)"}

r = requests.get(httpbin("user-agent"), headers=heads)
        assert heads[key] in r.text

def test_HTTP_200_OK_HEAD(self, httpbin):
        r = requests.head(httpbin("get"))
        assert r.status_code == 200

def test_HTTP_200_OK_PUT(self, httpbin):
        r = requests.put(httpbin("put"))
        assert r.status_code == 200

def test_BASICAUTH_TUPLE_HTTP_200_OK_GET(self, httpbin):
        auth = ("user", "pass")
        url = httpbin("basic-auth", "user", "pass")

r = requests.get(url, auth=auth)
        assert r.status_code == 200

r = requests.get(url)
        assert r.status_code == 401

s = requests.session()
        s.auth = auth
        r = s.get(url)
        assert r.status_code == 200

@pytest.mark.parametrize(
        "username, password",
        (
            ("user", "pass"),
            ("имя".encode(), "пароль".encode()),
            (42, 42),
            (None, None),
        ),
    )
    def test_set_basicauth(self, httpbin, username, password):
        auth = (username, password)
        url = httpbin("get")

r = requests.Request("GET", url, auth=auth)
        p = r.prepare()

assert p.headers["Authorization"] == _basic_auth_str(username, password)

def test_basicauth_encodes_byte_strings(self):
        """Ensure b'test' formats as the byte string "test" rather
        than the unicode string "b'test'" in Python 3.
        """
        auth = (b"\xc5\xafsername", b"test\xc6\xb6")
        r = requests.Request("GET", "http://localhost", auth=auth)
        p = r.prepare()

assert p.headers["Authorization"] == "Basic xa9zZXJuYW1lOnRlc3TGtg=="

@pytest.mark.parametrize(
        "url, exception",
        (
            # Connecting to an unknown domain should raise a ConnectionError
            ("http://doesnotexist.google.com", ConnectionError),
            # Connecting to an invalid port should raise a ConnectionError
            ("http://localhost:1", ConnectionError),
            # Inputing a URL that cannot be parsed should raise an InvalidURL error
            ("http://fe80::5054:ff:fe5a:fc0", InvalidURL),
        ),
    )
    def test_errors(self, url, exception):
        with pytest.raises(exception):
            requests.get(url, timeout=1)

def test_proxy_error(self):
        # any proxy related error (address resolution, no route to host, etc) should result in a ProxyError
        with pytest.raises(ProxyError):
            requests.get(
                "http://localhost:1", proxies={"http": "non-resolvable-address"}
            )

def test_proxy_error_on_bad_url(self, httpbin, httpbin_secure):
        with pytest.raises(InvalidProxyURL):
            requests.get(httpbin_secure(), proxies={"https": "http:/badproxyurl:3128"})

with pytest.raises(InvalidProxyURL):
            requests.get(httpbin(), proxies={"http": "http://:8080"})

with pytest.raises(InvalidProxyURL):
            requests.get(httpbin_secure(), proxies={"https": "https://"})

with pytest.raises(InvalidProxyURL):
            requests.get(httpbin(), proxies={"http": "http:///example.com:8080"})

def test_respect_proxy_env_on_send_self_prepared_request(self, httpbin):
        with override_environ(http_proxy=INVALID_PROXY):
            with pytest.raises(ProxyError):
                session = requests.Session()
                request = requests.Request("GET", httpbin())
                session.send(request.prepare())

def test_respect_proxy_env_on_send_session_prepared_request(self, httpbin):
        with override_environ(http_proxy=INVALID_PROXY):
            with pytest.raises(ProxyError):
                session = requests.Session()
                request = requests.Request("GET", httpbin())
                prepared = session.prepare_request(request)
                session.send(prepared)

def test_respect_proxy_env_on_send_with_redirects(self, httpbin):
        with override_environ(http_proxy=INVALID_PROXY):
            with pytest.raises(ProxyError):
                session = requests.Session()
                url = httpbin("redirect/1")
                print(url)
                request = requests.Request("GET", url)
                session.send(request.prepare())

def test_respect_proxy_env_on_get(self, httpbin):
        with override_environ(http_proxy=INVALID_PROXY):
            with pytest.raises(ProxyError):
                session = requests.Session()
                session.get(httpbin())

def test_respect_proxy_env_on_request(self, httpbin):
        with override_environ(http_proxy=INVALID_PROXY):
            with pytest.raises(ProxyError):
                session = requests.Session()
                session.request(method="GET", url=httpbin())

def test_proxy_authorization_preserved_on_request(self, httpbin):
        proxy_auth_value = "Bearer XXX"
        session = requests.Session()
        session.headers.update({"Proxy-Authorization": proxy_auth_value})
        resp = session.request(method="GET", url=httpbin("get"))
        sent_headers = resp.json().get("headers", {})

assert sent_headers.get("Proxy-Authorization") == proxy_auth_value

@pytest.mark.parametrize(
        "url,has_proxy_auth",
        (
            ("http://example.com", True),
            ("https://example.com", False),
        ),
    )
    def test_proxy_authorization_not_appended_to_https_request(
        self, url, has_proxy_auth
    ):
        session = requests.Session()
        proxies = {
            "http": "http://test:pass@localhost:8080",
            "https": "http://test:pass@localhost:8090",
        }
        req = requests.Request("GET", url)
        prep = req.prepare()
        session.rebuild_proxies(prep, proxies)

assert ("Proxy-Authorization" in prep.headers) is has_proxy_auth

def test_basicauth_with_netrc(self, httpbin):
        auth = ("user", "pass")
        wrong_auth = ("wronguser", "wrongpass")
        url = httpbin("basic-auth", "user", "pass")

old_auth = requests.sessions.get_netrc_auth

try:

def get_netrc_auth_mock(url):
                return auth

requests.sessions.get_netrc_auth = get_netrc_auth_mock

# Should use netrc and work.
            r = requests.get(url)
            assert r.status_code == 200

# Given auth should override and fail.
            r = requests.get(url, auth=wrong_auth)
            assert r.status_code == 401

s = requests.session()

# Should use netrc and work.
            r = s.get(url)
            assert r.status_code == 200

# Given auth should override and fail.
            s.auth = wrong_auth
            r = s.get(url)
            assert r.status_code == 401
        finally:
            requests.sessions.get_netrc_auth = old_auth

def test_DIGEST_HTTP_200_OK_GET(self, httpbin):
        for authtype in self.digest_auth_algo:
            auth = HTTPDigestAuth("user", "pass")
            url = httpbin("digest-auth", "auth", "user", "pass", authtype, "never")

r = requests.get(url, auth=auth)
            assert r.status_code == 200

r = requests.get(url)
            assert r.status_code == 401
            print(r.headers["WWW-Authenticate"])

s = requests.session()
            s.auth = HTTPDigestAuth("user", "pass")
            r = s.get(url)
            assert r.status_code == 200

def test_DIGEST_AUTH_RETURNS_COOKIE(self, httpbin):
        for authtype in self.digest_auth_algo:
            url = httpbin("digest-auth", "auth", "user", "pass", authtype)
            auth = HTTPDigestAuth("user", "pass")
            r = requests.get(url)
            assert r.cookies["fake"] == "fake_value"

r = requests.get(url, auth=auth)
            assert r.status_code == 200

def test_DIGEST_AUTH_SETS_SESSION_COOKIES(self, httpbin):
        for authtype in self.digest_auth_algo:
            url = httpbin("digest-auth", "auth", "user", "pass", authtype)
            auth = HTTPDigestAuth("user", "pass")
            s = requests.Session()
            s.get(url, auth=auth)
            assert s.cookies["fake"] == "fake_value"

def test_DIGEST_STREAM(self, httpbin):
        for authtype in self.digest_auth_algo:
            auth = HTTPDigestAuth("user", "pass")
            url = httpbin("digest-auth", "auth", "user", "pass", authtype)

r = requests.get(url, auth=auth, stream=True)
            assert r.raw.read() != b""

r = requests.get(url, auth=auth, stream=False)
            assert r.raw.read() == b""

def test_DIGESTAUTH_WRONG_HTTP_401_GET(self, httpbin):
        for authtype in self.digest_auth_algo:
            auth = HTTPDigestAuth("user", "wrongpass")
            url = httpbin("digest-auth", "auth", "user", "pass", authtype)

r = requests.get(url, auth=auth)
            assert r.status_code == 401

r = requests.get(url)
            assert r.status_code == 401

s = requests.session()
            s.auth = auth
            r = s.get(url)
            assert r.status_code == 401

def test_DIGESTAUTH_QUOTES_QOP_VALUE(self, httpbin):
        for authtype in self.digest_auth_algo:
            auth = HTTPDigestAuth("user", "pass")
            url = httpbin("digest-auth", "auth", "user", "pass", authtype)

r = requests.get(url, auth=auth)
            assert '"auth"' in r.request.headers["Authorization"]

def test_POSTBIN_GET_POST_FILES(self, httpbin):
        url = httpbin("post")
        requests.post(url).raise_for_status()

post1 = requests.post(url, data={"some": "data"})
        assert post1.status_code == 200

with open("requirements-dev.txt") as f:
            post2 = requests.post(url, files={"some": f})
        assert post2.status_code == 200

post4 = requests.post(url, data='[{"some": "json"}]')
        assert post4.status_code == 200

with pytest.raises(ValueError):
            requests.post(url, files=["bad file data"])

def test_invalid_files_input(self, httpbin):
        url = httpbin("post")
        post = requests.post(url, files={"random-file-1": None, "random-file-2": 1})
        assert b'name="random-file-1"' not in post.request.body
        assert b'name="random-file-2"' in post.request.body

def test_POSTBIN_SEEKED_OBJECT_WITH_NO_ITER(self, httpbin):
        class TestStream:
            def __init__(self, data):
                self.data = data.encode()
                self.length = len(self.data)
                self.index = 0

def __len__(self):
                return self.length

def read(self, size=None):
                if size:
                    ret = self.data[self.index : self.index + size]
                    self.index += size
                else:
                    ret = self.data[self.index :]
                    self.index = self.length
                return ret

def tell(self):
                return self.index

def seek(self, offset, where=0):
                if where == 0:
                    self.index = offset
                elif where == 1:
                    self.index += offset
                elif where == 2:
                    self.index = self.length + offset

test = TestStream("test")
        post1 = requests.post(httpbin("post"), data=test)
        assert post1.status_code == 200
        assert post1.json()["data"] == "test"

test = TestStream("test")
        test.seek(2)
        post2 = requests.post(httpbin("post"), data=test)
        assert post2.status_code == 200
        assert post2.json()["data"] == "st"

def test_POSTBIN_GET_POST_FILES_WITH_DATA(self, httpbin):
        url = httpbin("post")
        requests.post(url).raise_for_status()

post1 = requests.post(url, data={"some": "data"})
        assert post1.status_code == 200

with open("requirements-dev.txt") as f:
            post2 = requests.post(url, data={"some": "data"}, files={"some": f})
        assert post2.status_code == 200

post4 = requests.post(url, data='[{"some": "json"}]')
        assert post4.status_code == 200

with pytest.raises(ValueError):
            requests.post(url, files=["bad file data"])

def test_post_with_custom_mapping(self, httpbin):
        class CustomMapping(MutableMapping):
            def __init__(self, *args, **kwargs):
                self.data = dict(*args, **kwargs)

def __delitem__(self, key):
                del self.data[key]

def __getitem__(self, key):
                return self.data[key]

def __setitem__(self, key, value):
                self.data[key] = value

def __iter__(self):
                return iter(self.data)

def __len__(self):
                return len(self.data)

data = CustomMapping({"some": "data"})
        url = httpbin("post")
        found_json = requests.post(url, data=data).json().get("form")
        assert found_json == {"some": "data"}

def test_conflicting_post_params(self, httpbin):
        url = httpbin("post")
        with open("requirements-dev.txt") as f:
            with pytest.raises(ValueError):
                requests.post(url, data='[{"some": "data"}]', files={"some": f})

def test_request_ok_set(self, httpbin):
        r = requests.get(httpbin("status", "404"))
        assert not r.ok

def test_status_raising(self, httpbin):
        r = requests.get(httpbin("status", "404"))
        with pytest.raises(requests.exceptions.HTTPError):
            r.raise_for_status()

r = requests.get(httpbin("status", "500"))
        assert not r.ok

def test_decompress_gzip(self, httpbin):
        r = requests.get(httpbin("gzip"))
        r.content.decode("ascii")

@pytest.mark.parametrize(
        "url, params",
        (
            ("/get", {"foo": "føø"}),
            ("/get", {"føø": "føø"}),
            ("/get", {"føø": "føø"}),
            ("/get", {"foo": "foo"}),
            ("ø", {"foo": "foo"}),
        ),
    )
    def test_unicode_get(self, httpbin, url, params):
        requests.get(httpbin(url), params=params)

def test_unicode_header_name(self, httpbin):
        requests.put(
            httpbin("put"),
            headers={"Content-Type": "application/octet-stream"},
            data="\xff",
        )  # compat.str is unicode.

def test_pyopenssl_redirect(self, httpbin_secure, httpbin_ca_bundle):
        requests.get(httpbin_secure("status", "301"), verify=httpbin_ca_bundle)

def test_invalid_ca_certificate_path(self, httpbin_secure):
        INVALID_PATH = "/garbage"
        with pytest.raises(IOError) as e:
            requests.get(httpbin_secure(), verify=INVALID_PATH)
        assert str(
            e.value
        ) == "Could not find a suitable TLS CA certificate bundle, invalid path: {}".format(
            INVALID_PATH
        )

def test_invalid_ssl_certificate_files(self, httpbin_secure):
        INVALID_PATH = "/garbage"
        with pytest.raises(IOError) as e:
            requests.get(httpbin_secure(), cert=INVALID_PATH)
        assert str(
            e.value
        ) == "Could not find the TLS certificate file, invalid path: {}".format(
            INVALID_PATH
        )

with pytest.raises(IOError) as e:
            requests.get(httpbin_secure(), cert=(".", INVALID_PATH))
        assert str(e.value) == (
            f"Could not find the TLS key file, invalid path: {INVALID_PATH}"
        )

@pytest.mark.parametrize(
        "env, expected",
        (
            ({}, True),
            ({"REQUESTS_CA_BUNDLE": "/some/path"}, "/some/path"),
            ({"REQUESTS_CA_BUNDLE": ""}, True),
            ({"CURL_CA_BUNDLE": "/some/path"}, "/some/path"),
            ({"CURL_CA_BUNDLE": ""}, True),
            ({"REQUESTS_CA_BUNDLE": "", "CURL_CA_BUNDLE": ""}, True),
            (
                {
                    "REQUESTS_CA_BUNDLE": "/some/path",
                    "CURL_CA_BUNDLE": "/curl/path",
                },
                "/some/path",
            ),
            (
                {
                    "REQUESTS_CA_BUNDLE": "",
                    "CURL_CA_BUNDLE": "/curl/path",
                },
                "/curl/path",
            ),
        ),
    )
    def test_env_cert_bundles(self, httpbin, env, expected):
        s = requests.Session()
        with mock.patch("os.environ", env):
            settings = s.merge_environment_settings(
                url=httpbin("get"), proxies={}, stream=False, verify=True, cert=None
            )
        assert settings["verify"] == expected

def test_http_with_certificate(self, httpbin):
        r = requests.get(httpbin(), cert=".")
        assert r.status_code == 200

@pytest.mark.skipif(
        SNIMissingWarning is None,
        reason="urllib3 2.0 removed that warning and errors out instead",
    )
    def test_https_warnings(self, nosan_server):
        """warnings are emitted with requests.get"""
        host, port, ca_bundle = nosan_server
        if HAS_MODERN_SSL or HAS_PYOPENSSL:
            warnings_expected = ("SubjectAltNameWarning",)
        else:
            warnings_expected = (
                "SNIMissingWarning",
                "InsecurePlatformWarning",
                "SubjectAltNameWarning",
            )

with pytest.warns() as warning_records:
            warnings.simplefilter("always")
            requests.get(f"https://localhost:{port}/", verify=ca_bundle)

warning_records = [
            item
            for item in warning_records
            if item.category.__name__ != "ResourceWarning"
        ]

warnings_category = tuple(item.category.__name__ for item in warning_records)
        assert warnings_category == warnings_expected

def test_certificate_failure(self, httpbin_secure):
        """
        When underlying SSL problems occur, an SSLError is raised.
        """
        with pytest.raises(RequestsSSLError):
            # Our local httpbin does not have a trusted CA, so this call will
            # fail if we use our default trust bundle.
            requests.get(httpbin_secure("status", "200"))

def test_urlencoded_get_query_multivalued_param(self, httpbin):
        r = requests.get(httpbin("get"), params={"test": ["foo", "baz"]})
        assert r.status_code == 200
        assert r.url == httpbin("get?test=foo&test=baz")

def test_form_encoded_post_query_multivalued_element(self, httpbin):
        r = requests.Request(
            method="POST", url=httpbin("post"), data=dict(test=["foo", "baz"])
        )
        prep = r.prepare()
        assert prep.body == "test=foo&test=baz"

def test_different_encodings_dont_break_post(self, httpbin):
        with open(__file__, "rb") as f:
            r = requests.post(
                httpbin("post"),
                data={"stuff": json.dumps({"a": 123})},
                params={"blah": "asdf1234"},
                files={"file": ("test_requests.py", f)},
            )
        assert r.status_code == 200

@pytest.mark.parametrize(
        "data",
        (
            {"stuff": "ëlïxr"},
            {"stuff": "ëlïxr".encode()},
            {"stuff": "elixr"},
            {"stuff": b"elixr"},
        ),
    )
    def test_unicode_multipart_post(self, httpbin, data):
        with open(__file__, "rb") as f:
            r = requests.post(
                httpbin("post"),
                data=data,
                files={"file": ("test_requests.py", f)},
            )
        assert r.status_code == 200

def test_unicode_multipart_post_fieldnames(self, httpbin):
        filename = os.path.splitext(__file__)[0] + ".py"
        with open(filename, "rb") as f:
            r = requests.Request(
                method="POST",
                url=httpbin("post"),
                data={b"stuff": "elixr"},
                files={"file": ("test_requests.py", f)},
            )
            prep = r.prepare()

assert b'name="stuff"' in prep.body
        assert b"name=\"b'stuff'\"" not in prep.body

def test_unicode_method_name(self, httpbin):
        with open(__file__, "rb") as f:
            files = {"file": f}
            r = requests.request(
                method="POST",
                url=httpbin("post"),
                files=files,
            )
        assert r.status_code == 200

def test_unicode_method_name_with_request_object(self, httpbin):
        s = requests.Session()
        with open(__file__, "rb") as f:
            files = {"file": f}
            req = requests.Request("POST", httpbin("post"), files=files)
            prep = s.prepare_request(req)
        assert isinstance(prep.method, builtin_str)
        assert prep.method == "POST"

resp = s.send(prep)
        assert resp.status_code == 200

def test_non_prepared_request_error(self):
        s = requests.Session()
        req = requests.Request("POST", "/")

with pytest.raises(ValueError) as e:
            s.send(req)
        assert str(e.value) == "You can only send PreparedRequests."

def test_custom_content_type(self, httpbin):
        with open(__file__, "rb") as f1:
            with open(__file__, "rb") as f2:
                data = {"stuff": json.dumps({"a": 123})}
                files = {
                    "file1": ("test_requests.py", f1),
                    "file2": ("test_requests", f2, "text/py-content-type"),
                }
                r = requests.post(httpbin("post"), data=data, files=files)
        assert r.status_code == 200
        assert b"text/py-content-type" in r.request.body

def test_hook_receives_request_arguments(self, httpbin):
        def hook(resp, **kwargs):
            assert resp is not None
            assert kwargs != {}

s = requests.Session()
        r = requests.Request("GET", httpbin(), hooks={"response": hook})
        prep = s.prepare_request(r)
        s.send(prep)

def test_session_hooks_are_used_with_no_request_hooks(self, httpbin):
        def hook(*args, **kwargs):
            pass

s = requests.Session()
        s.hooks["response"].append(hook)
        r = requests.Request("GET", httpbin())
        prep = s.prepare_request(r)
        assert prep.hooks["response"] != []
        assert prep.hooks["response"] == [hook]

def test_session_hooks_are_overridden_by_request_hooks(self, httpbin):
        def hook1(*args, **kwargs):
            pass

def hook2(*args, **kwargs):
            pass

assert hook1 is not hook2
        s = requests.Session()
        s.hooks["response"].append(hook2)
        r = requests.Request("GET", httpbin(), hooks={"response": [hook1]})
        prep = s.prepare_request(r)
        assert prep.hooks["response"] == [hook1]

def test_prepared_request_hook(self, httpbin):
        def hook(resp, **kwargs):
            resp.hook_working = True
            return resp

req = requests.Request("GET", httpbin(), hooks={"response": hook})
        prep = req.prepare()

s = requests.Session()
        s.proxies = getproxies()
        resp = s.send(prep)

assert hasattr(resp, "hook_working")

def test_prepared_from_session(self, httpbin):
        class DummyAuth(requests.auth.AuthBase):
            def __call__(self, r):
                r.headers["Dummy-Auth-Test"] = "dummy-auth-test-ok"
                return r

req = requests.Request("GET", httpbin("headers"))
        assert not req.auth

s = requests.Session()
        s.auth = DummyAuth()

prep = s.prepare_request(req)
        resp = s.send(prep)

assert resp.json()["headers"]["Dummy-Auth-Test"] == "dummy-auth-test-ok"

def test_prepare_request_with_bytestring_url(self):
        req = requests.Request("GET", b"https://httpbin.org/")
        s = requests.Session()
        prep = s.prepare_request(req)
        assert prep.url == "https://httpbin.org/"

def test_request_with_bytestring_host(self, httpbin):
        s = requests.Session()
        resp = s.request(
            "GET",
            httpbin("cookies/set?cookie=value"),
            allow_redirects=False,
            headers={"Host": b"httpbin.org"},
        )
        assert resp.cookies.get("cookie") == "value"

def test_links(self):
        r = requests.Response()
        r.headers = {
            "cache-control": "public, max-age=60, s-maxage=60",
            "connection": "keep-alive",
            "content-encoding": "gzip",
            "content-type": "application/json; charset=utf-8",
            "date": "Sat, 26 Jan 2013 16:47:56 GMT",
            "etag": '"6ff6a73c0e446c1f61614769e3ceb778"',
            "last-modified": "Sat, 26 Jan 2013 16:22:39 GMT",
            "link": (
                "<https://api.github.com/users/kennethreitz/repos?"
                'page=2&per_page=10>; rel="next", <https://api.github.'
                "com/users/kennethreitz/repos?page=7&per_page=10>; "
                ' rel="last"'
            ),
            "server": "GitHub.com",
            "status": "200 OK",
            "vary": "Accept",
            "x-content-type-options": "nosniff",
            "x-github-media-type": "github.beta",
            "x-ratelimit-limit": "60",
            "x-ratelimit-remaining": "57",
        }
        assert r.links["next"]["rel"] == "next"

def test_cookie_parameters(self):
        key = "some_cookie"
        value = "some_value"
        secure = True
        domain = "test.com"
        rest = {"HttpOnly": True}

jar = requests.cookies.RequestsCookieJar()
        jar.set(key, value, secure=secure, domain=domain, rest=rest)

assert len(jar) == 1
        assert "some_cookie" in jar

cookie = list(jar)[0]
        assert cookie.secure == secure
        assert cookie.domain == domain
        assert cookie._rest["HttpOnly"] == rest["HttpOnly"]

def test_cookie_as_dict_keeps_len(self):
        key = "some_cookie"
        value = "some_value"

key1 = "some_cookie1"
        value1 = "some_value1"