Adds basic xfail tests for secure proxy support

aiohttp/client_reqrep.py

@@ -547,8 +547,8 @@ def update_proxy(
         proxy_auth: Optional[BasicAuth],
         proxy_headers: Optional[LooseHeaders],
     ) -> None:
-        if proxy and not proxy.scheme == "http":
-            raise ValueError("Only http proxies are supported")
+        # if proxy and not proxy.scheme == "http":
+        #     raise ValueError("Only http proxies are supported")
         if proxy_auth and not isinstance(proxy_auth, helpers.BasicAuth):
             raise ValueError("proxy_auth must be None or BasicAuth() tuple")
         self.proxy = proxy

setup.cfg

@@ -53,6 +53,7 @@ addopts =
 filterwarnings =
     error
     ignore:module 'ssl' has no attribute 'OP_NO_COMPRESSION'. The Python interpreter is compiled against OpenSSL < 1.0.0. Ref. https.//docs.python.org/3/library/ssl.html#ssl.OP_NO_COMPRESSION:UserWarning
+    ignore:The loop argument is deprecated since Python 3.8, and scheduled for removal in Python 3.10.:DeprecationWarning
 junit_suite_name = aiohttp_test_suite
 norecursedirs = dist docs build .tox .eggs
 minversion = 3.8.2

setup.py

@@ -89,6 +89,7 @@ def read(f):
     "pytest-timeout",
     "async-generator",
     "pytest-xdist",
+    "proxy.py",
 ]
 
 

tests/conftest.py

@@ -75,6 +75,12 @@ def tls_certificate_pem_path(tls_certificate):
         yield cert_pem
 
 
+@pytest.fixture
+def tls_certificate_private_key_path(tls_certificate_authority):
+    with tls_certificate_authority.private_key_pem.tempfile() as ca_key_pem:
+        yield ca_key_pem
+
+
 @pytest.fixture
 def tls_certificate_pem_bytes(tls_certificate):
     return tls_certificate.cert_chain_pems[0].bytes()

tests/test_proxy_functional.py

@@ -3,13 +3,40 @@
 import pathlib
 from unittest import mock
 
+import proxy
 import pytest
 from yarl import URL
 
 import aiohttp
 from aiohttp import web
 
 
+@pytest.fixture
+def secure_proxy_url(tls_ca_certificate_pem_path, tls_certificate_private_key_path):
+    """Return the an instance of a running secure proxy."""
+    port = str(proxy.common.utils.get_available_port())
+    address = '127.0.0.1'
+    input_args = [
+        '--num-workers', '1',
+        '--threadless',
+        '--hostname', '127.0.0.1',
+        '--port', port,
+        '--cert-file', tls_ca_certificate_pem_path,
+        '--key-file', tls_certificate_private_key_path,
+        '--threadless',
+    ]
+    proxy_url = f"https://{address}:{port}"
+    with proxy.Proxy(input_args=input_args):
+        yield proxy_url
+
+
+async def test_secure_proxy_http_absolute_path(secure_proxy_url, get_request) -> None:
+    """Test HTTP urls can be requested through a secure proxy."""
+    import pdb;pdb.set_trace()
+    url = "http://httpbin.org/get"
+    await get_request(url=url, proxy=secure_proxy_url)
+
+
 @pytest.fixture
 def proxy_test_server(aiohttp_raw_server, loop, monkeypatch):
     # Handle all proxy requests and imitate remote server response.