Use locally hosted services for tests

setup.py

@@ -60,6 +60,7 @@
             "pytest-cov",
             "pytest-mock",
             "pytest-asyncio",
+            "pytest-httpbin",
             "pytest-httpserver",
             "trustme",
             "requests",

tests/async_/test_async_transport.py

@@ -46,17 +46,21 @@
 
 
 @pytest.mark.asyncio
-async def test_async_transport_httpbin(httpbin_node_config):
+async def test_async_transport_httpbin(httpbin_node_config, httpbin):
     t = AsyncTransport([httpbin_node_config], meta_header=False)
     resp, data = await t.perform_request("GET", "/anything?key=value")
 
     assert resp.status == 200
     assert data["method"] == "GET"
-    assert data["url"] == "https://httpbin.org/anything?key=value"
+    assert data["url"] == f"{httpbin.url}/anything?key=value"
     assert data["args"] == {"key": "value"}
 
     data["headers"].pop("X-Amzn-Trace-Id", None)
-    assert data["headers"] == {"User-Agent": DEFAULT_USER_AGENT, "Host": "httpbin.org"}
+    assert data["headers"] == {
+        "User-Agent": DEFAULT_USER_AGENT,
+        "Connection": "keep-alive",
+        "Host": f"{httpbin.host}:{httpbin.port}",
+    }
 
 
 @pytest.mark.skipif(

tests/async_/test_httpbin.py

@@ -27,7 +27,7 @@
 
 
 @pytest.mark.asyncio
-async def test_simple_request(httpbin_node_config):
+async def test_simple_request(httpbin_node_config, httpbin):
     t = AsyncTransport([httpbin_node_config])
 
     resp, data = await t.perform_request(
@@ -38,7 +38,7 @@ async def test_simple_request(httpbin_node_config):
     )
     assert resp.status == 200
     assert data["method"] == "GET"
-    assert data["url"] == "https://httpbin.org/anything?key[]=1&key[]=2&q1&q2="
+    assert data["url"] == f"{httpbin.url}/anything?key[]=1&key[]=2&q1&q2="
 
     # httpbin makes no-value query params into ''
     assert data["args"] == {
@@ -53,13 +53,14 @@ async def test_simple_request(httpbin_node_config):
         "Content-Type": "application/json",
         "Content-Length": "15",
         "Custom": "headeR",
-        "Host": "httpbin.org",
+        "Connection": "keep-alive",
+        "Host": f"{httpbin.host}:{httpbin.port}",
     }
     assert all(v == data["headers"][k] for k, v in request_headers.items())
 
 
 @pytest.mark.asyncio
-async def test_node(httpbin_node_config):
+async def test_node(httpbin_node_config, httpbin):
     def new_node(**kwargs):
         return AiohttpHttpNode(dataclasses.replace(httpbin_node_config, **kwargs))
 
@@ -69,11 +70,12 @@ def new_node(**kwargs):
     parsed = parse_httpbin(data)
     assert parsed == {
         "headers": {
-            "Host": "httpbin.org",
+            "Connection": "keep-alive",
+            "Host": f"{httpbin.host}:{httpbin.port}",
             "User-Agent": DEFAULT_USER_AGENT,
         },
         "method": "GET",
-        "url": "https://httpbin.org/anything",
+        "url": f"{httpbin.url}/anything",
     }
 
     node = new_node(http_compress=True)
@@ -83,11 +85,12 @@ def new_node(**kwargs):
     assert parsed == {
         "headers": {
             "Accept-Encoding": "gzip",
-            "Host": "httpbin.org",
+            "Connection": "keep-alive",
+            "Host": f"{httpbin.host}:{httpbin.port}",
             "User-Agent": DEFAULT_USER_AGENT,
         },
         "method": "GET",
-        "url": "https://httpbin.org/anything",
+        "url": f"{httpbin.url}/anything",
     }
 
     resp, data = await node.perform_request("GET", "/anything", body=b"hello, world!")
@@ -99,11 +102,12 @@ def new_node(**kwargs):
             "Content-Encoding": "gzip",
             "Content-Type": "application/octet-stream",
             "Content-Length": "33",
-            "Host": "httpbin.org",
+            "Connection": "keep-alive",
+            "Host": f"{httpbin.host}:{httpbin.port}",
             "User-Agent": DEFAULT_USER_AGENT,
         },
         "method": "GET",
-        "url": "https://httpbin.org/anything",
+        "url": f"{httpbin.url}/anything",
     }
 
     resp, data = await node.perform_request(
@@ -120,9 +124,10 @@ def new_node(**kwargs):
             "Content-Encoding": "gzip",
             "Content-Length": "36",
             "Content-Type": "application/json",
-            "Host": "httpbin.org",
+            "Connection": "keep-alive",
+            "Host": f"{httpbin.host}:{httpbin.port}",
             "User-Agent": DEFAULT_USER_AGENT,
         },
         "method": "POST",
-        "url": "https://httpbin.org/anything",
+        "url": f"{httpbin.url}/anything",
     }

tests/conftest.py

