Bases: StateMachinePlugin
Psycopg2 interception plugin.
Patches psycopg2.connect at module level.
Uses reference counting so nested sandboxes work correctly.
States: disconnected -> connected -> in_transaction -> connected/closed
Source code in src/bigfoot/plugins/psycopg2_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/psycopg2_plugin.py
| def activate(self) -> None:
"""Reference-counted module-level patch installation."""
with Psycopg2Plugin._install_lock:
if Psycopg2Plugin._install_count == 0:
self._install_patches()
Psycopg2Plugin._install_count += 1
|
matches(interaction, expected)
Field-by-field comparison with dirty-equals support.
Source code in src/bigfoot/plugins/psycopg2_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/psycopg2_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(interaction.details.keys())
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 the next psycopg2 connect interaction.
Pass whichever connection fields were used: dsn, host, port, dbname, user.
Source code in src/bigfoot/plugins/psycopg2_plugin.py
| def assert_connect(self, **kwargs: object) -> None:
"""Assert the next psycopg2 connect interaction.
Pass whichever connection fields were used: dsn, host, port, dbname, user.
"""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._connect_sentinel, **kwargs
)
|
assert_execute(*, sql, parameters)
Assert the next psycopg2 execute interaction.
Source code in src/bigfoot/plugins/psycopg2_plugin.py
| def assert_execute(self, *, sql: str, parameters: object) -> None:
"""Assert the next psycopg2 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 psycopg2 commit interaction.
Source code in src/bigfoot/plugins/psycopg2_plugin.py
| def assert_commit(self) -> None:
"""Assert the next psycopg2 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 psycopg2 rollback interaction.
Source code in src/bigfoot/plugins/psycopg2_plugin.py
| def assert_rollback(self) -> None:
"""Assert the next psycopg2 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 psycopg2 close interaction.
Source code in src/bigfoot/plugins/psycopg2_plugin.py
| def assert_close(self) -> None:
"""Assert the next psycopg2 close interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(self._close_sentinel)
|