0
# External Server Integration
1

2
Fixtures for connecting to existing PostgreSQL servers without managing the server process. Ideal for containerized environments, CI systems, cloud databases, or when PostgreSQL is managed externally.
3

4
## Capabilities
5

6
### No-Process Fixture Factory
7

8
Creates fixtures that connect to existing PostgreSQL servers without starting or stopping processes.
9

10
```python { .api }
11
def postgresql_noproc(
12
    host: Optional[str] = None,
13
    port: Union[str, int, None] = None,
14
    user: Optional[str] = None,
15
    password: Optional[str] = None,
16
    dbname: Optional[str] = None,
17
    options: str = "",
18
    load: Optional[List[Union[Callable, str, Path]]] = None,
19
) -> Callable[[FixtureRequest], Iterator[NoopExecutor]]:
20
    """
21
    Create a no-process postgresql fixture factory for existing servers.
22

23
    Parameters:
24
    - host: PostgreSQL server hostname (default: 127.0.0.1)
25
    - port: PostgreSQL server port (default: 5432)
26
    - user: PostgreSQL username (default: postgres)
27
    - password: PostgreSQL password (default: None)
28
    - dbname: Database name (default: tests)
29
    - options: PostgreSQL connection options (default: "")
30
    - load: List of SQL files, Python callables, or import strings for initialization
31

32
    Returns:
33
    Function that creates NoopExecutor fixture with session scope
34
    """
35
```
36

37
### No-Operation Executor
38

39
Mock executor that provides connection details for existing PostgreSQL servers.
40

41
```python { .api }
42
class NoopExecutor:
43
    """
44
    Nooperator executor for existing PostgreSQL servers.
45
    
46
    Provides connection details and version detection without managing
47
    the server process lifecycle.
48
    """
49
    
50
    def __init__(
51
        self,
52
        host: str,
53
        port: Union[str, int],
54
        user: str,
55
        options: str,
56
        dbname: str,
57
        password: Optional[str] = None,
58
    ): ...
59
    
60
    @property
61
    def version(self) -> Any: ...
62
    @property
63
    def template_dbname(self) -> str: ...
64
    
65
    # Connection properties
66
    host: str
67
    port: int
68
    user: str
69
    password: Optional[str]
70
    dbname: str
71
    options: str
72
```
73

74
### XDist Database Naming
75

76
Helper function for pytest-xdist parallel execution support.
77

78
```python { .api }
79
def xdistify_dbname(dbname: str) -> str:
80
    """
81
    Modify database name for pytest-xdist parallel execution.
82
    
83
    Appends worker ID to database name when running with pytest-xdist
84
    to prevent conflicts between parallel test workers.
85
    
86
    Parameters:
87
    - dbname: Base database name
88
    
89
    Returns:
90
    Modified database name with worker suffix if running under xdist
91
    """
92
```
93

94
## Usage Examples
95

96
### Basic External Server Connection
97

98
```python
99
from pytest_postgresql import factories
100

101
# Connect to local PostgreSQL server
102
postgresql_noproc = factories.postgresql_noproc()
103

104
def test_external_postgresql(postgresql_noproc):
105
    """Test connection to external PostgreSQL server."""
106
    assert postgresql_noproc.host == '127.0.0.1'
107
    assert postgresql_noproc.port == 5432
108
    assert postgresql_noproc.version is not None
109
```
110

111
### Docker Container Connection
112

113
```python
114
from pytest_postgresql import factories
115

116
# Connect to PostgreSQL running in Docker container
117
postgresql_docker = factories.postgresql_noproc(
118
    host='localhost',
119
    port=5432,
120
    user='testuser',
121
    password='testpass',
122
    dbname='testdb'
123
)
124

125
def test_docker_postgresql(postgresql_docker):
126
    """Test connection to PostgreSQL in Docker."""
127
    assert postgresql_docker.host == 'localhost'
128
    assert postgresql_docker.user == 'testuser'
129
```
130

131
### Cloud Database Connection
132

133
```python
134
from pytest_postgresql import factories
135

136
# Connect to cloud PostgreSQL instance
137
postgresql_cloud = factories.postgresql_noproc(
138
    host='mydb.amazonaws.com',
139
    port=5432,
140
    user='dbuser',
141
    password='secure_password',
142
    dbname='production_test',
143
    options='sslmode=require'
144
)
145

146
def test_cloud_postgresql(postgresql_cloud):
147
    """Test connection to cloud PostgreSQL."""
148
    assert 'sslmode=require' in postgresql_cloud.options
149
```
150

