【微服务测试教程】使用Python测试gRPC接口案例

微服务离不开gRPC

现在大多数企业都会采用微服务作为软件的架构，在这种架构的大背景下，微服务框架和协议广泛流行，而RPC也开始流行。 grpc 是基于RPC的框架，性能高，使用非常广泛。
grpc 由谷歌公司开发和维护，支持几乎所有的主流编程语言。 不管你用的是 Java， 还是 Python, 还是 Go, 还是 Ruby 等等，都可以使用他来实现远程的服务。

Protocol Buffers

grpc 默认使用 protocol buffers 作为序列化传输格式，通常会把传输的数据类型用一个带有.proto扩展名的普通文本文件来存储，不管是请求还是响应的数据都需要符合这里面定义的数据要求。
比如在进行用户验证时往往需要传入登录的用户信息，服务端返回 token 值，对应的 proto 文件描述， 如果请求或者响应数据太大，不能一次获取完，可以通过 stream 流信息持续传输，此时在类前面加 stream 关键字。

			  // login.proto
			  
			  service User{
			    // login
			    rpc Login (LoginReqeust ) returns (LoginReply ) {}
			    // stream
			    rpc GetImage(LoginRequest) return (stream LoginReply ){}
			  
			  }
			  
			  // 登录请求数据
			  message LoginReqeust {
			    string username = 1;
			    string passwd = 2;
			  }
			  
			  //登录响应数据
			  message LoginReply {
			    string token = 1;
			    string msg = 2;
			  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

proto 文件生成 gPRC 代码

以Python为例，首先需要安装 Python 相关的包。

		  pip install grpcio
		  pip install grpcio-tools
1
2

对应生成Python代码的命令， 注意 -I 表示 proto 文件所在的子目录：

		  python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I protos login.proto
1

输入完命令以后，在目录下会多两个文件，一个叫 login_pb2.py， 里面报告了请求和响应相关的数据格式，还有一个叫 login_pb2_grpc.py， 包含了客户端和响应端的类。
生成的 login_pb2.grpc.py 中的代码…
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-f6pu4PtL-1658989710154)(C:\Users\muji\Desktop\logseq_local_D__logseq笔记_markdown_1658920736\logseq笔记_markdown_1658920736\assets\image_1658915681955_0.png)]

创建gRPC服务

创建接口需要用到上面生成的 UserServicer 类。

		  import login_pb2_grpc
		  import login_pb2
		  
		  class User(login_pb2_grpc.UserServicer):
		  
		    def Login(self, request, context):
		      return login_pb2.LoginReply(msg=f"hello, {request.username}",
		                                 token="It is my token")
1
2
3
4
5
6
7
8

运行服务：

		  import grpc
		  from concurrent import futures
		  
		  def serve():
		    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
		    login_pb2_grpc.add_UserServicer_to_server(
		        User(), server)
		    server.add_insecure_port('[::]:50051')
		    server.start()
		    server.wait_for_termination()
		  
		  serve()
1
2
3
4
5
6
7
8
9
10
11
12

创建gRPC客户端

一个例子：

		  import grpc
		  
		  def run():
		      with grpc.insecure_channel('localhost:50051') as channel:
		          stub = login_pb2_grpc.UserStub(channel)
		          response = stub.Login(login_pb2.LoginRequest(username='you', passwd="123"))
		          print("Client received: " + response.msg)
1
2
3
4
5
6
7

protocol buffer 转成 json

客户端拿到响应结果之后，可以通过类属性获取字段：

			  response.msg
			  response.token
1
2

有时候我们会把它转换成 json 格式，方便获取整个响应。此时可以使用 protobuf 这个库，先安装：

		  pip install protobuf
1

接下来使用转换成 json 格式的方法就可以了。

		  from google.protobuf.json_format import MessageToJson
		  import json
		  
		  def run():
		      with grpc.insecure_channel('localhost:50051') as channel:
		          stub = login_pb2_grpc.UserStub(channel)
		          response = stub.Login(login_pb2.LoginRequest(username='you', passwd="123"))
		  		print(json.loads(MessageToJson(response)))
1
2
3
4
5
6
7
8