Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions newsfragments/1225.feature.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Convert Config kept in TypedDict into a dataclass
52 changes: 44 additions & 8 deletions pytest_postgresql/config.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,16 @@
"""Plugin's configuration."""

from dataclasses import dataclass
from pathlib import Path
from typing import Any, TypedDict
from typing import Any

from _pytest._py.path import LocalPath
from pytest import FixtureRequest


class PostgresqlConfigDict(TypedDict):
"""Typed Config dictionary."""
@dataclass
class PostgreSQLConfig:
"""PostgreSQL Config."""

exec: str
host: str
Expand All @@ -25,16 +27,39 @@ class PostgresqlConfigDict(TypedDict):
drop_test_database: bool


def get_config(request: FixtureRequest) -> PostgresqlConfigDict:
"""Return a dictionary with config options."""
def get_config(request: FixtureRequest) -> PostgreSQLConfig:
"""Create a PostgreSQLConfig populated from pytest configuration options.

Reads pytest options and INI values prefixed with "postgresql_" to populate a PostgreSQLConfig dataclass. The "load" option is normalized to Paths or strings and "port_search_count" is converted to an int.

Parameters
----------
request (FixtureRequest): pytest fixture request used to read config options and INI values.

Returns
-------
PostgreSQLConfig: Configuration populated from the pytest settings.

"""

def get_postgresql_option(option: str) -> Any:
"""Retrieve a PostgreSQL-related pytest configuration value.

Parameters
----------
option (str): The suffix of the configuration name (without the "postgresql_" prefix).

Returns
-------
The value of the pytest configuration option named "postgresql_<option>", or `None` if not set.

"""
name = "postgresql_" + option
return request.config.getoption(name) or request.config.getini(name)

load_paths = detect_paths(get_postgresql_option("load"))
load_paths: list[Path | str] = detect_paths(get_postgresql_option("load"))

