audit_trail.testing

Test helpers for applications that use audit_trail.

assert_audited checks that an entry with the given fields was written and assert_not_audited that none was; aassert_audited and aassert_not_audited do the same on async binds. A failed assertion lists the closest entries and the fields that differ.

create_test_tables, drop_test_tables and clear_audit_entries set up the audit tables on a test database whose migrations do not create them. The pytest fixtures built on them live in audit_trail.pytest_plugin; enable them in the root conftest.py::

pytest_plugins = ["audit_trail.pytest_plugin"]

This module does not import pytest.

UNSET module-attribute

UNSET: Final = Unset.UNSET

Filter not given: the field is not checked. None means NULL.

Unset

Bases: Enum

Type of UNSET, the default of every assertion filter.

assert_audited

assert_audited(
    audit: AuditTrail,
    bind: Session | Connection | Engine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> list[ActivityRow]

Assert that at least one audit entry matches every given filter.

Reads the raw audit_activity rows as written, not compacted. A Session is flushed first, so its pending changes are audited. A Session or Connection also sees rows not committed yet; an Engine reads on a new connection and sees committed rows only.

Values are compared with their stored JSON form: a datetime is an ISO string, a redacted value "***". unittest.mock.ANY matches any value, e.g. changes={"status": [ANY, "paid"]}.

Parameters:

Name	Type	Description	Default
audit	AuditTrail	The AuditTrail whose tables are read.	required
bind	Session | Connection | Engine	Session, connection or engine to read with.	required
verb	str | Unset	The verb; an AuditEvent member works too.	UNSET
obj	object	The object: a mapped instance, a Target or a (type, id) tuple.	UNSET
target	object	The parent object, like obj; None means no target.	UNSET
actor_id	str | Unset | None	The actor id; None means no actor.	UNSET
scope_id	str | Unset | None	The scope id; None means no scope.	UNSET
severity	int | Unset	The severity.	UNSET
changes	Mapping[str, object] | Unset	Changes the entry must contain; other keys are ignored. A column change is [old, new], a relationship change {"added": [...], "removed": [...]}.	UNSET
payload	Mapping[str, object] | Unset	Payload fields the entry must contain; other keys are ignored.	UNSET

Returns:

Type	Description
list[ActivityRow]	The matching entries, oldest first.

Raises:

Type	Description
AssertionError	No entry matches; the message lists the closest ones and how they differ.
ValueError	obj or target has no primary key, or obj is None.

Source code in audit_trail/testing.py

def assert_audited(
    audit: AuditTrail,
    bind: Session | Connection | Engine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> list[ActivityRow]:
    """Assert that at least one audit entry matches every given filter.

    Reads the raw ``audit_activity`` rows as written, not compacted. A
    ``Session`` is flushed first, so its pending changes are audited. A
    ``Session`` or ``Connection`` also sees rows not committed yet; an
    ``Engine`` reads on a new connection and sees committed rows only.

    Values are compared with their stored JSON form: a datetime is an ISO
    string, a redacted value ``"***"``. ``unittest.mock.ANY`` matches any
    value, e.g. ``changes={"status": [ANY, "paid"]}``.

    Args:
        audit: The ``AuditTrail`` whose tables are read.
        bind: Session, connection or engine to read with.
        verb: The verb; an ``AuditEvent`` member works too.
        obj: The object: a mapped instance, a ``Target`` or a
            ``(type, id)`` tuple.
        target: The parent object, like ``obj``; ``None`` means no target.
        actor_id: The actor id; ``None`` means no actor.
        scope_id: The scope id; ``None`` means no scope.
        severity: The severity.
        changes: Changes the entry must contain; other keys are ignored.
            A column change is ``[old, new]``, a relationship change
            ``{"added": [...], "removed": [...]}``.
        payload: Payload fields the entry must contain; other keys are
            ignored.

    Returns:
        The matching entries, oldest first.

    Raises:
        AssertionError: No entry matches; the message lists the closest ones
            and how they differ.
        ValueError: ``obj`` or ``target`` has no primary key, or ``obj`` is
            ``None``.
    """
    __tracebackhide__ = True
    filters = _Filters(
        verb, obj, target, actor_id, scope_id, severity, changes, payload
    )
    return _check_audited(_run(bind, lambda sync: _search(audit, sync, filters)))

assert_not_audited

assert_not_audited(
    audit: AuditTrail,
    bind: Session | Connection | Engine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> None

Assert that no audit entry matches every given filter.

Reads as assert_audited does, with the same filters.

Parameters:

Name	Type	Description	Default
audit	AuditTrail	See assert_audited.	required
bind	Session | Connection | Engine	See assert_audited.	required
verb	str | Unset	See assert_audited.	UNSET
obj	object	See assert_audited.	UNSET
target	object	See assert_audited.	UNSET
actor_id	str | Unset | None	See assert_audited.	UNSET
scope_id	str | Unset | None	See assert_audited.	UNSET
severity	int | Unset	See assert_audited.	UNSET
changes	Mapping[str, object] | Unset	See assert_audited.	UNSET
payload	Mapping[str, object] | Unset	See assert_audited.	UNSET

Raises:

Type	Description
AssertionError	An entry matches; the message lists the matches.
ValueError	See assert_audited.

Source code in audit_trail/testing.py

def assert_not_audited(
    audit: AuditTrail,
    bind: Session | Connection | Engine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> None:
    """Assert that no audit entry matches every given filter.

    Reads as ``assert_audited`` does, with the same filters.

    Args:
        audit: See ``assert_audited``.
        bind: See ``assert_audited``.
        verb: See ``assert_audited``.
        obj: See ``assert_audited``.
        target: See ``assert_audited``.
        actor_id: See ``assert_audited``.
        scope_id: See ``assert_audited``.
        severity: See ``assert_audited``.
        changes: See ``assert_audited``.
        payload: See ``assert_audited``.

    Raises:
        AssertionError: An entry matches; the message lists the matches.
        ValueError: See ``assert_audited``.
    """
    __tracebackhide__ = True
    filters = _Filters(
        verb, obj, target, actor_id, scope_id, severity, changes, payload
    )
    _check_not_audited(_run(bind, lambda sync: _search(audit, sync, filters)))

aassert_audited async

aassert_audited(
    audit: AuditTrail,
    bind: AsyncSession | AsyncConnection | AsyncEngine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> list[ActivityRow]

Async assert_audited, for an async session, connection or engine.

Parameters:

Name	Type	Description	Default
audit	AuditTrail	See assert_audited.	required
bind	AsyncSession | AsyncConnection | AsyncEngine	Async session, connection or engine to read with.	required
verb	str | Unset	See assert_audited.	UNSET
obj	object	See assert_audited.	UNSET
target	object	See assert_audited.	UNSET
actor_id	str | Unset | None	See assert_audited.	UNSET
scope_id	str | Unset | None	See assert_audited.	UNSET
severity	int | Unset	See assert_audited.	UNSET
changes	Mapping[str, object] | Unset	See assert_audited.	UNSET
payload	Mapping[str, object] | Unset	See assert_audited.	UNSET

Returns:

Type	Description
list[ActivityRow]	The matching entries, oldest first.

Raises:

Type	Description
AssertionError	See assert_audited.
ValueError	See assert_audited.

Source code in audit_trail/testing.py

async def aassert_audited(
    audit: AuditTrail,
    bind: AsyncSession | AsyncConnection | AsyncEngine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> list[ActivityRow]:
    """Async ``assert_audited``, for an async session, connection or engine.

    Args:
        audit: See ``assert_audited``.
        bind: Async session, connection or engine to read with.
        verb: See ``assert_audited``.
        obj: See ``assert_audited``.
        target: See ``assert_audited``.
        actor_id: See ``assert_audited``.
        scope_id: See ``assert_audited``.
        severity: See ``assert_audited``.
        changes: See ``assert_audited``.
        payload: See ``assert_audited``.

    Returns:
        The matching entries, oldest first.

    Raises:
        AssertionError: See ``assert_audited``.
        ValueError: See ``assert_audited``.
    """
    __tracebackhide__ = True
    filters = _Filters(
        verb, obj, target, actor_id, scope_id, severity, changes, payload
    )
    result = await _arun(bind, lambda sync: _search(audit, sync, filters))
    return _check_audited(result)

aassert_not_audited async

aassert_not_audited(
    audit: AuditTrail,
    bind: AsyncSession | AsyncConnection | AsyncEngine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> None

Async assert_not_audited, for an async session, connection or engine.

Parameters:

Name	Type	Description	Default
audit	AuditTrail	See assert_audited.	required
bind	AsyncSession | AsyncConnection | AsyncEngine	Async session, connection or engine to read with.	required
verb	str | Unset	See assert_audited.	UNSET
obj	object	See assert_audited.	UNSET
target	object	See assert_audited.	UNSET
actor_id	str | Unset | None	See assert_audited.	UNSET
scope_id	str | Unset | None	See assert_audited.	UNSET
severity	int | Unset	See assert_audited.	UNSET
changes	Mapping[str, object] | Unset	See assert_audited.	UNSET
payload	Mapping[str, object] | Unset	See assert_audited.	UNSET

Raises:

Type	Description
AssertionError	See assert_not_audited.
ValueError	See assert_audited.

Source code in audit_trail/testing.py

async def aassert_not_audited(
    audit: AuditTrail,
    bind: AsyncSession | AsyncConnection | AsyncEngine,
    *,
    verb: str | Unset = UNSET,
    obj: object = UNSET,
    target: object = UNSET,
    actor_id: str | Unset | None = UNSET,
    scope_id: str | Unset | None = UNSET,
    severity: int | Unset = UNSET,
    changes: Mapping[str, object] | Unset = UNSET,
    payload: Mapping[str, object] | Unset = UNSET,
) -> None:
    """Async ``assert_not_audited``, for an async session, connection or engine.

    Args:
        audit: See ``assert_audited``.
        bind: Async session, connection or engine to read with.
        verb: See ``assert_audited``.
        obj: See ``assert_audited``.
        target: See ``assert_audited``.
        actor_id: See ``assert_audited``.
        scope_id: See ``assert_audited``.
        severity: See ``assert_audited``.
        changes: See ``assert_audited``.
        payload: See ``assert_audited``.

    Raises:
        AssertionError: See ``assert_not_audited``.
        ValueError: See ``assert_audited``.
    """
    __tracebackhide__ = True
    filters = _Filters(
        verb, obj, target, actor_id, scope_id, severity, changes, payload
    )
    _check_not_audited(await _arun(bind, lambda sync: _search(audit, sync, filters)))

create_test_tables

create_test_tables(
    audit: AuditTrail,
    connection: Connection,
    *,
    months_ahead: int = 1,
    now: datetime | None = None,
) -> None

Create the audit tables and their partitions on a test database.

For databases whose migrations do not create the audit tables: it fails when they exist. Runs in the connection's current transaction, which must not be AUTOCOMMIT; the caller commits.

Parameters:

Name	Type	Description	Default
audit	AuditTrail	The AuditTrail whose tables are created.	required
connection	Connection	Connection with DDL privileges.	required
months_ahead	int	Monthly partitions to create after the current month.	1
now	datetime | None	Reference time for the partitions. None uses the database's now().	None

Source code in audit_trail/testing.py

def create_test_tables(
    audit: AuditTrail,
    connection: Connection,
    *,
    months_ahead: int = 1,
    now: datetime | None = None,
) -> None:
    """Create the audit tables and their partitions on a test database.

    For databases whose migrations do not create the audit tables: it fails
    when they exist. Runs in the connection's current transaction, which must
    not be ``AUTOCOMMIT``; the caller commits.

    Args:
        audit: The ``AuditTrail`` whose tables are created.
        connection: Connection with DDL privileges.
        months_ahead: Monthly partitions to create after the current month.
        now: Reference time for the partitions. ``None`` uses the database's
            ``now()``.
    """
    create_audit_tables(connection, audit.tables, audit.severities)
    ensure_partitions(
        connection,
        audit.tables,
        audit.severities,
        months_ahead=months_ahead,
        now=now,
    )

drop_test_tables

drop_test_tables(
    audit: AuditTrail, connection: Connection
) -> None

Drop the audit tables with their partitions; the schema is kept.

Parameters:

Name	Type	Description	Default
audit	AuditTrail	The AuditTrail whose tables are dropped.	required
connection	Connection	Connection with DDL privileges.	required

Source code in audit_trail/testing.py

def drop_test_tables(audit: AuditTrail, connection: Connection) -> None:
    """Drop the audit tables with their partitions; the schema is kept.

    Args:
        audit: The ``AuditTrail`` whose tables are dropped.
        connection: Connection with DDL privileges.
    """
    drop_audit_tables(connection, audit.tables)

clear_audit_entries

clear_audit_entries(
    audit: AuditTrail, connection: Connection
) -> None

Delete every audit entry and transaction row with TRUNCATE.

Parameters:

Name	Type	Description	Default
audit	AuditTrail	The AuditTrail whose tables are emptied.	required
connection	Connection	Connection to run on.	required

Source code in audit_trail/testing.py

def clear_audit_entries(audit: AuditTrail, connection: Connection) -> None:
    """Delete every audit entry and transaction row with ``TRUNCATE``.

    Args:
        audit: The ``AuditTrail`` whose tables are emptied.
        connection: Connection to run on.
    """
    names = ", ".join(
        qualified_name(table.schema, table.name)
        for table in (audit.tables.activity, audit.tables.transaction)
    )
    connection.exec_driver_sql(
        f"TRUNCATE {names}", execution_options={"no_parameters": True}
    )