Cassette

The Cassette class is the core container for recording and replaying gRPC interactions.

Overview

A cassette stores a collection of recorded gRPC interactions (request/response pairs) and provides methods to find matching interactions during playback.

from grpcvcr import Cassette, RecordMode, RecordingChannel

# Create a cassette for recording
cassette = Cassette("tests/cassettes/my_test.yaml", record_mode=RecordMode.NEW_EPISODES)

# Use with a RecordingChannel
with RecordingChannel(cassette, "localhost:50051") as recording:
    stub = MyServiceStub(recording.channel)
    response = stub.GetUser(GetUserRequest(id=1))

API Reference

A collection of recorded gRPC interactions.

Cassettes store request/response pairs and can be saved to disk for later playback during tests. Use as a context manager to ensure changes are saved.

Example

from grpcvcr import Cassette, RecordMode, RecordingChannel

# Record interactions
with Cassette("tests/cassettes/my_test.yaml", record_mode=RecordMode.ALL) as cassette:
    with RecordingChannel(cassette, "localhost:50051") as recording:
        stub = MyServiceStub(recording.channel)
        response = stub.GetUser(GetUserRequest(id=1))

# Playback interactions
with Cassette("tests/cassettes/my_test.yaml", record_mode=RecordMode.NONE) as cassette:
    with RecordingChannel(cassette, "localhost:50051") as recording:
        stub = MyServiceStub(recording.channel)
        response = stub.GetUser(GetUserRequest(id=1))  # Returns recorded response

interactions property

interactions: list[Interaction]

All recorded interactions in this cassette.

record_mode class-attribute instance-attribute

record_mode: RecordMode = NEW_EPISODES

How to handle recording vs playback.

can_record property

can_record: bool

Whether recording is allowed in the current mode.

save

save() -> None

Save cassette to disk if it has been modified.

Called automatically when using the cassette as a context manager.

Source code in src/grpcvcr/cassette.py

def save(self) -> None:
    """Save cassette to disk if it has been modified.

    Called automatically when using the cassette as a context manager.
    """
    if self._dirty:
        CassetteSerializer.save(self.path, self._data)
        self._dirty = False

find_interaction

find_interaction(request: InteractionRequest) -> Interaction | None

Find a matching recorded interaction for a request.

Parameters:

Name	Type	Description	Default
request	InteractionRequest	The request to match.	required

Returns:

Type	Description
Interaction | None	The matching Interaction, or None if not found.

Source code in src/grpcvcr/cassette.py

def find_interaction(self, request: InteractionRequest) -> Interaction | None:
    """Find a matching recorded interaction for a request.

    Args:
        request: The request to match.

    Returns:
        The matching Interaction, or None if not found.
    """
    return find_matching_interaction(request, self.interactions, self.match_on)

record_interaction

record_interaction(interaction: Interaction) -> None

Record a new interaction to the cassette.

In RecordMode.ALL, existing interactions with the same request (based on the matcher) are replaced. In other modes, new interactions are appended.

Parameters:

Name	Type	Description	Default
interaction	Interaction	The interaction to record.	required

Source code in src/grpcvcr/cassette.py

def record_interaction(self, interaction: Interaction) -> None:
    """Record a new interaction to the cassette.

    In `RecordMode.ALL`, existing interactions with the same request
    (based on the matcher) are replaced. In other modes, new
    interactions are appended.

    Args:
        interaction: The interaction to record.
    """
    with self._lock:
        if self.record_mode == RecordMode.ALL:
            self._data.interactions = [
                i for i in self._data.interactions if not self.match_on.matches(interaction.request, i.request)
            ]

        self._data.interactions.append(interaction)
        self._dirty = True

Context manager for using a cassette.

This is a convenience wrapper around Cassette that ensures the cassette is saved when the context exits.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the cassette file.	required
record_mode	RecordMode	How to handle recording vs playback.	NEW_EPISODES
match_on	Matcher | None	Matcher to use for finding recorded interactions.	None

Yields:

Type	Description
Cassette	The Cassette instance.

Example

from grpcvcr import use_cassette, RecordMode, RecordingChannel

with use_cassette("tests/cassettes/test.yaml", record_mode=RecordMode.ALL) as cassette:
    with RecordingChannel(cassette, "localhost:50051") as recording:
        stub = MyServiceStub(recording.channel)
        response = stub.MyMethod(request)

Source code in src/grpcvcr/cassette.py

@contextmanager
def use_cassette(
    path: str | Path,
    record_mode: RecordMode = RecordMode.NEW_EPISODES,
    match_on: Matcher | None = None,
) -> Generator[Cassette, None, None]:
    """Context manager for using a cassette.

    This is a convenience wrapper around `Cassette` that ensures the
    cassette is saved when the context exits.

    Args:
        path: Path to the cassette file.
        record_mode: How to handle recording vs playback.
        match_on: Matcher to use for finding recorded interactions.

    Yields:
        The Cassette instance.

    Example:
        ```python
        from grpcvcr import use_cassette, RecordMode, RecordingChannel

        with use_cassette("tests/cassettes/test.yaml", record_mode=RecordMode.ALL) as cassette:
            with RecordingChannel(cassette, "localhost:50051") as recording:
                stub = MyServiceStub(recording.channel)
                response = stub.MyMethod(request)
        ```
    """
    cassette = Cassette(
        path=Path(path),
        record_mode=record_mode,
        match_on=match_on or DEFAULT_MATCHER,
    )
    try:
        yield cassette
    finally:
        cassette.save()