ditto.snapshot¶

Snapshot fixture implementation and supporting types.

`SnapshotKey(module, group_name, key, identifier)` `dataclass` ¶

Fully-qualified identity for a single snapshot value.

Parameters:

Name	Type	Description	Default
`module`	`str`	Rootdir-relative test file stem, e.g. "tests/bar/test_api". Provides namespace isolation across files in shared backends.	required
`group_name`	`str`	Test function name, e.g. "test_something". For class-based tests includes the class prefix: "TestClass.test_something".	required
`key`	`str`	Per-snapshot identifier within the test.	required
`identifier`	`str`	Recorder identifier, e.g. "json", "yaml", "pandas.parquet".	required

`filename` `property` ¶

Short key: 'group@key.ext'. Not used as the storage key for any backend.

Kept for reference and user code that inspects SnapshotKey objects. File backends use _flat_key ('module.group@key.ext') and remote backends use str(key) ('module/group@key.ext').

`display_name` `property` ¶

Human-readable label for the session report: 'module/group@key.ext'.

`str()` ¶

Namespaced key for remote backends: 'module/group@key.ext'.

Unique across all test files in a shared backend (Redis, S3, etc.). Also used as the human-readable display name in session reports.

Source code in ditto/snapshot.py

def __str__(self) -> str:
    """Namespaced key for remote backends: 'module/group@key.ext'.

    Unique across all test files in a shared backend (Redis, S3, etc.).
    Also used as the human-readable display name in session reports.
    """
    return f"{self.module}/{self.group_name}@{self.key}.{self.identifier}"

`LockSeen(target_id, scheme, nodeid, key, recorder)` `dataclass` ¶

A lock entry observed this session, plus where it lives.

Parameters:

Name	Type	Description	Default
`target_id`	`str`	Portable lock-file target id (rootdir-relative for `file://`, URI otherwise).	required
`scheme`	`str`	The target's URI scheme, e.g. `file` or `s3`. Used to derive the storage key for this entry later.	required
`nodeid`	`str`	Full pytest node id for the owning test, e.g. `tests/test_api.py::test_foo`.	required
`key`	`str`	Per-snapshot identifier within the test.	required
`recorder`	`str`	Recorder identifier, e.g. `json`, `yaml`, `pandas.parquet`.	required

`SnapshotMode` ¶

Bases: Enum

How resolve_snapshot treats the stored value for a key.

Attributes:

Name	Type	Description
`RECORD`		Save the value when the key is absent; otherwise return the stored value.
`UPDATE`		Always save the value, overwriting a stored one. Set by `--ditto-update`.
`VERIFY`		Never write: return the stored value, or the given value when the key is absent. Set by `--ditto-verify`, so a verify run cannot recreate a deleted snapshot.
`A value that is saved, or returned under VERIFY for an absent key, is`
`serialised and deserialised first, and the deserialised value is returned.`

`Snapshot(group_name, module, target, _backend, recorder=_UNSET, recorder_name=_UNSET, mode=SnapshotMode.RECORD, nodeid='', target_id='', _tracker=_SessionTracker())` `dataclass` ¶

Immutable configuration for a snapshot: where to store it and how to record it.

Instances are created by the snapshot fixture and hold no I/O state. All persistence is handled by the module-level free functions save_snapshot, load_snapshot, and resolve_snapshot.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Prefix used in snapshot keys. Derived from the pytest nodeid minus the file path (e.g. "test_something" or "TestClass.test_something").	required
`module`	`str`	Rootdir-relative test file stem (e.g. "tests/bar/test_api"). Required for all backends. Provides namespace isolation across test files.	required
`target`	`str`	URI identifying the storage location. The scheme controls key format: `file://` uses flat dotted keys (`module.group@key.ext`); all other schemes use slash-separated keys (`module/group@key.ext`). Always use absolute `file://` URIs (e.g. `file:///home/user/proj/tests/.ditto`).	required
`_backend`	`MutableMapping[str, bytes]`	Resolved storage backend. Conventionally private — set by the fixture via `_resolve_target`. Use `target=` to communicate where data goes.	required
`recorder`	`Recorder`	Serialisation strategy. Defaults to strict JSON.	`_UNSET`
`recorder_name`	`str`	The recorder's registered name, which is its persisted identifier: it ends snapshot filenames and is recorded in `ditto.lock`. The fixture passes the name the recorder was selected by. A directly constructed `Snapshot` passes `recorder` and `recorder_name` together, or neither for strict JSON (`"json"`). The name may be omitted with the strict JSON recorder itself, whose name is `"json"`.	`_UNSET`
`mode`	`SnapshotMode`	Whether a snapshot is recorded, updated, or only verified. Defaults to `SnapshotMode.RECORD`.	`RECORD`
`nodeid`	`str`	Full pytest node id for the owning test, e.g. `tests/test_api.py::test_foo`. Used to build lock-file entries. Empty when constructed outside the fixture.	`''`
`target_id`	`str`	Portable lock-file target id (rootdir-relative for `file://`, URI otherwise).	`''`
`_tracker`	`_SessionTracker`	Where snapshot activity is recorded. The fixture passes its pytest session's tracker; a directly constructed `Snapshot` gets a private one.	`_SessionTracker()`

`call(data, key)` ¶

Save or load the snapshot for key.

Delegates to resolve_snapshot: saves data on first call and returns the stored value on subsequent calls. Either way the value returned is what the recorder reads back, not data itself.

Source code in ditto/snapshot.py

