aeat.adapters.persistence.storage.envelope._repository_test_suite module

Shared contract test suite for SecureBoundRepository subclasses.

The 8 concrete secure-storage repositories (filing drafts, submissions, filing history, complementaria, justificantes, observations, assets, inventory and friends) each ship a near-identical block of pytest functions that exercise the same anti-tautology / anti-regression properties of the encrypted persistence boundary:

• test_round_trip_preserves_payload - save -> load equality across the Envelope boundary.

• test_save_is_idempotent - repeated save does not duplicate rows.

• test_load_returns_none_when_absent.

• test_delete_removes / test_delete_removes_object.

• test_delete_missing_returns_false.

• test_object_marker_identifies_secure_backend - logical store_dir pointer carries the namespace marker so diagnostic CLI emits stay honest about the backend.

• test_unsafe_id_rejected - the path-safety gate refuses filesystem-traversal identifiers.

• test_database_payload_is_encrypted_audit_data - the raw on-disk SQLite file does not surface plaintext that the payload contained.

• test_foreign_class_object_refused - a row written at the wrong SensitivityClass raises ClassificationError on load.

• test_boundary_catches_simulated_field_drop_via_corrupted_payload - the strict-equality witness used by every roundtrip is honest: mutating the on-disk JSON envelope to drop a typed field surfaces either as ValidationError or strict inequality.

This module captures the 11 canonical checks behind one assert_secure_repository_contract() function. The contract runs each check against a real SQLite-backed SecureObjectRepository wired through the SecureBoundRepository under test. Mocks are forbidden by the roundtrip discipline; the suite uses EphemeralMasterKeyProvider and a real create_engine_from_settings() engine.

The contract function returns the number of checks executed so the caller’s self-test can assert the expected count and prove no check was silently skipped.

class SecureRepositoryContractCase(repository_factory, first_payload, second_payload, plaintext_witnesses, mutation_field)

Bases: Generic

Inputs required to run the contract against a concrete repository.

The contract is parameterised by:

• repository_factory - zero-arg callable returning a fresh SecureBoundRepository[T] instance. Invoked repeatedly so cross-instance roundtrips are exercised. The factory MUST route its underlying SecureObjectRepository at the currently-active process-default engine (i.e. construct Concrete() rather than passing an explicit engine), because the contract harness rebinds AEAT_DATABASE_URL to a fresh SQLite file per check and disposes cached engines between checks.

• first_payload / second_payload - two distinct, fully-populated payload instances whose extracted identifiers differ. Both must use non-default values for every optional field so the strict-equality witness has signal.

• plaintext_witnesses - a tuple of byte strings that the contract asserts MUST NOT appear in the raw on-disk SQLite file after the first payload is persisted. Callers supply tax identifiers, monetary amounts, or other plaintext that the payload encodes; the contract simply checks absence.

• mutation_field - the JSON key inside payload (the inner object that the envelope wraps) whose deletion should produce either a ValidationError on reload or a strictly unequal payload. The field MUST be required by the payload schema or carry a non-default value in first_payload so that the negative case is observable.

Parameters:

• repository_factory (Callable[[], SecureBoundRepository])

• first_payload (T)

• second_payload (T)

• plaintext_witnesses (tuple[bytes, ...])

• mutation_field (str)

repository_factory: Callable[[], SecureBoundRepository[TypeVar(T, bound= BaseModel)]]

first_payload: TypeVar(T, bound= BaseModel)

second_payload: TypeVar(T, bound= BaseModel)

plaintext_witnesses: tuple[bytes, ...]

mutation_field: str

assert_secure_repository_contract(case, *, tmp_path)

Run all 11 canonical contract checks against case.

For each check the function:

1. Disposes any cached process-default engine.

2. Rebinds AEAT_DATABASE_URL and AEAT_LOCAL_STORAGE_ROOT through override_settings to a fresh SQLite file under tmp_path.

3. Builds a real engine for that URL and materialises the ORM schema.

4. Activates a real EphemeralMasterKeyProvider.

5. Invokes the check, which constructs the repository via case.repository_factory; the repository’s internal SecureObjectRepository lookup resolves to the engine activated above.

6. Disposes the engine.

Returns the number of checks executed. Callers SHOULD assert this count equals EXPECTED_CHECK_COUNT so a silently-skipped check is visible at the self-test boundary.

Return type:

int

Parameters:

• case (SecureRepositoryContractCase)

• tmp_path (Path)

EXPECTED_CHECK_COUNT: int

The number of canonical contract checks executed by assert_secure_repository_contract().

The 11 canonical anti-tautology tests reduce to 10 distinct check callables (test_delete_removes and test_delete_removes_object share an implementation; the contract honours both names but the underlying behaviour runs once per invocation). Plus the raw-bytes inspection (test_database_payload_is_encrypted_audit_data) and the field-drop simulation (test_boundary_catches_simulated_field_drop_via_corrupted_payload) that need the database path / engine handle directly.

Total: len(_PARAM_CHECKS) + 2.