@@ -66,32 +66,33 @@ async def perform_request(self, *args, **kwargs):
         return NodeApiResponse(meta, self.body)
 
 
-@pytest.fixture(scope="session", params=[True, False])
-def httpbin_cert_fingerprint(request) -> str:
-    """Gets the SHA256 fingerprint of the certificate for 'httpbin.org'"""
-    sock = socket.create_connection(("httpbin.org", 443))
+@pytest.fixture(
+    scope="session", params=["short-lower", "short-upper", "long-lower", "long-upper"]
+)
+def cert_fingerprint(request, httpbin_secure) -> str:
+    """Gets the SHA256 fingerprint of the certificate for the secure httpbin"""
+    sock = socket.create_connection((httpbin_secure.host, httpbin_secure.port))
     ctx = ssl.create_default_context()
     ctx.check_hostname = False
     ctx.verify_mode = ssl.CERT_NONE
     sock = ctx.wrap_socket(sock)
     digest = hashlib.sha256(sock.getpeercert(binary_form=True)).hexdigest()
     assert len(digest) == 64
     sock.close()
-    if request.param:
+    if "upper" in request.param:
+        digest = digest.upper()
+    else:
+        digest = digest.lower()
+    if "short" in request.param:
         return digest
     else:
         return ":".join([digest[i : i + 2] for i in range(0, len(digest), 2)])
 
 
 @pytest.fixture(scope="session")
-def httpbin_node_config() -> NodeConfig:
-    try:
-        sock = socket.create_connection(("httpbin.org", 443))
-    except Exception as e:
-        pytest.skip(f"Couldn't connect to httpbin.org, internet not connected? {e}")
-    sock.close()
+def httpbin_node_config(httpbin) -> NodeConfig:
     return NodeConfig(
-        "https", "httpbin.org", 443, verify_certs=False, ssl_show_warn=False
+        "http", httpbin.host, httpbin.port, verify_certs=False, ssl_show_warn=False
     )
 
 

tests/node/test_http_aiohttp.py

@@ -290,14 +290,14 @@ async def test_head_workaround(self, aiohttp_fixed_head_bug):
 
 
 @pytest.mark.asyncio
-async def test_ssl_assert_fingerprint(httpbin_cert_fingerprint):
+async def test_ssl_assert_fingerprint(cert_fingerprint, httpbin_secure):
     with warnings.catch_warnings(record=True) as w:
         node = AiohttpHttpNode(
             NodeConfig(
                 scheme="https",
-                host="httpbin.org",
-                port=443,
-                ssl_assert_fingerprint=httpbin_cert_fingerprint,
+                host=httpbin_secure.host,
+                port=httpbin_secure.port,
+                ssl_assert_fingerprint=cert_fingerprint,
             )
         )
         resp, _ = await node.perform_request("GET", "/")
@@ -307,23 +307,29 @@ async def test_ssl_assert_fingerprint(httpbin_cert_fingerprint):
 
 
 @pytest.mark.asyncio
-async def test_default_headers():
-    node = AiohttpHttpNode(NodeConfig(scheme="https", host="httpbin.org", port=443))
+async def test_default_headers(httpbin):
+    node = AiohttpHttpNode(
+        NodeConfig(scheme="http", host=httpbin.host, port=httpbin.port)
+    )
     resp, data = await node.perform_request("GET", "/anything")
 
     assert resp.status == 200
     headers = json.loads(data)["headers"]
     headers.pop("X-Amzn-Trace-Id", None)
-    assert headers == {"Host": "httpbin.org", "User-Agent": DEFAULT_USER_AGENT}
+    assert headers == {
+        "Connection": "keep-alive",
+        "Host": f"{httpbin.host}:{httpbin.port}",
+        "User-Agent": DEFAULT_USER_AGENT,
+    }
 
 
 @pytest.mark.asyncio
-async def test_custom_headers():
+async def test_custom_headers(httpbin):
     node = AiohttpHttpNode(
         NodeConfig(
-            scheme="https",
-            host="httpbin.org",
-            port=443,
+            scheme="http",
+            host=httpbin.host,
+            port=httpbin.port,
             headers={"accept-encoding": "gzip", "Content-Type": "application/json"},
         )
     )
@@ -341,19 +347,20 @@ async def test_custom_headers():
     headers.pop("X-Amzn-Trace-Id", None)
     assert headers == {
         "Accept-Encoding": "gzip",
+        "Connection": "keep-alive",
         "Content-Type": "application/x-ndjson",
-        "Host": "httpbin.org",
+        "Host": f"{httpbin.host}:{httpbin.port}",
         "User-Agent": "custom-agent/1.2.3",
     }
 
 
 @pytest.mark.asyncio
