Add comprehensive README.md for pytest-global-fixture plugin

Documents the shared infrastructure pattern for pytest tests, explaining
how the plugin enables globally-shared resources (like Docker containers)
across pytest-xdist workers while maintaining per-test isolation through
tenant namespaces.

The README includes:
- Problem statement and solution overview
- Architecture details (InfrastructureService, ServiceManager, Plugin)
- Complete implementation guide with code examples
- Performance analysis showing ~35x speedup
- Technical details on XML-RPC, thread safety, and dynamic loading
- Example services for PostgreSQL, Redis, and RabbitMQ
- Guidance on when to use this pattern

Co-Authored-By: Claude <[email protected]>

Author: cdecker

conftest.py

@@ -0,0 +1,35 @@
+[project]
+name = "pytest-global-fixture"
+version = "0.1.0"
+description = "A pytest plugin for coordinating global resources across xdist workers"
+readme = "README.md"
+requires-python = ">=3.9"
+dependencies = [
+    "pytest>=7.0.0",
+    "pytest-xdist>=3.0.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "testcontainers>=3.7.0",
+    "psycopg2-binary>=2.9.0",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["pytest_global_fixture"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+# Register the plugin explicitly for local development
+addopts = "-p pytest_global_fixture.plugin -n auto"
+
+[dependency-groups]
+dev = [
+    "psycopg2-binary>=2.9.10",
+    "pytest-flakefinder>=1.1.0",
+    "testcontainers>=4.13.3",
+]

contrib/pytest-global-fixture/README.md

@@ -0,0 +1,45 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict
+
+class InfrastructureService(ABC):
+    """
+    Interface that all shared resources must implement.
+    These methods run exclusively on the Coordinator (Main) Process.
+    """
+
+    @abstractmethod
+    def start_global(self) -> None:
+        """
+        Initialize the heavy, global resource (e.g., Docker container).
+        This is called exactly once, the first time a worker requests this service.
+        """
+        pass
+
+    @abstractmethod
+    def stop_global(self) -> None:
+        """
+        Teardown the global resource.
+        Called at the very end of the pytest session.
+        """
+        pass
+
+    @abstractmethod
+    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 
+            (host, port, user, password, db_name, etc.)
+        """
+        pass
+
+    @abstractmethod
+    def remove_tenant(self, tenant_id: str) -> None:
+        """
+        Clean up the logical isolation unit.
+        """
+        pass

contrib/pytest-global-fixture/pyproject.toml

@@ -0,0 +1,93 @@
+import threading
+import importlib
+import sys
+from typing import Dict, Set
+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()
+
+    def _load_class(self, class_path: str) -> InfrastructureService:
+        """
+        Dynamically imports a class from a string 'module.path:ClassName'.
+        """
+        try:
+            module_name, class_name = class_path.split(":")
+        except ValueError:
+            raise ValueError(f"Invalid format '{class_path}'. Expected 'module:Class'")
+
+        try:
+            # 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
+        except (ImportError, AttributeError) as e:
+            raise RuntimeError(f"Could not load service class '{class_path}': {e}")
+
+    # --- RPC Exposed Methods ---
+
+    def rpc_provision(self, class_path: str, tenant_id: str) -> Dict:
+        """
+        Idempotent method to start a global service (if needed) and create a tenant.
+        """
+        with self._lock:
+            # 1. Lazy Load & Start Global
+            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
+            print(f"[Coordinator] Provisioning tenant '{tenant_id}' on {class_path}")
+            try:
+                config = service.create_tenant(tenant_id)
+                return config
+            except Exception as e:
+                print(f"[Coordinator] Failed to create tenant {tenant_id}: {e}")
+                raise e
+
+    def rpc_deprovision(self, class_path: str, tenant_id: str) -> bool:
+        """
+        Removes a tenant.
+        """
+        print(f"[Coordinator] De-Provisioning tenant '{tenant_id}' on {class_path}")
+        with self._lock:
+            service = self._services.get(class_path)
+            if service:
+                try:
+                    service.remove_tenant(tenant_id)
+                    return True
+                except Exception as e:
+                    print(f"[Coordinator] Error removing tenant {tenant_id}: {e}")
+        return False
+
+    def teardown_all(self):
+        """
+        Stop all global services.
+        """
+        print("\n[Coordinator] Shutting down all global resources...")
+        for name, service in self._services.items():
+            try:
+                print(f"[Coordinator] Stopping {name}")
+                service.stop_global()
+            except Exception as e:
+                print(f"[Coordinator] Error stopping {name}: {e}")

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

@@ -0,0 +1,108 @@
+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"):
+        manager = ServiceManager()
+        
+        # Bind to port 0 (ephemeral)
+        server = SimpleXMLRPCServer(
+            ("localhost", 0), 
+            requestHandler=QuietXMLRPCRequestHandler, 
+            allow_none=True,
+            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.
+    """
+    if hasattr(config, "infra_manager"):
+        config.infra_manager.teardown_all()
+
+
+# --- Fixture (Runs on Workers) ---
+
+@pytest.fixture(scope="session")
+def coordinator_client(request):
+    """
+    Returns the XML-RPC client to talk to the manager.
+    """
+    if hasattr(request.config, "workerinput"):
+        addr = request.config.workerinput["infra_rpc_addr"]
+    else:
+        # We are running sequentially (no xdist), or we are the master
+        addr = request.config.infra_rpc_addr
+        
+    return xmlrpc.client.ServerProxy(addr)
+
+@pytest.fixture(scope="function")
+def global_resource(request, coordinator_client):
+    """
+    Factory fixture.
+    Usage: global_resource("path.to:Class")
+    """
+    
+    # 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("]", "_") 
+        # 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)
+        
+        created_resources.append((class_path, tenant_id))
+        return config
+
+    yield _provision
+
+    # Teardown logic
+    for class_path, tenant_id in reversed(created_resources):
+        try:
+            coordinator_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}")

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

@@ -0,0 +1,23 @@
+import pytest
+import psycopg2
+
+@pytest.fixture(scope="function")
+def postgres_db(global_resource):
+    """
+    User-facing fixture.
+    1. Tells coordinator to load 'tests.resources:PostgresService'
+    2. Coordinator starts Docker (if not running).
+    3. Coordinator creates a dedicated DB for this test.
+    4. Returns connection to that dedicated DB.
+    """
+    # 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