test_opentracing.py 文件源码

def testUnaryStreamOpenTracing(self):
        multi_callable = self._service.unary_stream_multi_callable
        request = b'\x01'
        response = multi_callable(request)
        self.assertRaises(grpc.RpcError, list, response)

        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'))