Bases: BasePlugin
pymemcache interception plugin.
Patches pymemcache.client.base.Client methods at the class level.
Uses reference counting so nested sandboxes work correctly.
Source code in src/tripwire/plugins/memcache_plugin.py
| def __init__(self, verifier: StrictVerifier) -> None:
super().__init__(verifier)
self._queues: dict[str, deque[MemcacheMockConfig]] = {}
self._registry_lock: threading.Lock = threading.Lock()
|
mock_command(command, *, returns, raises=None, required=True)
Register a mock for a single memcache command invocation.
Source code in src/tripwire/plugins/memcache_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 memcache command invocation."""
cmd_upper = command.upper()
config = MemcacheMockConfig(
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)
|
Install pymemcache Client method patches.
Source code in src/tripwire/plugins/memcache_plugin.py
| def install_patches(self) -> None:
"""Install pymemcache Client method patches."""
if not _PYMEMCACHE_AVAILABLE:
raise ImportError(
"Install python-tripwire[pymemcache] to use MemcachePlugin: "
"pip install python-tripwire[pymemcache]"
)
from pymemcache.client.base import Client
# Patch __init__ to capture connection metadata
if MemcachePlugin._original_init is None:
MemcachePlugin._original_init = Client.__init__
def _patched_init(self_: object, server: Any, *args: Any, **kwargs: Any) -> None: # noqa: ANN401
assert MemcachePlugin._original_init is not None
MemcachePlugin._original_init(self_, server, *args, **kwargs)
if isinstance(server, tuple):
host, port = str(server[0]), int(server[1]) if len(server) > 1 else 11211
else:
host, port = str(server), 11211
_memcache_conn_meta[self_] = (normalize_host(host), port)
Client.__init__ = _patched_init
for method_name in _ALL_INTERCEPTED:
MemcachePlugin._originals[method_name] = getattr(Client, method_name, None)
setattr(Client, method_name, _make_patched_method(method_name))
|
Restore original pymemcache Client methods.
Source code in src/tripwire/plugins/memcache_plugin.py
| def restore_patches(self) -> None:
"""Restore original pymemcache Client methods."""
from pymemcache.client.base import Client
for method_name, original in MemcachePlugin._originals.items():
if original is not None:
setattr(Client, method_name, original)
MemcachePlugin._originals = {k: None for k in MemcachePlugin._originals}
if MemcachePlugin._original_init is not None:
Client.__init__ = MemcachePlugin._original_init
MemcachePlugin._original_init = None
|
Typed helper: assert the next memcache GET interaction.
Source code in src/tripwire/plugins/memcache_plugin.py
| def assert_get(self, command: str, key: str) -> None:
"""Typed helper: assert the next memcache GET interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = f"memcache:{command.lower()}"
sentinel = _MemcacheSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
command=command,
key=key,
)
|
assert_set(command, key, value, expire=0)
Typed helper: assert the next memcache SET/ADD/REPLACE interaction.
Source code in src/tripwire/plugins/memcache_plugin.py
| def assert_set(
self,
command: str,
key: str,
value: Any, # noqa: ANN401
expire: int = 0,
) -> None:
"""Typed helper: assert the next memcache SET/ADD/REPLACE interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = f"memcache:{command.lower()}"
sentinel = _MemcacheSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
command=command,
key=key,
value=value,
expire=expire,
)
|
assert_delete(command, key)
Typed helper: assert the next memcache DELETE interaction.
Source code in src/tripwire/plugins/memcache_plugin.py
| def assert_delete(self, command: str, key: str) -> None:
"""Typed helper: assert the next memcache DELETE interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = f"memcache:{command.lower()}"
sentinel = _MemcacheSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
command=command,
key=key,
)
|
assert_incr(command, key, value=1)
Typed helper: assert the next memcache INCR/DECR interaction.
Source code in src/tripwire/plugins/memcache_plugin.py
| def assert_incr(self, command: str, key: str, value: int = 1) -> None:
"""Typed helper: assert the next memcache INCR/DECR interaction."""
from tripwire._context import _get_test_verifier_or_raise # noqa: PLC0415
source_id = f"memcache:{command.lower()}"
sentinel = _MemcacheSentinel(source_id)
_get_test_verifier_or_raise().assert_interaction(
sentinel,
command=command,
key=key,
value=value,
)
|