-async def test_custom_user_agent():
+async def test_custom_user_agent(httpbin):
     node = AiohttpHttpNode(
         NodeConfig(
-            scheme="https",
-            host="httpbin.org",
-            port=443,
+            scheme="http",
+            host=httpbin.host,
+            port=httpbin.port,
             headers={
                 "accept-encoding": "gzip",
                 "Content-Type": "application/json",
@@ -371,8 +378,9 @@ async def test_custom_user_agent():
     headers.pop("X-Amzn-Trace-Id", None)
     assert headers == {
         "Accept-Encoding": "gzip",
+        "Connection": "keep-alive",
         "Content-Type": "application/json",
-        "Host": "httpbin.org",
+        "Host": f"{httpbin.host}:{httpbin.port}",
         "User-Agent": "custom-agent/1.2.3",
     }
 
@@ -383,9 +391,11 @@ def test_repr():
 
 
 @pytest.mark.asyncio
-async def test_head():
+async def test_head(httpbin):
     node = AiohttpHttpNode(
-        NodeConfig(scheme="https", host="httpbin.org", port=443, http_compress=True)
+        NodeConfig(
+            scheme="http", host=httpbin.host, port=httpbin.port, http_compress=True
+        )
     )
     resp, data = await node.perform_request("HEAD", "/anything")
 

tests/node/test_http_httpx.py

@@ -145,13 +145,13 @@ async def test_merge_headers(self):
         assert request.headers["h3"] == "v3"
 
 
-def test_ssl_assert_fingerprint(httpbin_cert_fingerprint):
+def test_ssl_assert_fingerprint(cert_fingerprint, httpbin_secure):
     with pytest.raises(ValueError, match="httpx does not support certificate pinning"):
         HttpxAsyncHttpNode(
             NodeConfig(
                 scheme="https",
-                host="httpbin.org",
-                port=443,
-                ssl_assert_fingerprint=httpbin_cert_fingerprint,
+                host=httpbin_secure.host,
+                port=httpbin_secure.port,
+                ssl_assert_fingerprint=cert_fingerprint,
             )
         )

tests/node/test_urllib3_chain_certs.py

@@ -30,13 +30,13 @@
 
 @requires_ssl_assert_fingerprint_in_chain
 @pytest.mark.parametrize("node_cls", [Urllib3HttpNode, RequestsHttpNode])
-def test_ssl_assert_fingerprint_invalid_length(node_cls):
+def test_ssl_assert_fingerprint_invalid_length(node_cls, httpbin_secure):
     with pytest.raises(ValueError) as e:
         node_cls(
             NodeConfig(
                 "https",
-                "httpbin.org",
-                443,
+                httpbin_secure.host,
+                httpbin_secure.port,
                 ssl_assert_fingerprint="0000",
             )
         )
@@ -49,22 +49,14 @@ def test_ssl_assert_fingerprint_invalid_length(node_cls):
 
 @requires_ssl_assert_fingerprint_in_chain
 @pytest.mark.parametrize("node_cls", [Urllib3HttpNode, RequestsHttpNode])
-@pytest.mark.parametrize(
-    "ssl_assert_fingerprint",
-    [
-        "8ecde6884f3d87b1125ba31ac3fcb13d7016de7f57cc904fe1cb97c6ae98196e",
-        "8e:cd:e6:88:4f:3d:87:b1:12:5b:a3:1a:c3:fc:b1:3d:70:16:de:7f:57:cc:90:4f:e1:cb:97:c6:ae:98:19:6e",
-        "8ECDE6884F3D87B1125BA31AC3FCB13D7016DE7F57CC904FE1CB97C6AE98196E",
-    ],
-)
-def test_assert_fingerprint_in_cert_chain(node_cls, ssl_assert_fingerprint):
+def test_assert_fingerprint_in_cert_chain(node_cls, cert_fingerprint, httpbin_secure):
     with warnings.catch_warnings(record=True) as w:
         node = node_cls(
             NodeConfig(
                 "https",
-                "httpbin.org",
-                443,
-                ssl_assert_fingerprint=ssl_assert_fingerprint,
+                httpbin_secure.host,
+                httpbin_secure.port,
+                ssl_assert_fingerprint=cert_fingerprint,
             )
         )
         meta, _ = node.perform_request("GET", "/")
@@ -75,11 +67,13 @@ def test_assert_fingerprint_in_cert_chain(node_cls, ssl_assert_fingerprint):
 
 @requires_ssl_assert_fingerprint_in_chain
 @pytest.mark.parametrize("node_cls", [Urllib3HttpNode, RequestsHttpNode])
-def test_assert_fingerprint_in_cert_chain_failure(node_cls):
+def test_assert_fingerprint_in_cert_chain_failure(
+    node_cls, httpbin_secure, cert_fingerprint
+):
     node = node_cls(
         NodeConfig(
             "https",
-            "httpbin.org",
+            "www.elastic.co",
             443,
             ssl_assert_fingerprint="0" * 64,
         )
@@ -96,4 +90,4 @@ def test_assert_fingerprint_in_cert_chain_failure(node_cls):
         in err
     )
     # This is the root CA for httpbin.org with a leading comma to denote more than one cert was listed.
-    assert ', "8ecde6884f3d87b1125ba31ac3fcb13d7016de7f57cc904fe1cb97c6ae98196e"' in err
+    assert ', "cbb522d7b7f127ad6a0113865bdf1cd4102e7d0759af635a7cf4720dc963c53b"' in err