How to test grpc status codes with unittests

[ozangungor12]

Hello,

I am implementing custom exception handling with gRPC status codes in which I send specific status codes for specific exceptions such as INVALID_ARGUMENT or FAILED_PRECONDITION instead of general UNKNOWN status.

I managed to send the status codes I want by using await stream.send_trailing_metadata().

However, when I try to create a similar scenario in unit tests, I don't see that specific status code but STATUS_UNKNOWN error appears.

Is there a correct way to extract the status code from the metadata itself? This is what I am doing in my test:

servicer = MyServicer()
    async with ChannelFor([servicer]) as channel:
        stub = MyStub(channel)
        request = MyRequest()
        response = await stub.MyService([request])

Thanks for the help!

[vmagamedov]

Just wrote a test to check your issue:

grpclib/tests/test_testing.py

Lines 19 to 29 in a290e41

	@pytest.mark.asyncio
	async def test_failure():
	class FailingService(DummyService):
	async def UnaryUnary(self, stream):
	raise GRPCError(Status.FAILED_PRECONDITION)
	async with ChannelFor([FailingService()]) as channel:
	stub = DummyServiceStub(channel)
	with pytest.raises(GRPCError) as err:
	await stub.UnaryUnary(DummyRequest(value='ping'))
	assert err.value.status is Status.FAILED_PRECONDITION

And seems that everything works as expected.

And the same is true for:

await stream.send_trailing_metadata(status=Status.FAILED_PRECONDITION)

[ozangungor12]

Thanks for the insights!