Bases: BasePlugin
Redis interception plugin.
Patches redis.Redis.execute_command at the class level.
Uses reference counting so nested sandboxes work correctly.
Each command name (uppercase) has its own FIFO deque of RedisMockConfig
objects. Calls are stateless -- there are no state transitions.
Source code in src/bigfoot/plugins/redis_plugin.py
| def __init__(self, verifier: StrictVerifier) -> None:
super().__init__(verifier)
self._queues: dict[str, deque[RedisMockConfig]] = {}
self._registry_lock: threading.Lock = threading.Lock()
|
mock_command(command, *, returns, raises=None, required=True)
Register a mock for a single Redis command invocation.
Args:
command: The Redis command name (case-insensitive).
returns: Value to return when this mock is consumed.
raises: If provided, this exception is raised instead of returning.
required: If False, the mock is not reported as unused at teardown.
Source code in src/bigfoot/plugins/redis_plugin.py
| def mock_command(
self,
command: str,
*,
returns: Any, # noqa: ANN401
raises: BaseException | None = None,
required: bool = True,
) -> None:
"""Register a mock for a single Redis command invocation.
Args:
command: The Redis command name (case-insensitive).
returns: Value to return when this mock is consumed.
raises: If provided, this exception is raised instead of returning.
required: If False, the mock is not reported as unused at teardown.
"""
cmd_upper = command.upper()
config = RedisMockConfig(
command=cmd_upper,
returns=returns,
raises=raises,
required=required,
)
with self._registry_lock:
if cmd_upper not in self._queues:
self._queues[cmd_upper] = deque()
self._queues[cmd_upper].append(config)
|
Reference-counted class-level patch installation.
Source code in src/bigfoot/plugins/redis_plugin.py
| def activate(self) -> None:
"""Reference-counted class-level patch installation."""
if not _REDIS_AVAILABLE:
raise ImportError(
"Install bigfoot[redis] to use RedisPlugin: pip install bigfoot[redis]"
)
with RedisPlugin._install_lock:
if RedisPlugin._install_count == 0:
RedisPlugin._original_execute_command = redis_lib.Redis.execute_command
redis_lib.Redis.execute_command = _patched_execute_command # type: ignore[assignment]
RedisPlugin._install_count += 1
|
matches(interaction, expected)
Field-by-field comparison with dirty-equals support.
Source code in src/bigfoot/plugins/redis_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)
All three fields (command, args, kwargs) are required in assert_interaction().
Source code in src/bigfoot/plugins/redis_plugin.py
| def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
"""All three fields (command, args, kwargs) are required in assert_interaction()."""
return frozenset({"command", "args", "kwargs"})
|
Return all RedisMockConfig with required=True still in any queue.
Source code in src/bigfoot/plugins/redis_plugin.py
| def get_unused_mocks(self) -> list[RedisMockConfig]:
"""Return all RedisMockConfig with required=True still in any queue."""
unused: list[RedisMockConfig] = []
with self._registry_lock:
for queue in self._queues.values():
for config in queue:
if config.required:
unused.append(config)
return unused
|
assert_command(command, args=(), kwargs=None)
Typed helper: assert the next Redis command interaction.
Wraps assert_interaction() for ergonomic use. All three fields
(command, args, kwargs) are required.
Source code in src/bigfoot/plugins/redis_plugin.py
| def assert_command(
self,
command: str,
args: tuple[Any, ...] = (),
kwargs: dict[str, Any] | None = None,
) -> None:
"""Typed helper: assert the next Redis command interaction.
Wraps assert_interaction() for ergonomic use. All three fields
(command, args, kwargs) are required.
"""
from bigfoot._context import _get_test_verifier_or_raise # noqa: PLC0415
kw = kwargs if kwargs is not None else {}
cmd_upper = command.upper()
source_id = f"redis:{cmd_upper.lower()}"
sentinel = _RedisSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
command=cmd_upper,
args=args,
kwargs=kw,
)
|