151
### CI/CD Environment
152

153
```python
154
import os
155
from pytest_postgresql import factories
156

157
# Use environment variables for CI/CD
158
postgresql_ci = factories.postgresql_noproc(
159
    host=os.getenv('POSTGRES_HOST', 'localhost'),
160
    port=int(os.getenv('POSTGRES_PORT', '5432')),
161
    user=os.getenv('POSTGRES_USER', 'postgres'),
162
    password=os.getenv('POSTGRES_PASSWORD'),
163
    dbname=os.getenv('POSTGRES_DB', 'test')
164
)
165

166
def test_ci_postgresql(postgresql_ci):
167
    """Test PostgreSQL in CI environment."""
168
    # Connection details from environment
169
    pass
170
```
171

172
### External Server with Data Loading
173

174
```python
175
from pytest_postgresql import factories
176
from pathlib import Path
177

178
def setup_test_data(**kwargs):
179
    """Setup test data in external server."""
180
    import psycopg
181
    with psycopg.connect(**kwargs) as conn:
182
        with conn.cursor() as cur:
183
            cur.execute("CREATE SCHEMA IF NOT EXISTS test_schema;")
184
            cur.execute("""
185
                CREATE TABLE IF NOT EXISTS test_schema.products (
186
                    id SERIAL PRIMARY KEY,
187
                    name VARCHAR(100),
188
                    price DECIMAL(10,2)
189
                );
190
            """)
191
            cur.execute("""
192
                INSERT INTO test_schema.products (name, price) 
193
                VALUES ('Test Product', 99.99)
194
                ON CONFLICT DO NOTHING;
195
            """)
196
        conn.commit()
197

198
postgresql_with_data = factories.postgresql_noproc(
199
    host='existing-server.com',
200
    user='app_user',
201
    password='app_password',
202
    dbname='app_test',
203
    load=[setup_test_data]
204
)
205

206
def test_external_with_data(postgresql_with_data):
207
    """Test external PostgreSQL with test data loaded."""
208
    # Test data is loaded into existing server
209
    pass
210
```
211

212
### Multiple External Servers
213

214
```python
215
from pytest_postgresql import factories
216

217
# Primary database server
218
postgresql_primary = factories.postgresql_noproc(
219
    host='primary.db.com',
220
    port=5432,
221
    dbname='primary_db'
222
)
223

224
# Read replica server
225
postgresql_replica = factories.postgresql_noproc(
226
    host='replica.db.com',
227
    port=5432,
228
    dbname='primary_db'
229
)
230

231
def test_primary_replica_setup(postgresql_primary, postgresql_replica):
232
    """Test primary-replica database setup."""
233
    assert postgresql_primary.host != postgresql_replica.host
234
    assert postgresql_primary.dbname == postgresql_replica.dbname
235
```
236

237
### XDist Parallel Testing
238

239
```python
240
from pytest_postgresql import factories
241
from pytest_postgresql.factories.noprocess import xdistify_dbname
242

243
# Database name will be modified for parallel workers
244
postgresql_parallel = factories.postgresql_noproc(
245
    host='shared-server.com',
246
    dbname='test_db'  # Becomes test_db_gw0, test_db_gw1, etc. with xdist
247
)
248

249
def test_parallel_execution(postgresql_parallel):
250
    """Test parallel execution with xdist."""
251
    # Each worker gets its own database name
252
    db_name = xdistify_dbname('test_db')
253
    # db_name will include worker ID if running under xdist
254
```
255

256
## Configuration Considerations
257

258
### Connection Security
259

260
```python
261
# SSL configuration for secure connections
262
postgresql_secure = factories.postgresql_noproc(
263
    host='secure-db.com',
264
    port=5432,
265
    user='secure_user',
266
    password='secure_password',
267
    options='sslmode=require sslcert=/path/to/client.crt sslkey=/path/to/client.key'
268
)
269
```
270

271
### Connection Pooling
272

273
```python
274
# Connection with pooling options
275
postgresql_pooled = factories.postgresql_noproc(
276
    host='pooled-db.com',
277
    options='application_name=pytest_app connect_timeout=10'
278
)
279
```
280

281
### Version Detection
282

283
```python
284
def test_version_detection(postgresql_noproc):
285
    """Test PostgreSQL version detection."""
286
    version = postgresql_noproc.version
287
    assert version is not None
288
    print(f"Connected to PostgreSQL version: {version}")
289
```