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

Author: cdecker

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

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 from typing import Any, Dict
 
+
 class InfrastructureService(ABC):
     """
     Interface that all shared resources must implement.
@@ -27,12 +28,12 @@ def stop_global(self) -> None:
     def create_tenant(self, tenant_id: str) -> Dict[str, Any]:
         """
         Create a logical isolation unit (Database, Schema, VHost).
-        
+
         Args:
             tenant_id: A unique string identifier for the requester (e.g., 'gw0_test_uuid').
-        
+
         Returns:
-            A JSON-serializable dictionary containing connection details 
+            A JSON-serializable dictionary containing connection details
             (host, port, user, password, db_name, etc.)
         """
         pass

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

@@ -1,14 +1,16 @@
 import threading
 import importlib
 import sys
-from typing import Dict, Set
+from typing import Dict
 from .base import InfrastructureService
 
+
 class ServiceManager:
     """
     Runs on the Master process.
     Manages the lifecycle of services and exposes them via XML-RPC.
     """
+
     def __init__(self):
         self._services: Dict[str, InfrastructureService] = {}  # Map path -> Instance
         self._lock = threading.Lock()
@@ -26,7 +28,7 @@ def _load_class(self, class_path: str) -> InfrastructureService:
             # Ensure the current directory is in path so we can import local tests
             if "." not in sys.path:
                 sys.path.insert(0, ".")
-            
+
             module = importlib.import_module(module_name)
             cls = getattr(module, class_name)
             return cls()  # Instantiate
@@ -44,16 +46,16 @@ def rpc_provision(self, class_path: str, tenant_id: str) -> Dict:
             if class_path not in self._services:
                 print(f"[Coordinator] Dynamically loading: {class_path}")
                 service = self._load_class(class_path)
-                
+
                 print(f"[Coordinator] Starting Global Resource: {class_path}")
                 try:
                     service.start_global()
                 except Exception as e:
                     print(f"[Coordinator] Failed to start {class_path}: {e}")
                     raise e
-                
+
                 self._services[class_path] = service
-            
+
             service = self._services[class_path]
 
             # 2. Create Tenant

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

@@ -1,58 +1,62 @@
 import pytest
 import threading
 import uuid
-import sys
 from xmlrpc.server import SimpleXMLRPCServer, SimpleXMLRPCRequestHandler
 import xmlrpc.client
 from .manager import ServiceManager
 
 # --- RPC Server Setup (Runs on Master) ---
 
+
 class QuietXMLRPCRequestHandler(SimpleXMLRPCRequestHandler):
     """Suppress standard logging from XML-RPC server."""
+
     def log_message(self, format, *args):
         pass
 
+
 def pytest_configure(config):
     """
     If this is the Master/Coordinator process, start the RPC server.
     """
     # Check if we are a worker (xdist). If no workerinput, we are Master.
     if not hasattr(config, "workerinput"):
-        print("\n" + "="*80)
+        print("\n" + "=" * 80)
         print("PYTEST-GLOBAL-FIXTURE: Coordinator mode - managing shared resources")
-        print("="*80)
+        print("=" * 80)
         manager = ServiceManager()
-        
+
         # Bind to port 0 (ephemeral)
         server = SimpleXMLRPCServer(
-            ("localhost", 0), 
-            requestHandler=QuietXMLRPCRequestHandler, 
+            ("localhost", 0),
+            requestHandler=QuietXMLRPCRequestHandler,
             allow_none=True,
-            logRequests=False
+            logRequests=False,
         )
         server.register_instance(manager)
-        
+
         # Run server in daemon thread
         t = threading.Thread(target=server.serve_forever, daemon=True)
         t.start()
-        
+
         host, port = server.server_address
         rpc_addr = f"http://{host}:{port}/"
-        
+
         # Store in config to pass to workers/hooks
         config.infra_rpc_addr = rpc_addr
         config.infra_manager = manager
-        
+
         print(f"--- [Coordinator] Infrastructure Manager listening at {rpc_addr} ---")
 
+
 def pytest_configure_node(node):
     """
     This runs on Master for each Worker node being created.
     Pass the RPC address to the worker.
     """
     node.workerinput["infra_rpc_addr"] = node.config.infra_rpc_addr
 
+
 def pytest_unconfigure(config):
     """
     Run teardown on Master when session ends.
@@ -63,6 +67,7 @@ def pytest_unconfigure(config):
 
 # --- Fixture (Runs on Workers) ---
 
+
 @pytest.fixture(scope="session")
 def coordinator_client(request):
     """
@@ -71,14 +76,17 @@ 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}")
+        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):
     """
@@ -98,7 +106,6 @@ def global_resource(request):
     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("]", "_")
         # Short uuid for uniqueness
         uid = uuid.uuid4().hex[:6]
         tenant_id = f"{worker_id}_{uid}"

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

