Bases: StateMachinePlugin
Database interception plugin.
Patches sqlite3.connect at module level.
Uses reference counting so nested sandboxes work correctly.
States: connected -> in_transaction -> connected/closed
Source code in src/bigfoot/plugins/database_plugin.py
| def __init__(self, verifier: "StrictVerifier") -> None:
super().__init__(verifier)
self._connect_sentinel = _StepSentinel(_SOURCE_CONNECT)
self._execute_sentinel = _StepSentinel(_SOURCE_EXECUTE)
self._commit_sentinel = _StepSentinel(_SOURCE_COMMIT)
self._rollback_sentinel = _StepSentinel(_SOURCE_ROLLBACK)
self._close_sentinel = _StepSentinel(_SOURCE_CLOSE)
|
Reference-counted module-level patch installation.
Source code in src/bigfoot/plugins/database_plugin.py
| def activate(self) -> None:
"""Reference-counted module-level patch installation."""
with DatabasePlugin._install_lock:
if DatabasePlugin._install_count == 0:
self._install_patches()
DatabasePlugin._install_count += 1
|
matches(interaction, expected)
Field-by-field comparison with dirty-equals support.
Source code in src/bigfoot/plugins/database_plugin.py
| def matches(self, interaction: Interaction, expected: dict[str, Any]) -> bool:
"""Field-by-field comparison with dirty-equals support."""
try:
for key, expected_val in expected.items():
actual_val = interaction.details.get(key)
if expected_val != actual_val:
return False
return True
except Exception:
return False
|
assertable_fields(interaction)
Return assertable fields for each step type.
Source code in src/bigfoot/plugins/database_plugin.py
| def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
"""Return assertable fields for each step type."""
if interaction.source_id == _SOURCE_CONNECT:
return frozenset({"database"})
if interaction.source_id == _SOURCE_EXECUTE:
return frozenset({"sql", "parameters"})
if interaction.source_id in (_SOURCE_COMMIT, _SOURCE_ROLLBACK, _SOURCE_CLOSE):
return frozenset()
return frozenset(interaction.details.keys())
|
assert_connect(*, database)
Assert the next database connect interaction.
Source code in src/bigfoot/plugins/database_plugin.py
| def assert_connect(self, *, database: str) -> None:
"""Assert the next database connect interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._connect_sentinel, database=database
)
|
assert_execute(*, sql, parameters)
Assert the next database execute interaction.
Source code in src/bigfoot/plugins/database_plugin.py
| def assert_execute(self, *, sql: str, parameters: object) -> None:
"""Assert the next database execute interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._execute_sentinel, sql=sql, parameters=parameters
)
|
Assert the next database commit interaction.
Source code in src/bigfoot/plugins/database_plugin.py
| def assert_commit(self) -> None:
"""Assert the next database commit interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(self._commit_sentinel)
|
Assert the next database rollback interaction.
Source code in src/bigfoot/plugins/database_plugin.py
| def assert_rollback(self) -> None:
"""Assert the next database rollback interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(self._rollback_sentinel)
|
Assert the next database close interaction.
Source code in src/bigfoot/plugins/database_plugin.py
| def assert_close(self) -> None:
"""Assert the next database close interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(self._close_sentinel)
|