test: Integrate the pytest-global-fixture as postgres fixture

Author: cdecker

contrib/pyln-testing/pyln/testing/db.py

@@ -148,6 +148,56 @@ def stop(self) -> None:
         pass
 
 
+class GlobalPostgresDbProvider(object):
+    """
+    Provider that uses pytest-global-fixture for a shared PostgreSQL instance.
+    This provider coordinates with the global_resource fixture to get databases
+    from a single globally-shared PostgreSQL instance.
+    """
+    def __init__(self, directory):
+        self.directory = directory
+        self.global_resource = None
+        self.port = None
+        print("Starting GlobalPostgresDbProvider (using global fixture)")
+
+    def set_global_resource(self, global_resource):
+        """Called by the db_provider fixture to inject the global_resource."""
+        self.global_resource = global_resource
+
+    def start(self):
+        """No-op: the global service is started by the coordinator."""
+        if self.global_resource is None:
+            raise RuntimeError(
+                "GlobalPostgresDbProvider requires global_resource fixture. "
+                "Make sure pytest-global-fixture plugin is loaded."
+            )
+        pass
+
+    def get_db(self, node_directory, testname, node_id):
+        """Get a database by requesting a tenant from the global service."""
+        if self.global_resource is None:
+            raise RuntimeError(
+                "global_resource not set. Did you call set_global_resource()?"
+            )
+
+        # Request a tenant from the global PostgreSQL service
+        config = self.global_resource(
+            "pytest_global_fixture.postgres_service:NativePostgresService"
+        )
+
+        # Store port for compatibility
+        if self.port is None:
+            self.port = config["port"]
+
+        # Create a PostgresDb instance
+        db = PostgresDb(config["dbname"], config["port"])
+        return db
+
+    def stop(self):
+        """No-op: global service cleanup is handled by the coordinator."""
+        pass
+
+
 class PostgresDbProvider(object):
     def __init__(self, directory):
         self.directory = directory

contrib/pyln-testing/pyln/testing/fixtures.py

@@ -1,5 +1,5 @@
 from concurrent import futures
-from pyln.testing.db import SqliteDbProvider, PostgresDbProvider
+from pyln.testing.db import SqliteDbProvider, PostgresDbProvider, GlobalPostgresDbProvider
 from pyln.testing.utils import NodeFactory, BitcoinD, ElementsD, env, LightningNode, TEST_DEBUG, TEST_NETWORK
 from pyln.client import Millisatoshi
 from typing import Dict
@@ -699,12 +699,20 @@ def checkMemleak(node):
 providers = {
     'sqlite3': SqliteDbProvider,
     'postgres': PostgresDbProvider,
+    'gpostgres': GlobalPostgresDbProvider,
 }
 
 
 @pytest.fixture
-def db_provider(test_base_dir):
+def db_provider(request, test_base_dir):
     provider = providers[os.getenv('TEST_DB_PROVIDER', 'sqlite3')](test_base_dir)
+
+    # If using GlobalPostgresDbProvider, inject the global_resource fixture
+    if isinstance(provider, GlobalPostgresDbProvider):
+        # Get the global_resource fixture from the same request context
+        global_resource = request.getfixturevalue('global_resource')
+        provider.set_global_resource(global_resource)
+
     provider.start()
     yield provider
     provider.stop()

contrib/pyln-testing/pyproject.toml

@@ -8,6 +8,8 @@ readme = "README.md"
 requires-python = ">=3.9,<4.0"
 dependencies = [
   "pytest>=8.0.0",
+  "pytest-xdist>=3.0.0",
+  "pytest-global-fixture",
   "ephemeral-port-reserve>=1.1.4",
   "psycopg2-binary>=2.9.0",
   "python-bitcoinlib>=0.11.0",

contrib/pytest-global-fixture/README.md

@@ -9,6 +9,9 @@ dependencies = [
     "pytest-xdist>=3.0.0",
 ]
 
+[project.entry-points.pytest11]
+pytest_global_fixture = "pytest_global_fixture.plugin"
+
 [project.optional-dependencies]
 dev = [
     "testcontainers>=3.7.0",

contrib/pytest-global-fixture/pyproject.toml

@@ -0,0 +1,11 @@
+"""
+pytest-global-fixture: A pytest plugin for globally shared infrastructure resources.
+"""
+
+from .base import InfrastructureService
+from .postgres_service import NativePostgresService
+
+__all__ = [
+    'InfrastructureService',
+    'NativePostgresService',
+]

contrib/pytest-global-fixture/pytest_global_fixture/__init__.py

@@ -19,6 +19,9 @@ def pytest_configure(config):
     """
     # Check if we are a worker (xdist). If no workerinput, we are Master.
     if not hasattr(config, "workerinput"):
+        print("\n" + "="*80)
+        print("PYTEST-GLOBAL-FIXTURE: Coordinator mode - managing shared resources")
+        print("="*80)
         manager = ServiceManager()
 
         # Bind to port 0 (ephemeral)
@@ -67,33 +70,44 @@ def coordinator_client(request):
     """
     if hasattr(request.config, "workerinput"):
         addr = request.config.workerinput["infra_rpc_addr"]
+        worker_id = request.config.workerinput.get("workerid", "unknown")
+        print(f"[{worker_id}] PYTEST-GLOBAL-FIXTURE: Worker connecting to coordinator at {addr}")
     else:
         # We are running sequentially (no xdist), or we are the master
         addr = request.config.infra_rpc_addr
-        
+        print(f"PYTEST-GLOBAL-FIXTURE: Sequential mode, using coordinator at {addr}")
+
     return xmlrpc.client.ServerProxy(addr)
 
 @pytest.fixture(scope="function")
-def global_resource(request, coordinator_client):
+def global_resource(request):
     """
     Factory fixture.
     Usage: global_resource("path.to:Class")
     """
-    
+
+    # Get RPC address
+    if hasattr(request.config, "workerinput"):
+        addr = request.config.workerinput["infra_rpc_addr"]
+    else:
+        addr = request.config.infra_rpc_addr
+
     # Track resources created in this scope for cleanup
     created_resources = []
 
     def _provision(class_path):
         # Create unique tenant ID: "gwX_testName_UUID"
         worker_id = getattr(request.config, "workerinput", {}).get("workerid", "master")
-        test_name = request.node.name.replace("[", "_").replace("]", "_") 
+        test_name = request.node.name.replace("[", "_").replace("]", "_")
         # Short uuid for uniqueness
         uid = uuid.uuid4().hex[:6]
         tenant_id = f"{worker_id}_{uid}"
-        
-        # RPC Call
-        config = coordinator_client.rpc_provision(class_path, tenant_id)
-        
+
+        # Create a new ServerProxy for each call to avoid connection reuse issues
+        # This prevents http.client.CannotSendRequest errors in multi-threaded scenarios
+        client = xmlrpc.client.ServerProxy(addr)
+        config = client.rpc_provision(class_path, tenant_id)
+
         created_resources.append((class_path, tenant_id))
         return config
 
@@ -102,7 +116,9 @@ def _provision(class_path):
     # Teardown logic
     for class_path, tenant_id in reversed(created_resources):
         try:
-            coordinator_client.rpc_deprovision(class_path, tenant_id)
+            # Create a new client for cleanup too
+            client = xmlrpc.client.ServerProxy(addr)
+            client.rpc_deprovision(class_path, tenant_id)
         except Exception as e:
             # We print but don't raise, to avoid masking test failures
             print(f"Warning: Failed to deprovision {tenant_id}: {e}")