return PostgresqlConfigDict(
cfg = PostgreSQLConfig(
exec=get_postgresql_option("exec"),
host=get_postgresql_option("host"),
port=get_postgresql_option("port"),
Expand All @@ -50,10 +75,21 @@ def get_postgresql_option(option: str) -> Any:
postgres_options=get_postgresql_option("postgres_options"),
drop_test_database=request.config.getoption("postgresql_drop_test_database"),
)
return cfg


def detect_paths(load_paths: list[LocalPath | str]) -> list[Path | str]:
"""Convert path to sql files to Path instances."""
"""Normalize a sequence of load paths so SQL file paths are Path objects and other entries are preserved.

Parameters
----------
load_paths (list[LocalPath | str]): Iterable of paths to normalize; entries may be pytest LocalPath objects or strings.

Returns
-------
list[Path | str]: A new list where entries that refer to files ending with ".sql" are returned as pathlib.Path objects and all other entries are returned unchanged (strings).

"""
converted_load_paths: list[Path | str] = []
for path in load_paths:
if isinstance(path, LocalPath):
Expand Down
33 changes: 23 additions & 10 deletions pytest_postgresql/factories/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,21 +35,34 @@ def postgresql(
dbname: str | None = None,
isolation_level: "psycopg.IsolationLevel | None" = None,
) -> Callable[[FixtureRequest], Iterator[Connection]]:
"""Return connection fixture factory for PostgreSQL.
"""Create a pytest fixture factory that yields a PostgreSQL connection.

Parameters
----------
process_fixture_name (str): Name of the pytest fixture that provides the database process executor (used to obtain host, port, user, password, template DB name, and server version).
dbname (str | None): Database name to connect to; if None, use the executor's database name.
isolation_level (psycopg.IsolationLevel | None): Optional transaction isolation level to configure the janitor; if None, use the server default.

Returns
-------
Callable[[FixtureRequest], Iterator[psycopg.Connection]]: A pytest fixture factory function which, when used in a test, yields an open psycopg Connection to the specified database and ensures database janitor lifecycle management around the connection.

:param process_fixture_name: name of the process fixture
:param dbname: database name
:param isolation_level: optional postgresql isolation level
defaults to server's default
:returns: function which makes a connection to postgresql
"""

@pytest.fixture
def postgresql_factory(request: FixtureRequest) -> Iterator[Connection]:
"""Fixture factory for PostgreSQL.
"""Provide a pytest fixture that yields a psycopg Connection to the test PostgreSQL database.

The fixture resolves the process executor and global config from the given request, prepares or drops the test database as configured, and manages the database janitor and connection lifecycle so the connection is open for the duration of the consuming test.

Parameters
----------
request (FixtureRequest): Pytest fixture request used to obtain the process fixture and test configuration.

Returns
-------
Connection: A psycopg Connection connected to the selected test database; the connection is closed after the fixture completes.

:param request: fixture request object
:returns: postgresql client
"""
proc_fixture: PostgreSQLExecutor | NoopExecutor = request.getfixturevalue(process_fixture_name)
config = get_config(request)
Expand All @@ -70,7 +83,7 @@ def postgresql_factory(request: FixtureRequest) -> Iterator[Connection]:
password=pg_password,
isolation_level=isolation_level,
)
if config["drop_test_database"]:
if config.drop_test_database:
janitor.drop()
with janitor:
db_connection: Connection = psycopg.connect(
Expand Down
58 changes: 37 additions & 21 deletions pytest_postgresql/factories/noprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,34 +46,50 @@ def postgresql_noproc(
options: str = "",
load: list[Callable | str | Path] | None = None,
) -> Callable[[FixtureRequest], Iterator[NoopExecutor]]:
"""Postgresql noprocess factory.

:param host: hostname
:param port: exact port (e.g. '8000', 8000)
:param user: postgresql username
:param password: postgresql password
:param dbname: postgresql database name
:param options: Postgresql connection options
:param load: List of functions used to initialize database's template.
:returns: function which makes a postgresql process
"""Create a pytest session-scoped fixture that provides a NoopExecutor connected to an existing PostgreSQL server.

The returned fixture resolves connection parameters from the explicit arguments or from the test configuration, applies xdist worker-specific adjustment to the database name, and uses a DatabaseJanitor to optionally drop the test database and load initialization elements into the template before yielding the configured NoopExecutor.

Parameters
----------
host (str | None): Hostname to connect to; if None, taken from test configuration.
port (str | int | None): Port to connect to; if None, taken from configuration or defaults to 5432.
user (str | None): Username to authenticate as; if None, taken from configuration.
password (str | None): Password to authenticate with; if None, taken from configuration.
dbname (str | None): Base database name; if None, taken from configuration. The name is adjusted when pytest-xdist is in use.
options (str): Additional connection options; if empty, taken from configuration.
load (list[Callable | str | Path] | None): Sequence of initialization elements (callables or filesystem paths) to load into the database template; if None, taken from configuration.

Returns
-------
Callable[[FixtureRequest], Iterator[NoopExecutor]]: A pytest fixture function which yields a configured NoopExecutor instance.

"""

@pytest.fixture(scope="session")
def postgresql_noproc_fixture(request: FixtureRequest) -> Iterator[NoopExecutor]:
"""Noop Process fixture for PostgreSQL.
"""Provide a pytest fixture that yields a NoopExecutor configured for an existing PostgreSQL server.

The fixture resolves connection parameters from the fixture request and the factory's closure values, applies xdist-aware database name transformation, and uses a DatabaseJanitor context to optionally drop the test database (if configured) and load initialization elements into the database template before yielding the executor.

Parameters
----------
request (FixtureRequest): Pytest fixture request used to obtain configuration.

Returns
-------
noop_exec (NoopExecutor): Executor-like object configured with the resolved host, port, user, password, dbname, and options.

:param request: fixture request object
:returns: tcp executor-like object
"""
config = get_config(request)
pg_host = host or config["host"]
pg_port = port or config["port"] or 5432
pg_user = user or config["user"]
pg_password = password or config["password"]
pg_dbname = xdistify_dbname(dbname or config["dbname"])
pg_options = options or config["options"]
pg_load = load or config["load"]
drop_test_database = config["drop_test_database"]
pg_host = host or config.host
pg_port = port or config.port or 5432
pg_user = user or config.user
pg_password = password or config.password
pg_dbname = xdistify_dbname(dbname or config.dbname)
pg_options = options or config.options
pg_load = load or config.load
drop_test_database = config.drop_test_database

noop_exec = NoopExecutor(
host=pg_host,
Expand Down
118 changes: 76 additions & 42 deletions pytest_postgresql/factories/process.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,17 +28,36 @@
from port_for import PortForException, get_port
from pytest import FixtureRequest, TempPathFactory

from pytest_postgresql.config import PostgresqlConfigDict, get_config
from pytest_postgresql.config import PostgreSQLConfig, get_config
from pytest_postgresql.exceptions import ExecutableMissingException
from pytest_postgresql.executor import PostgreSQLExecutor
from pytest_postgresql.janitor import DatabaseJanitor

PortType = port_for.PortType # mypy requires explicit export


def _pg_exe(executable: str | None, config: PostgresqlConfigDict) -> str:
"""If executable is set, use it. Otherwise best effort to find the executable."""
postgresql_ctl = executable or config["exec"]
def _pg_exe(executable: str | None, config: PostgreSQLConfig) -> str:
"""Resolve the filesystem path to the PostgreSQL control executable (pg_ctl).

If `executable` is provided, it is returned as-is. Otherwise the function uses
`config.exec` if that path exists; if not, it attempts to locate `pg_ctl` using
`pg_config --bindir` and returns the `pg_ctl` path from that bindir.

Parameters
----------
executable (str | None): Explicit path to a pg_ctl-like executable, or None to auto-resolve.
config (PostgreSQLConfig): Configuration providing a fallback executable path via `config.exec`.

Returns
-------
str: Absolute path to the pg_ctl executable to use.

Raises
------
ExecutableMissingException: If neither an existing executable path nor `pg_config` can be found.

"""
postgresql_ctl = executable or config.exec
# check if that executable exists, as it's no on systems' PATH
# only replace it if executable isn't passed manually
if not os.path.exists(postgresql_ctl) and executable is None:
Expand All @@ -50,9 +69,21 @@ def _pg_exe(executable: str | None, config: PostgresqlConfigDict) -> str:
return postgresql_ctl


def _pg_port(port: PortType | None, config: PostgresqlConfigDict, excluded_ports: Iterable[int]) -> int:
"""User specified port, otherwise find an unused port from config."""
pg_port = get_port(port, excluded_ports) or get_port(config["port"], excluded_ports)
def _pg_port(port: PortType | None, config: PostgreSQLConfig, excluded_ports: Iterable[int]) -> int:
"""Select the PostgreSQL port to use, preferring an explicit port and falling back to the configured port.

Parameters
----------
port (PortType | None): Preferred port provided by the caller; may be None.
config (PostgreSQLConfig): Configuration containing the default port to use when `port` is not specified.
excluded_ports (Iterable[int]): Ports that must not be selected.

Returns
-------
int: A port number that is not in `excluded_ports`.

"""
pg_port = get_port(port, excluded_ports) or get_port(config.port, excluded_ports)
assert pg_port is not None
return pg_port

Expand Down Expand Up @@ -82,41 +113,44 @@ def postgresql_proc(
postgres_options: str | None = None,
load: list[Callable | str | Path] | None = None,
) -> Callable[[FixtureRequest, TempPathFactory], Iterator[PostgreSQLExecutor]]:
"""Postgresql process factory.

:param executable: path to postgresql_ctl
:param host: hostname
:param port:
exact port (e.g. '8000', 8000)
randomly selected port (None) - any random available port
-1 - command line or pytest.ini configured port
[(2000,3000)] or (2000,3000) - random available port from a given range
[{4002,4003}] or {4002,4003} - random of 4002 or 4003 ports
[(2000,3000), {4002,4003}] - random of given range and set
:param user: postgresql username
:param password: postgresql password
:param dbname: postgresql database name
:param options: Postgresql connection options
:param startparams: postgresql starting parameters
:param unixsocketdir: directory to create postgresql's unixsockets
:param postgres_options: Postgres executable options for use by pg_ctl
:param load: List of functions used to initialize database's template.
:returns: function which makes a postgresql process
"""Create a pytest fixture factory that starts a temporary PostgreSQL server process for tests.

This factory returns a session-scoped fixture which allocates a port, initializes a data directory, starts PostgreSQL, runs initial load steps into the template database, and yields a PostgreSQLExecutor for test use. The fixture ensures the server is stopped and cleaned up when tests finish.

Parameters
----------
executable (str | None): Path to the PostgreSQL control executable (pg_ctl). If None, the configured executable or pg_config discovery will be used.
port (PortType | None | int): Port selection specification. Accepts:
- an exact port (e.g. 8000 or "8000"),
- None to select any available port,
- -1 to use the command-line or pytest.ini configured port,
- a range tuple/list (e.g. (2000, 3000)) to pick a random available port from that range,
- a set/list of ports (e.g. {4002, 4003}) to pick a random port from the set,
- a list combining ranges and sets (e.g. [(2000,3000), {4002,4003}]).
postgres_options (str | None): Additional options for the PostgreSQL server process passed through pg_ctl.
load (list[Callable | str | Path] | None): Initialization steps applied to the template database before tests run; each element is either a callable or a path/SQL identifier that DatabaseJanitor.load understands.

Returns
-------
Callable[[FixtureRequest, TempPathFactory], Iterator[PostgreSQLExecutor]]: A pytest fixture factory that yields a started PostgreSQLExecutor configured per the provided arguments and test configuration.

"""

@pytest.fixture(scope="session")
def postgresql_proc_fixture(
request: FixtureRequest, tmp_path_factory: TempPathFactory
) -> Iterator[PostgreSQLExecutor]:
"""Process fixture for PostgreSQL.
"""Create, start, and yield a PostgreSQL server process configured for the requesting test.

This fixture selects an available port, prepares a data directory and logfile, starts a PostgreSQL server via PostgreSQLExecutor, applies any configured initialization/load steps, and yields the running executor to the test. The server is stopped and resources are cleaned up when the fixture context exits.

Returns:
PostgreSQLExecutor: A configured and started executor connected to the test PostgreSQL instance.

:param request: fixture request object
:param tmp_path_factory: temporary path object (fixture)
:returns: tcp executor
"""
config = get_config(request)
pg_dbname = dbname or config["dbname"]
pg_load = load or config["load"]
pg_dbname = dbname or config.dbname
pg_load = load or config.load
postgresql_ctl = _pg_exe(executable, config)
port_path = tmp_path_factory.getbasetemp()
if hasattr(request.config, "workerinput"):
Expand All @@ -138,7 +172,7 @@ def postgresql_proc_fixture(
port_file.write(f"pg_port {pg_port}\n")
break
except FileExistsError:
if n >= config["port_search_count"]:
if n >= config.port_search_count:
raise PortForException(
f"Attempted {n} times to select ports. "
f"All attempted ports: {', '.join(map(str, used_ports))} are already "
Expand All @@ -151,17 +185,17 @@ def postgresql_proc_fixture(

postgresql_executor = PostgreSQLExecutor(
executable=postgresql_ctl,
host=host or config["host"],
host=host or config.host,
port=pg_port,
user=user or config["user"],
password=password or config["password"],
user=user or config.user,
password=password or config.password,
dbname=pg_dbname,
options=options or config["options"],
options=options or config.options,
datadir=str(datadir),
unixsocketdir=unixsocketdir or config["unixsocketdir"],
unixsocketdir=unixsocketdir or config.unixsocketdir,
logfile=str(logfile_path),
startparams=startparams or config["startparams"],
postgres_options=postgres_options or config["postgres_options"],
startparams=startparams or config.startparams,
postgres_options=postgres_options or config.postgres_options,
)
# start server
with postgresql_executor:
Expand All @@ -174,7 +208,7 @@ def postgresql_proc_fixture(
version=postgresql_executor.version,
password=postgresql_executor.password,
)
if config["drop_test_database"]:
if config.drop_test_database:
janitor.drop()
with janitor:
for load_element in pg_load:
Expand Down
2 changes: 1 addition & 1 deletion tests/examples/test_assert_port_search_count_is_ten.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,4 +11,4 @@
def test_assert_port_search_count_is_ten(request: FixtureRequest) -> None:
"""Asserts that port_search_count is 10."""
config = get_config(request)
assert config["port_search_count"] == 10
assert config.port_search_count == 10
Loading