test_opentracing.py 文件源码

def testStreamUnaryOpenTracing(self):
        multi_callable = self._service.stream_unary_multi_callable
        requests = [b'\x01', b'\x02']
        self.assertRaises(grpc.RpcError, multi_callable, iter(requests))

        span0 = self._tracer.get_span(0)
        self.assertIsNotNone(span0)
        self.assertTrue(span0.get_tag('error'))

        span1 = self._tracer.get_span(1)
        self.assertIsNotNone(span1)
        self.assertTrue(span1.get_tag('error'))