@@ -12,7 +12,6 @@
 import subprocess
 import tempfile
 import time
-from pathlib import Path
 from psycopg2 import sql
 from typing import Dict
 from .base import InfrastructureService

contrib/pytest-global-fixture/tests/conftest.py

@@ -1,6 +1,7 @@
 import pytest
 import psycopg2
 
+
 @pytest.fixture(scope="function")
 def postgres_db(global_resource):
     """
@@ -12,12 +13,12 @@ def postgres_db(global_resource):
     """
     # Request the service by Class Path
     db_config = global_resource("tests.resources:PostgresService")
-    
+
     # Create the actual connection object for the test to use
     conn = psycopg2.connect(**db_config)
     conn.autocommit = True
-    
+
     yield conn
-    
+
     conn.close()
     # After yield, 'global_resource' fixture automatically calls remove_tenant

contrib/pytest-global-fixture/tests/resources.py

@@ -4,6 +4,7 @@
 from testcontainers.postgres import PostgresContainer
 from pytest_global_fixture.base import InfrastructureService
 
+
 class PostgresService(InfrastructureService):
     def __init__(self):
         self.container = None
@@ -17,16 +18,16 @@ def start_global(self) -> None:
         time.sleep(30)
         self.container = PostgresContainer("postgres:15-alpine")
         self.container.start()
-        
+
         # Connection info for the Superuser
         self.master_config = {
             "host": self.container.get_container_host_ip(),
             "port": self.container.get_exposed_port(5432),
             "user": self.container.username,
             "password": self.container.password,
-            "dbname": self.container.dbname
+            "dbname": self.container.dbname,
         }
-        
+
         # Wait for readiness
         self._wait_for_ready()
 
@@ -47,7 +48,9 @@ def create_tenant(self, tenant_id: str) -> dict:
             with conn.cursor() as cur:
                 # Sanitize the tenant_id slightly for SQL (simple alphanumeric check is best)
                 safe_name = tenant_id.replace("-", "_")
-                cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(safe_name)))
+                cur.execute(
+                    sql.SQL("CREATE DATABASE {}").format(sql.Identifier(safe_name))
+                )
         finally:
             conn.close()
 
@@ -66,14 +69,20 @@ def remove_tenant(self, tenant_id: str) -> None:
         try:
             with conn.cursor() as cur:
                 # Force disconnect users before dropping
-                cur.execute(sql.SQL("""
+                cur.execute(
+                    sql.SQL("""
                     SELECT pg_terminate_backend(pg_stat_activity.pid)
                     FROM pg_stat_activity
                     WHERE pg_stat_activity.datname = {}
                     AND pid <> pg_backend_pid();
-                """).format(sql.Literal(safe_name)))
-                
-                cur.execute(sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(safe_name)))
+                """).format(sql.Literal(safe_name))
+                )
+
+                cur.execute(
+                    sql.SQL("DROP DATABASE IF EXISTS {}").format(
+                        sql.Identifier(safe_name)
+                    )
+                )
         finally:
             conn.close()
 

contrib/pytest-global-fixture/tests/test_postgres.py

@@ -1,5 +1,4 @@
-import pytest
-import os
+
 
 def test_table_isolation_a(postgres_db, worker_id):
     """
@@ -11,13 +10,14 @@ def test_table_isolation_a(postgres_db, worker_id):
         cur.execute("SELECT current_database();")
         db_name = cur.fetchone()[0]
         print(f"Worker {worker_id} connected to {db_name}")
-        
+
         cur.execute("CREATE TABLE items (id serial PRIMARY KEY, name text);")
         cur.execute("INSERT INTO items (name) VALUES ('item_from_a');")
-        
+
         cur.execute("SELECT count(*) FROM items;")
         assert cur.fetchone()[0] == 1
 
+
 def test_table_isolation_b(postgres_db, worker_id):
     """
     Runs in parallel with A.
@@ -36,8 +36,8 @@ def test_table_isolation_b(postgres_db, worker_id):
         except Exception:
             exists = False
             # Reset transaction if error occurred
-            postgres_db.rollback() 
-        
+            postgres_db.rollback()
+
         if not exists:
             # Good, table doesn't exist, let's create our own
             cur.execute("CREATE TABLE items (id serial PRIMARY KEY, name text);")

tests/fixtures.py

@@ -1,7 +1,7 @@
 from utils import TEST_NETWORK, VALGRIND  # noqa: F401,F403
 from pyln.testing.fixtures import directory, test_base_dir, test_name, chainparams, node_factory, bitcoind, teardown_checks, db_provider, executor, setup_logging, jsonschemas  # noqa: F401,F403
 from pyln.testing import utils
-from pytest_global_fixture.plugin import global_resource, coordinator_client
+from pytest_global_fixture.plugin import global_resource, coordinator_client  # noqa: F401
 from utils import COMPAT
 from pathlib import Path