]> git.ipfire.org Git - thirdparty/psycopg.git/commitdiff
test: fix pproxy compatibility with Python 3.14
authorDaniele Varrazzo <daniele.varrazzo@gmail.com>
Sun, 31 Aug 2025 19:20:50 +0000 (21:20 +0200)
committerDaniele Varrazzo <daniele.varrazzo@gmail.com>
Tue, 2 Sep 2025 22:24:39 +0000 (00:24 +0200)
Work around the issue reported in https://github.com/qwj/python-proxy/issues/200

tests/fix_proxy.py
tests/pproxy_fix.py [new file with mode: 0644]

index ba8575e994430575e9afa0316f176852f21732c7..b2086b250231c1511ea6737d759ae4d5fece129b 100644 (file)
@@ -1,9 +1,9 @@
 import os
+import sys
 import time
 import socket
 import logging
 import subprocess as sp
-from shutil import which
 from contextlib import contextmanager
 
 import pytest
@@ -73,12 +73,9 @@ class Proxy:
             return
 
         logging.info("starting proxy")
-
-        if not (pproxy := which("pproxy")):
-            raise ValueError("pproxy program not found")
-        cmdline = [pproxy, "--reuse"]
-        cmdline.extend(["-l", f"tunnel://:{self.client_port}"])
-        cmdline.extend(["-r", f"tunnel://{self.server_host}:{self.server_port}"])
+        cmdline = [sys.executable, "-m", "tests.pproxy_fix", "--reuse"]
+        cmdline += ["-l", f"tunnel://:{self.client_port}"]
+        cmdline += ["-r", f"tunnel://{self.server_host}:{self.server_port}"]
 
         self.proc = sp.Popen(cmdline, stdout=sp.DEVNULL)
         logging.info("proxy started")
diff --git a/tests/pproxy_fix.py b/tests/pproxy_fix.py
new file mode 100644 (file)
index 0000000..b60ec37
--- /dev/null
@@ -0,0 +1,26 @@
+#!/usr/bin/env python
+"""Wrapper for pproxy to fix Python 3.14 compatibility
+
+Work around https://github.com/qwj/python-proxy/issues/200
+"""
+
+import sys
+import asyncio
+from typing import Any
+
+from pproxy.server import main as main_  # type: ignore[import-untyped]
+
+
+def main() -> Any:
+    # Before Python 3.14 `get_event_loop()` used to create a new loop.
+    # From Python 3.14 it raises a `RuntimeError`.
+    try:
+        asyncio.get_event_loop()
+    except RuntimeError:
+        asyncio.set_event_loop(asyncio.new_event_loop())
+
+    return main_()
+
+
+if __name__ == "__main__":
+    sys.exit(main())