Added pproxy test fixture, which allows to test requesting through so…

…cks proxy

httpcore/_types.py

@@ -1,12 +1,38 @@
 """
 Type definitions for type checking purposes.
 """
-
-from typing import List, Mapping, Optional, Tuple, TypeVar, Union
+import enum
+from typing import List, Mapping, NamedTuple, Optional, Tuple, TypeVar, Union
 
 T = TypeVar("T")
 StrOrBytes = Union[str, bytes]
 Origin = Tuple[bytes, bytes, int]
 URL = Tuple[bytes, bytes, Optional[int], bytes]
 Headers = List[Tuple[bytes, bytes]]
 TimeoutDict = Mapping[str, Optional[float]]
+
+SocksProxyOrigin = Tuple[bytes, int]
+
+
+class SocksProxyType(enum.Enum):
+    socks5 = "socks5"
+    socks4a = "socks4a"
+    socks4 = "socks4"
+
+
+class Socks4ProxyCredentials(NamedTuple):
+    user_id: bytes
+
+
+class Socks5ProxyCredentials(NamedTuple):
+    username: bytes
+    password: bytes
+
+
+SocksProxyCredentials = Union[Socks4ProxyCredentials, Socks5ProxyCredentials, None]
+
+
+class SocksProxyConfig(NamedTuple):
+    proxy_type: SocksProxyType
+    origin: SocksProxyOrigin
+    auth_credentials: SocksProxyCredentials = None

requirements.txt

@@ -26,6 +26,7 @@ flake8-pie==0.6.1
 isort==5.5.2
 mitmproxy==5.2
 mypy==0.782
+pproxy==2.3.5
 pytest==6.0.2
 pytest-trio==0.5.2
 pytest-cov==2.10.1

tests/conftest.py

@@ -1,7 +1,9 @@
 import asyncio
 import contextlib
 import os
+import shlex
 import ssl
+import subprocess
 import threading
 import time
 import typing
@@ -12,7 +14,13 @@
 from mitmproxy import options, proxy
 from mitmproxy.tools.dump import DumpMaster
 
-from httpcore._types import URL
+from httpcore._types import (
+    URL,
+    Socks4ProxyCredentials,
+    Socks5ProxyCredentials,
+    SocksProxyConfig,
+    SocksProxyType,
+)
 
 PROXY_HOST = "127.0.0.1"
 PROXY_PORT = 8080
@@ -141,3 +149,52 @@ def uds_server() -> typing.Iterator[Server]:
             yield server
     finally:
         os.remove(uds)
+
+
+class SocksProxyFixture(typing.NamedTuple):
+    socks5_with_auth: SocksProxyConfig
+    socks5_without_auth: SocksProxyConfig
+    socks4: SocksProxyConfig
+
+
+@pytest.fixture(scope="session")
+def socks() -> typing.Generator[SocksProxyFixture, None, None]:
+    socks4_type = SocksProxyType.socks4
+    no_auth_host = "localhost"
+    no_auth_port = 1085
+
+    socks5_type = SocksProxyType.socks5
+    auth_host = "localhost"
+    auth_port = 1086
+    auth_user = "user"
+    auth_pwd = "password"
+
+    cfg = SocksProxyFixture(
+        socks5_with_auth=SocksProxyConfig(
+            socks5_type, (no_auth_host.encode(), no_auth_port)
+        ),
+        socks5_without_auth=SocksProxyConfig(
+            socks5_type,
+            (auth_host.encode(), auth_port),
+            Socks5ProxyCredentials(auth_user.encode(), auth_pwd.encode()),
+        ),
+        socks4=SocksProxyConfig(
+            socks4_type,
+            (no_auth_host.encode(), no_auth_port),
+            Socks4ProxyCredentials(b"test_user_id"),
+        ),
+    )
+
+    command = (
+        f"pproxy -l socks4+socks5://{no_auth_host}:{no_auth_port} "
+        f"--auth 0 -l 'socks5://{auth_host}:{auth_port}#{auth_user}:{auth_pwd}'"
+    )
+
+    popen_args = shlex.split(command)
+
+    proc = subprocess.Popen(popen_args)
+    try:
+        time.sleep(1)  # a small delay to let the pproxy start to serve
+        yield cfg
+    finally:
+        proc.kill()