Bases: StateMachinePlugin
asyncpg interception plugin.
Patches asyncpg.connect at module level.
Uses reference counting so nested sandboxes work correctly.
States: disconnected -> connected -> closed
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def __init__(self, verifier: "StrictVerifier") -> None:
super().__init__(verifier)
self._connect_sentinel = _StepSentinel(_SOURCE_CONNECT)
self._execute_sentinel = _StepSentinel(_SOURCE_EXECUTE)
self._fetch_sentinel = _StepSentinel(_SOURCE_FETCH)
self._fetchrow_sentinel = _StepSentinel(_SOURCE_FETCHROW)
self._fetchval_sentinel = _StepSentinel(_SOURCE_FETCHVAL)
self._close_sentinel = _StepSentinel(_SOURCE_CLOSE)
|
Reference-counted module-level patch installation.
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def activate(self) -> None:
"""Reference-counted module-level patch installation."""
with AsyncpgPlugin._install_lock:
if AsyncpgPlugin._install_count == 0:
self._install_patches()
AsyncpgPlugin._install_count += 1
|
matches(interaction, expected)
Field-by-field comparison with dirty-equals support.
Source code in src/bigfoot/plugins/asyncpg_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/asyncpg_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({"query", "args"})
if interaction.source_id == _SOURCE_FETCH:
return frozenset({"query", "args"})
if interaction.source_id == _SOURCE_FETCHROW:
return frozenset({"query", "args"})
if interaction.source_id == _SOURCE_FETCHVAL:
return frozenset({"query", "args"})
if interaction.source_id == _SOURCE_CLOSE:
return frozenset()
return frozenset(interaction.details.keys())
|
Assert the next asyncpg connect interaction.
Pass whichever connection fields were used: dsn, host, port, database, user.
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def assert_connect(self, **kwargs: object) -> None:
"""Assert the next asyncpg connect interaction.
Pass whichever connection fields were used: dsn, host, port, database, 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(*, query, args)
Assert the next asyncpg execute interaction.
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def assert_execute(self, *, query: str, args: object) -> None:
"""Assert the next asyncpg execute interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._execute_sentinel, query=query, args=args
)
|
assert_fetch(*, query, args)
Assert the next asyncpg fetch interaction.
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def assert_fetch(self, *, query: str, args: object) -> None:
"""Assert the next asyncpg fetch interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._fetch_sentinel, query=query, args=args
)
|
assert_fetchrow(*, query, args)
Assert the next asyncpg fetchrow interaction.
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def assert_fetchrow(self, *, query: str, args: object) -> None:
"""Assert the next asyncpg fetchrow interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._fetchrow_sentinel, query=query, args=args
)
|
assert_fetchval(*, query, args)
Assert the next asyncpg fetchval interaction.
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def assert_fetchval(self, *, query: str, args: object) -> None:
"""Assert the next asyncpg fetchval interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(
self._fetchval_sentinel, query=query, args=args
)
|
Assert the next asyncpg close interaction.
Source code in src/bigfoot/plugins/asyncpg_plugin.py
| def assert_close(self) -> None:
"""Assert the next asyncpg close interaction."""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
_get_test_verifier_or_raise().assert_interaction(self._close_sentinel)
|