test: harden integration tests against transient loopback resets

Shared retry helper for both the HTTP and raw-CONNECT paths (6 tries, backoff),
longer timeouts, and a small delay between rotation requests. Fixes occasional
Windows-CI flakiness without changing library behavior.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: SAKMZ

tests/test_relay.py

@@ -244,36 +244,51 @@ def setUpClass(cls):
     def tearDownClass(cls):
         cls.origin.shutdown()
 
-    def _get(self, relay, path="/hi", retries=4):
-        # Loopback connections under CI load can occasionally reset; retry the
-        # transient ones so the rotation/throughput tests stay deterministic.
+    @staticmethod
+    def _retry(fn, retries=6):
+        # Loopback connections under CI load (especially Windows) can reset;
+        # retry transient failures so the integration tests stay deterministic.
         last = None
         for attempt in range(retries):
             try:
-                opener = urllib.request.build_opener(
-                    urllib.request.ProxyHandler({"http": relay.url}))
-                return opener.open(
-                    f"http://127.0.0.1:{self.origin_port}{path}", timeout=10
-                ).read().decode()
+                return fn()
             except Exception as e:  # noqa: BLE001 - retry transient socket errors
                 last = e
-                time.sleep(0.2 * (attempt + 1))
+                time.sleep(0.15 * (attempt + 1))
         raise last
 
-    def _connect_get(self, relay, path="/tunnel"):
-        s = socket.create_connection((relay.listen_host, relay.port), timeout=10)
-        s.sendall(f"CONNECT 127.0.0.1:{self.origin_port} HTTP/1.1\r\n"
-                  f"Host: 127.0.0.1:{self.origin_port}\r\n\r\n".encode())
-        self.assertIn(b"200", s.recv(4096).split(b"\r\n", 1)[0])
-        s.sendall(f"GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n".encode())
-        data = b""
-        while True:
-            chunk = s.recv(4096)
-            if not chunk:
-                break
-            data += chunk
-        s.close()
-        return data
+    def _get(self, relay, path="/hi", retries=6):
+        def once():
+            opener = urllib.request.build_opener(
+                urllib.request.ProxyHandler({"http": relay.url}))
+            return opener.open(
+                f"http://127.0.0.1:{self.origin_port}{path}", timeout=15
+            ).read().decode()
+        return self._retry(once, retries=retries)
+
+    def _connect_get(self, relay, path="/tunnel", retries=6):
+        def once():
+            s = socket.create_connection((relay.listen_host, relay.port), timeout=15)
+            try:
+                s.sendall(f"CONNECT 127.0.0.1:{self.origin_port} HTTP/1.1\r\n"
+                          f"Host: 127.0.0.1:{self.origin_port}\r\n\r\n".encode())
+                status = s.recv(4096).split(b"\r\n", 1)[0]
+                if b"200" not in status:
+                    raise ConnectionError(f"CONNECT not established: {status!r}")
+                s.sendall(f"GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\n"
+                          f"Connection: close\r\n\r\n".encode())
+                data = b""
+                while True:
+                    chunk = s.recv(4096)
+                    if not chunk:
+                        break
+                    data += chunk
+                if b"OK " not in data:
+                    raise ConnectionError("no response body")
+                return data
+            finally:
+                s.close()
+        return self._retry(once, retries=retries)
 
 
 class TestSocks5(_Base):
@@ -359,6 +374,7 @@ def test_roundrobin_uses_both(self):
         with ProxyRelay(pool, rotation="roundrobin") as relay:
             for _ in range(4):
                 self.assertEqual(self._get(relay, "/x"), "OK /x")
+                time.sleep(0.05)
         # Round-robin must have used BOTH upstreams (exact split can shift by one
         # if a transient loopback reset triggered a retry).
         self.assertGreaterEqual(self.a.conns, 1)