def __call__(self, data: Any, key: str) -> Any:
    """Save or load the snapshot for `key`.

    Delegates to `resolve_snapshot`: saves `data` on first call and
    returns the stored value on subsequent calls. Either way the value
    returned is what the recorder reads back, not `data` itself.
    """
    return resolve_snapshot(self, data, key)

`save_snapshot(snapshot, data, key)` ¶

Persist data to the backend as the snapshot for key.

Nothing is written if the recorder cannot deserialise the bytes it produced.

Source code in ditto/snapshot.py

def save_snapshot(snapshot: Snapshot, data: Any, key: str) -> None:
    """Persist `data` to the backend as the snapshot for `key`.

    Nothing is written if the recorder cannot deserialise the bytes it produced.
    """
    sk = snapshot._key(key)
    storage_key = snapshot._key_of()(sk)
    raw, _ = _round_trip(snapshot.recorder, data)
    snapshot._backend[storage_key] = raw

`load_snapshot(snapshot, key)` ¶

Load and return the stored snapshot value for key.

Raises:

Type	Description
`FileNotFoundError`	When no snapshot exists for `key`.

Source code in ditto/snapshot.py

def load_snapshot(snapshot: Snapshot, key: str) -> Any:
    """Load and return the stored snapshot value for `key`.

    Raises
    ------
    FileNotFoundError
        When no snapshot exists for `key`.
    """
    sk = snapshot._key(key)
    storage_key = snapshot._key_of()(sk)
    backend = snapshot._backend
    if storage_key not in backend:
        raise FileNotFoundError(
            f"No snapshot file found for key {key!r} (storage key: {storage_key!r})"
        )
    return snapshot.recorder.loads(backend[storage_key])

`resolve_snapshot(snapshot, data, key)` ¶

Return the snapshot value for key, first writing data if the mode requires.

How the stored value is treated depends on snapshot.mode; see SnapshotMode. Whenever data is returned in place of a stored value, it is first passed through the recorder (dumps, then loads), so the caller's assertion sees what a later run would read back.

Raises:

Type	Description
`DuplicateSnapshotKeyError`	When the same `key` is used more than once within a test.

Source code in ditto/snapshot.py

def resolve_snapshot(snapshot: Snapshot, data: Any, key: str) -> Any:
    """Return the snapshot value for `key`, first writing `data` if the mode requires.

    How the stored value is treated depends on `snapshot.mode`; see `SnapshotMode`.
    Whenever `data` is returned in place of a stored value, it is first passed
    through the recorder (`dumps`, then `loads`), so the caller's assertion sees
    what a later run would read back.

    Raises
    ------
    DuplicateSnapshotKeyError
        When the same `key` is used more than once within a test.
    """
    sk = snapshot._key(key)
    key_of = snapshot._key_of()
    storage_key = key_of(sk)

    backend = snapshot._backend
    tracker = snapshot._tracker
    used_key = (id(backend), storage_key)
    if used_key in tracker.used_keys:
        raise DuplicateSnapshotKeyError(key)
    tracker.used_keys.add(used_key)
    tracker.register_access(backend, key_of, sk)

    recorder = snapshot.recorder
    exists = storage_key in backend

    # Build the lock observation up front (pure), but only record it AFTER the
    # backend access succeeds — recording before the write would leave a phantom
    # lock entry for a snapshot that failed to persist (see #84).
    seen = (
        LockSeen(
            target_id=snapshot.target_id,
            scheme=urlparse(snapshot.target).scheme,
            nodeid=snapshot.nodeid,
            key=key,
            recorder=snapshot.recorder_name,
        )
        if snapshot.target_id
        else None
    )

    match snapshot.mode, exists:
        case SnapshotMode.RECORD | SnapshotMode.VERIFY, True:
            value = recorder.loads(backend[storage_key])
        case SnapshotMode.VERIFY, False:
            # Never write to the backend: leave it untouched so the drift check
            # can detect the missing key.
            _, value = _round_trip(recorder, data)
        case _:
            raw, value = _round_trip(recorder, data)
            backend[storage_key] = raw
            (tracker.updated if exists else tracker.created).append(sk)

    # A missing key under VERIFY is recorded as "created" so the verify hook
    # reports it as unsynced.
    if seen is not None:
        tracker.record_lock_seen(seen, created=not exists)
    return value

ditto.snapshot¶

SnapshotKey(module, group_name, key, identifier) dataclass ¶

filename property ¶

display_name property ¶

__str__() ¶

LockSeen(target_id, scheme, nodeid, key, recorder) dataclass ¶

SnapshotMode ¶

Snapshot(group_name, module, target, _backend, recorder=_UNSET, recorder_name=_UNSET, mode=SnapshotMode.RECORD, nodeid='', target_id='', _tracker=_SessionTracker()) dataclass ¶

__call__(data, key) ¶

save_snapshot(snapshot, data, key) ¶

load_snapshot(snapshot, key) ¶

resolve_snapshot(snapshot, data, key) ¶

`SnapshotKey(module, group_name, key, identifier)` `dataclass` ¶

`filename` `property` ¶

`display_name` `property` ¶

`str()` ¶

`LockSeen(target_id, scheme, nodeid, key, recorder)` `dataclass` ¶

`SnapshotMode` ¶

`Snapshot(group_name, module, target, _backend, recorder=_UNSET, recorder_name=_UNSET, mode=SnapshotMode.RECORD, nodeid='', target_id='', _tracker=_SessionTracker())` `dataclass` ¶

`call(data, key)` ¶

`save_snapshot(snapshot, data, key)` ¶

`load_snapshot(snapshot, key)` ¶

`resolve_snapshot(snapshot, data, key)` ¶