Bases: BasePlugin
gRPC interception plugin.
Patches grpc.insecure_channel and grpc.secure_channel at the module level.
Uses reference counting so nested sandboxes work correctly.
Each (call_type, method) pair has its own FIFO deque of GrpcMockConfig objects.
Source code in src/tripwire/plugins/grpc_plugin.py
| def __init__(self, verifier: StrictVerifier) -> None:
super().__init__(verifier)
self._queues: dict[str, deque[GrpcMockConfig]] = {}
self._registry_lock: threading.Lock = threading.Lock()
|
mock_unary_unary(method, *, returns, raises=None, required=True)
Register a mock for a unary-unary gRPC call.
Source code in src/tripwire/plugins/grpc_plugin.py
| def mock_unary_unary(
self,
method: str,
*,
returns: Any, # noqa: ANN401
raises: BaseException | None = None,
required: bool = True,
) -> None:
"""Register a mock for a unary-unary gRPC call."""
self._mock_call("unary_unary", method, returns=returns, raises=raises, required=required)
|
mock_unary_stream(method, *, returns, raises=None, required=True)
Register a mock for a unary-stream (server streaming) gRPC call.
Source code in src/tripwire/plugins/grpc_plugin.py
| def mock_unary_stream(
self,
method: str,
*,
returns: list[Any],
raises: BaseException | None = None,
required: bool = True,
) -> None:
"""Register a mock for a unary-stream (server streaming) gRPC call."""
self._mock_call("unary_stream", method, returns=returns, raises=raises, required=required)
|
mock_stream_unary(method, *, returns, raises=None, required=True)
Register a mock for a stream-unary (client streaming) gRPC call.
Source code in src/tripwire/plugins/grpc_plugin.py
| def mock_stream_unary(
self,
method: str,
*,
returns: Any, # noqa: ANN401
raises: BaseException | None = None,
required: bool = True,
) -> None:
"""Register a mock for a stream-unary (client streaming) gRPC call."""
self._mock_call("stream_unary", method, returns=returns, raises=raises, required=required)
|
mock_stream_stream(method, *, returns, raises=None, required=True)
Register a mock for a stream-stream (bidi streaming) gRPC call.
Source code in src/tripwire/plugins/grpc_plugin.py
| def mock_stream_stream(
self,
method: str,
*,
returns: list[Any],
raises: BaseException | None = None,
required: bool = True,
) -> None:
"""Register a mock for a stream-stream (bidi streaming) gRPC call."""
self._mock_call("stream_stream", method, returns=returns, raises=raises, required=required)
|
Install gRPC channel patches.
Source code in src/tripwire/plugins/grpc_plugin.py
| def install_patches(self) -> None:
"""Install gRPC channel patches."""
if not _GRPC_AVAILABLE:
raise ImportError(
"Install python-tripwire[grpc] to use GrpcPlugin: pip install python-tripwire[grpc]"
)
GrpcPlugin._original_insecure_channel = grpc_lib.insecure_channel
GrpcPlugin._original_secure_channel = grpc_lib.secure_channel
grpc_lib.insecure_channel = _patched_insecure_channel
grpc_lib.secure_channel = _patched_secure_channel
|
Restore original gRPC channel functions.
Source code in src/tripwire/plugins/grpc_plugin.py
| def restore_patches(self) -> None:
"""Restore original gRPC channel functions."""
if GrpcPlugin._original_insecure_channel is not None:
grpc_lib.insecure_channel = GrpcPlugin._original_insecure_channel
GrpcPlugin._original_insecure_channel = None
if GrpcPlugin._original_secure_channel is not None:
grpc_lib.secure_channel = GrpcPlugin._original_secure_channel
GrpcPlugin._original_secure_channel = None
|
matches(interaction, expected)
Field-by-field comparison with dirty-equals support.
Source code in src/tripwire/plugins/grpc_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
|
Return all GrpcMockConfig with required=True still in any queue.
Source code in src/tripwire/plugins/grpc_plugin.py
| def get_unused_mocks(self) -> list[GrpcMockConfig]:
"""Return all GrpcMockConfig with required=True still in any queue."""
unused: list[GrpcMockConfig] = []
with self._registry_lock:
for queue in self._queues.values():
for config in queue:
if config.required:
unused.append(config)
return unused
|
assert_unary_unary(method, request, metadata=None, raised=_ABSENT)
Typed helper: assert the next unary_unary interaction.
Source code in src/tripwire/plugins/grpc_plugin.py
| def assert_unary_unary(
self,
method: str,
request: Any, # noqa: ANN401
metadata: Any = None, # noqa: ANN401
raised: Any = _ABSENT, # noqa: ANN401
) -> None:
"""Typed helper: assert the next unary_unary interaction."""
self._assert_call("unary_unary", method, request, metadata, raised)
|
assert_unary_stream(method, request, metadata=None, raised=_ABSENT)
Typed helper: assert the next unary_stream interaction.
Source code in src/tripwire/plugins/grpc_plugin.py
| def assert_unary_stream(
self,
method: str,
request: Any, # noqa: ANN401
metadata: Any = None, # noqa: ANN401
raised: Any = _ABSENT, # noqa: ANN401
) -> None:
"""Typed helper: assert the next unary_stream interaction."""
self._assert_call("unary_stream", method, request, metadata, raised)
|
assert_stream_unary(method, request, metadata=None, raised=_ABSENT)
Typed helper: assert the next stream_unary interaction.
Source code in src/tripwire/plugins/grpc_plugin.py
| def assert_stream_unary(
self,
method: str,
request: Any, # noqa: ANN401
metadata: Any = None, # noqa: ANN401
raised: Any = _ABSENT, # noqa: ANN401
) -> None:
"""Typed helper: assert the next stream_unary interaction."""
self._assert_call("stream_unary", method, request, metadata, raised)
|
assert_stream_stream(method, request, metadata=None, raised=_ABSENT)
Typed helper: assert the next stream_stream interaction.
Source code in src/tripwire/plugins/grpc_plugin.py
| def assert_stream_stream(
self,
method: str,
request: Any, # noqa: ANN401
metadata: Any = None, # noqa: ANN401
raised: Any = _ABSENT, # noqa: ANN401
) -> None:
"""Typed helper: assert the next stream_stream interaction."""
self._assert_call("stream_stream", method, request, metadata, raised)
|