Surface replayed errors faithfully and key cassettes by a text method

docs/examples/advanced.md

@@ -188,6 +188,21 @@ with recorded_channel(
         assert e.code() == grpc.StatusCode.NOT_FOUND
 ```
 
+Streaming calls behave the same way. Messages received before the failure are
+delivered, and the error is raised when the stream reaches the point where it
+failed:
+
+```python
+with recorded_channel("stream_error.yaml", target) as channel:
+    stub = MyServiceStub(channel)
+    received = []
+    try:
+        for user in stub.ListUsers(ListUsersRequest(limit=-1)):
+            received.append(user)
+    except grpc.RpcError as e:
+        assert e.code() == grpc.StatusCode.INVALID_ARGUMENT
+```
+
 ## Parallel Test Isolation
 
 ```python

src/grpcvcr/interceptors/_base.py

@@ -86,6 +86,68 @@ def create_streaming_response(
     )
 
 
+def normalize_method(method: str | bytes) -> str:
+    """Return the method path as text.
+
+    grpc.aio hands the method over as bytes where the sync API uses str. Cassettes
+    key interactions by this value, so it is normalized on the way in to keep the
+    two channel kinds interchangeable and the recorded file readable.
+    """
+    return method.decode("utf-8") if isinstance(method, bytes) else method
+
+
+class _RecordedRpcError(grpc.RpcError, grpc.Call):  # type: ignore[misc]
+    """The error a recorded failure surfaces as on the sync path.
+
+    gRPC raises an object that is both the error and the call, so callers reach the
+    status through `code()` and `details()` on the exception itself. Its own version
+    of this, `_InactiveRpcError`, is private, hence the stand-in; the async path has
+    `aio.AioRpcError` and uses that instead.
+
+    Deliberately not exported: the two paths raise different types, so callers catch
+    `grpc.RpcError`, which covers both.
+    """
+
+    def __init__(
+        self,
+        code: grpc.StatusCode,
+        details: str | None,
+        trailing_metadata: tuple[tuple[str, str], ...] = (),
+    ) -> None:
+        super().__init__()
+        self._code = code
+        self._details = details
+        self._trailing_metadata = trailing_metadata
+
+    def __str__(self) -> str:
+        return f"{self._code.name}: {self._details}"
+
+    def code(self) -> grpc.StatusCode:
+        return self._code
+
+    def details(self) -> str | None:
+        return self._details
+
+    def trailing_metadata(self) -> tuple[tuple[str, str], ...]:
+        return self._trailing_metadata
+
+    def initial_metadata(self) -> tuple[tuple[str, str], ...]:
+        return ()
+
+    def is_active(self) -> bool:
+        return False
+
+    def time_remaining(self) -> float | None:
+        return None
+
+    def cancel(self) -> bool:
+        return False
+
+    def add_callback(self, callback: Callable[[Any], None]) -> bool:
+        callback(self)
+        return True
+
+
 def _dict_to_metadata(d: dict[str, list[str]]) -> tuple[tuple[str, str], ...]:
     """Convert metadata dict back to gRPC tuple format."""
     result: list[tuple[str, str]] = []
@@ -117,7 +179,7 @@ def __init__(
 
     def result(self, timeout: float | None = None) -> object:
         if self._code != grpc.StatusCode.OK:
-            raise grpc.RpcError()
+            raise self._error()
         return self._result
 
     def code(self) -> grpc.StatusCode:
@@ -146,9 +208,12 @@ def add_done_callback(self, fn: Callable[[Any], None]) -> None:
 
     def exception(self, timeout: float | None = None) -> Exception | None:
         if self._code != grpc.StatusCode.OK:
-            return grpc.RpcError()
+            return self._error()
         return None
 
+    def _error(self) -> _RecordedRpcError:
+        return _RecordedRpcError(self._code, self._details, self._trailing_metadata)
+
     def traceback(self, timeout: float | None = None) -> Any:
         return None
 
@@ -192,6 +257,8 @@ def __iter__(self) -> _FakeStreamingCall:
 
     def __next__(self) -> object:
         if self._index >= len(self._messages):
+            if self._code != grpc.StatusCode.OK:
+                raise _RecordedRpcError(self._code, self._details, self._trailing_metadata)
             raise StopIteration
         msg = self._messages[self._index]
         self._index += 1

src/grpcvcr/interceptors/aio.py

@@ -9,7 +9,7 @@
 from grpc import aio
 
 from grpcvcr.errors import RecordingDisabledError
-from grpcvcr.interceptors._base import RecordingInterceptorBase
+from grpcvcr.interceptors._base import RecordingInterceptorBase, normalize_method
 from grpcvcr.record_modes import RecordMode
 from grpcvcr.serialization import (
     Interaction,
@@ -158,7 +158,7 @@ async def intercept_unary_unary(
         client_call_details: aio.ClientCallDetails,
         request: Any,
     ) -> aio.Call:
-        method = client_call_details.method
+        method = normalize_method(client_call_details.method)
         request_bytes = request.SerializeToString()
         metadata = client_call_details.metadata
 
@@ -228,7 +228,7 @@ async def intercept_unary_stream(
         client_call_details: aio.ClientCallDetails,
         request: Any,
     ) -> aio.Call:
-        method = client_call_details.method
+        method = normalize_method(client_call_details.method)
         request_bytes = request.SerializeToString()
         metadata = client_call_details.metadata
 
@@ -312,7 +312,7 @@ async def intercept_stream_unary(
         client_call_details: aio.ClientCallDetails,
         request_iterator: Any,
     ) -> aio.Call:
-        method = client_call_details.method
+        method = normalize_method(client_call_details.method)
         metadata = client_call_details.metadata
 
         requests = [r async for r in request_iterator]
@@ -392,7 +392,7 @@ async def intercept_stream_stream(
         client_call_details: aio.ClientCallDetails,
         request_iterator: Any,
     ) -> aio.Call:
-        method = client_call_details.method
+        method = normalize_method(client_call_details.method)
         metadata = client_call_details.metadata
 
         requests = [r async for r in request_iterator]

tests/test_cassette_portability.py

@@ -0,0 +1,70 @@
+"""A cassette is a plain, portable artifact.
+
+The sync and async channels must record interchangeable cassettes, and the method
+must be written as readable text rather than a binary blob.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+import yaml
+
+from grpcvcr import AsyncRecordingChannel, Cassette, RecordingChannel, RecordMode
+
+METHOD = "/test.TestService/GetUser"
+
+
+def record_sync(path: Path, target: str, pb2, pb2_grpc) -> None:
+    with Cassette(path, record_mode=RecordMode.ALL) as cassette, RecordingChannel(cassette, target) as recording:
+        pb2_grpc.TestServiceStub(recording.channel).GetUser(pb2.GetUserRequest(id=3))
+
+
+async def record_async(path: Path, target: str, pb2, pb2_grpc) -> None:
+    with Cassette(path, record_mode=RecordMode.ALL) as cassette:
+        async with AsyncRecordingChannel(cassette, target) as recording:
+            await pb2_grpc.TestServiceStub(recording.channel).GetUser(pb2.GetUserRequest(id=3))
+
+
+class TestRecordedMethodIsText:
+    def test_sync(self, grpc_target: str, tmp_cassette_path: Path, pb2, pb2_grpc) -> None:
+        record_sync(tmp_cassette_path, grpc_target, pb2, pb2_grpc)
+
+        assert yaml.safe_load(tmp_cassette_path.read_text())["interactions"][0]["request"]["method"] == METHOD
+
+    @pytest.mark.asyncio
+    async def test_async(self, grpc_target: str, tmp_cassette_path: Path, pb2, pb2_grpc) -> None:
+        await record_async(tmp_cassette_path, grpc_target, pb2, pb2_grpc)
+
+        assert yaml.safe_load(tmp_cassette_path.read_text())["interactions"][0]["request"]["method"] == METHOD
+
+
+class TestCassettesReplayAcrossChannelKinds:
+    @pytest.mark.asyncio
+    async def test_sync_recorded_replays_on_async_channel(
+        self, grpc_target: str, tmp_cassette_path: Path, grpc_servicer, pb2, pb2_grpc
+    ) -> None:
+        record_sync(tmp_cassette_path, grpc_target, pb2, pb2_grpc)
+
+        grpc_servicer.call_count = 0
+        playback = Cassette(tmp_cassette_path, record_mode=RecordMode.NONE)
+        async with AsyncRecordingChannel(playback, grpc_target) as recording:
+            response = await pb2_grpc.TestServiceStub(recording.channel).GetUser(pb2.GetUserRequest(id=3))
+
+        assert response.user.id == 3
+        assert grpc_servicer.call_count == 0
+
+    @pytest.mark.asyncio
+    async def test_async_recorded_replays_on_sync_channel(
+        self, grpc_target: str, tmp_cassette_path: Path, grpc_servicer, pb2, pb2_grpc
+    ) -> None:
+        await record_async(tmp_cassette_path, grpc_target, pb2, pb2_grpc)
+
+        grpc_servicer.call_count = 0
+        playback = Cassette(tmp_cassette_path, record_mode=RecordMode.NONE)
+        with RecordingChannel(playback, grpc_target) as recording:
+            response = pb2_grpc.TestServiceStub(recording.channel).GetUser(pb2.GetUserRequest(id=3))
+
+        assert response.user.id == 3
+        assert grpc_servicer.call_count == 0

tests/test_interceptor_paths.py

@@ -47,12 +47,11 @@ def test_unary_error(self, grpc_target: str, tmp_cassette_path: Path, pb2, pb2_g
                 stub.GetUser(pb2.GetUserRequest(id=FAIL_ID))
 
     def test_server_streaming_error(self, grpc_target: str, tmp_cassette_path: Path, pb2, pb2_grpc) -> None:
-        # A streaming RPC that fails is recorded with its status code, but the
-        # error is not re-raised to the caller: the stream just ends early.
         cassette = Cassette(tmp_cassette_path, record_mode=RecordMode.NEW_EPISODES)
         with RecordingChannel(cassette, grpc_target) as recording:
             stub = pb2_grpc.TestServiceStub(recording.channel)
-            assert list(stub.ListUsers(pb2.ListUsersRequest(limit=FAIL_ID))) == []
+            with pytest.raises(grpc.RpcError):
+                list(stub.ListUsers(pb2.ListUsersRequest(limit=FAIL_ID)))
 
         assert cassette.interactions[0].response.code == "INVALID_ARGUMENT"
         assert cassette.interactions[0].response.details == "limit must be positive"
@@ -70,7 +69,8 @@ def test_bidi_streaming_error(self, grpc_target: str, tmp_cassette_path: Path, p
         cassette = Cassette(tmp_cassette_path, record_mode=RecordMode.NEW_EPISODES)
         with RecordingChannel(cassette, grpc_target) as recording:
             stub = pb2_grpc.TestServiceStub(recording.channel)
-            assert list(stub.Chat(chat_messages(pb2, FAIL_NAME))) == []
+            with pytest.raises(grpc.RpcError):
+                list(stub.Chat(chat_messages(pb2, FAIL_NAME)))
 
         assert cassette.interactions[0].response.code == "INVALID_ARGUMENT"