0
# Process Management
1

2
PostgreSQL process fixture factory and executor for starting, managing, and stopping PostgreSQL server instances during testing. Provides full control over server configuration, data directory management, and process lifecycle.
3

4
## Capabilities
5

6
### PostgreSQL Process Factory
7

8
Creates session-scoped fixtures that start and manage PostgreSQL server processes with configurable parameters.
9

10
```python { .api }
11
def postgresql_proc(
12
    executable: Optional[str] = None,
13
    host: Optional[str] = None,
14
    port: Optional[PortType] = -1,
15
    user: Optional[str] = None,
16
    password: Optional[str] = None,
17
    dbname: Optional[str] = None,
18
    options: str = "",
19
    startparams: Optional[str] = None,
20
    unixsocketdir: Optional[str] = None,
21
    postgres_options: Optional[str] = None,
22
    load: Optional[List[Union[Callable, str, Path]]] = None,
23
) -> Callable[[FixtureRequest, TempPathFactory], Iterator[PostgreSQLExecutor]]:
24
    """
25
    Create a postgresql process fixture factory.
26

27
    Parameters:
28
    - executable: Path to pg_ctl executable (default: /usr/lib/postgresql/13/bin/pg_ctl)
29
    - host: Host address for connections (default: 127.0.0.1)
30
    - port: Port configuration - exact port, random, range, or set (default: -1 for auto-detect)
31
    - user: PostgreSQL username (default: postgres)
32
    - password: PostgreSQL password (default: None)
33
    - dbname: Default database name (default: tests)
34
    - options: PostgreSQL connection options (default: "")
35
    - startparams: PostgreSQL server start parameters (default: "-w")
36
    - unixsocketdir: Unix socket directory (default: temp directory)
37
    - postgres_options: Additional postgres executable options (default: "")
38
    - load: List of SQL files, Python callables, or import strings for initialization
39

40
    Returns:
41
    Function that creates PostgreSQLExecutor fixture with session scope
42
    """
43
```
44

45
### PostgreSQL Executor
46

47
The executor class that manages the PostgreSQL server process lifecycle.
48

49
```python { .api }
50
class PostgreSQLExecutor(TCPExecutor):
51
    """
52
    PostgreSQL executor running on pg_ctl.
53
    
54
    Manages PostgreSQL server process including data directory initialization,
55
    server startup/shutdown, and connection management.
56
    """
57
    
58
    def __init__(
59
        self,
60
        executable: str,
61
        host: str,
62
        port: Union[str, int],
63
        user: str,
64
        options: str,
65
        startparams: str,
66
        unixsocketdir: str,
67
        dbname: str,
68
        postgres_options: str,
69
        password: Optional[str] = None,
70
    ): ...
71
    
72
    def start(self) -> "PostgreSQLExecutor": ...
73
    def stop(self) -> "PostgreSQLExecutor": ...
74
    def init_directory(self) -> None: ...
75
    def clean_directory(self) -> None: ...
76
    def wait_for_postgres(self) -> None: ...
77
    def running(self) -> bool: ...
78
    
79
    @property
80
    def version(self) -> Any: ...
81
    @property
82
    def template_dbname(self) -> str: ...
83
```
84

85
### Port Configuration
86

87
```python { .api }
88
PortType = port_for.PortType
89
# Supports: exact port (int), random (None), port range (tuple), or port set (list)
90
```
91

92
## Usage Examples
93

94
### Basic Process Fixture
95

96
```python
97
from pytest_postgresql import factories
98

99
# Create basic process fixture
100
postgresql_proc = factories.postgresql_proc()
101

102
def test_postgresql_server(postgresql_proc):
103
    """Test that PostgreSQL server is running."""
104
    assert postgresql_proc.running() is True
105
    assert postgresql_proc.version is not None
106
```
107

108
### Custom Configuration
109

110
```python
111
from pytest_postgresql import factories
112

113
# Custom configuration with specific port and user
114
custom_postgresql_proc = factories.postgresql_proc(
115
    port=5433,
116
    user='test_user',
117
    dbname='test_db',
118
    unixsocketdir='/tmp/postgresql'
119
)
120

121
def test_custom_postgresql(custom_postgresql_proc):
122
    """Test custom PostgreSQL configuration."""
123
    assert custom_postgresql_proc.port == 5433
124
    assert custom_postgresql_proc.user == 'test_user'
125
    assert custom_postgresql_proc.dbname == 'test_db'
126
```
127

128
### Multiple Process Fixtures
129

130
```python
131
from pytest_postgresql import factories
132

133
# Different PostgreSQL instances for different test scenarios
134
postgresql_proc_main = factories.postgresql_proc(port=5432)
135
postgresql_proc_secondary = factories.postgresql_proc(port=5433)
136

137
def test_multiple_postgresql_instances(postgresql_proc_main, postgresql_proc_secondary):
138
    """Test multiple PostgreSQL instances running simultaneously."""
139
    assert postgresql_proc_main.running() is True
140
    assert postgresql_proc_secondary.running() is True
141
    assert postgresql_proc_main.port != postgresql_proc_secondary.port
142
```
143

144
### Process with Data Loading
145

146
```python
147
from pytest_postgresql import factories
148
from pathlib import Path
149

150
def init_test_schema(**kwargs):
151
    """Initialize test schema."""
152
    import psycopg
153
    with psycopg.connect(**kwargs) as conn:
154
        with conn.cursor() as cur:
155
            cur.execute("""
156
                CREATE TABLE users (
157
                    id SERIAL PRIMARY KEY,
158
                    name VARCHAR(100) NOT NULL,
159
                    email VARCHAR(100) UNIQUE
160
                );
161
            """)
162
        conn.commit()
163

164
postgresql_with_schema = factories.postgresql_proc(
165
    load=[
166
        Path('/path/to/schema.sql'),
167
        init_test_schema,
168
        'myapp.fixtures.load_test_data'
169
    ]
170
)
171

172
def test_preloaded_database(postgresql_with_schema):
173
    """Test PostgreSQL with preloaded schema."""
174
    # Schema and data are already loaded
175
    pass
176
```
177

178
### Process Management Methods
179

180
```python
181
def test_process_lifecycle(postgresql_proc):
182
    """Test PostgreSQL process lifecycle management."""
183
    # Server should be running
184
    assert postgresql_proc.running() is True
185
    
186
    # Stop the server
187
    postgresql_proc.stop()
188
    assert postgresql_proc.running() is False
189
    
190
    # Restart the server
191
    postgresql_proc.start()
192
    assert postgresql_proc.running() is True
193
    
194
    # Clean up data directory (use with caution)
195
    postgresql_proc.stop()
196
    postgresql_proc.clean_directory()
197
```
198

199
## Error Handling
200

201
```python
202
from pytest_postgresql.exceptions import ExecutableMissingException, PostgreSQLUnsupported
203

204
def test_error_handling():
205
    """Test error handling scenarios."""
206
    try:
207
        # This might fail if pg_ctl is not found
208
        postgresql_proc = factories.postgresql_proc(
209
            executable='/nonexistent/pg_ctl'
210
        )
211
    except ExecutableMissingException:
212
        # Handle missing PostgreSQL installation
213
        pass
214
    
215
    try:
216
        # This might fail with very old PostgreSQL versions
217
        postgresql_proc = factories.postgresql_proc()
218
    except PostgreSQLUnsupported:
219
        # Handle unsupported PostgreSQL version
220
        pass
221
```