服务端测试实战之RPC协议（三）
在服务端测试实战之RPC协议(二)中详细了介绍了gRPC协议的通信，以及针对Python编程语言怎么使用来进行测试。首先需要明确的是在服务端的测试中，不管协议，保持的一个原则是客户端与服务端的交互，都是根据不同协议编写不同的客户端模拟与服务端的交互，最后拿到服务端的数据来验证结果的准确性，那么简单的总结就是第一步先确认协议，第二步就是使用对应的编程语言编写协议的客户端模拟与服务端的交互，最后是验证。
针对gRPC的协议中，它的交互主要是单向，应答流，请求流，和双向流，应答流简单的理解就是一次请求，服务端N次返回结果，也就是说通过循环的方式拿到服务端的数据，那么请求流可以理解为N次请求，一次返回结果，请求流需要使迭代器的方式，也就是yield的方式来进行发送请求，双向流就很好理解了，N次请求，N次回应，请求方是迭代器的方式，拿到服务端就是循环的方式。其实在gRPC协议中，针对这种流式的请求在编程模式中更多实用的是异步编程，同步编程方式很难复合流式的诉求，比如N次请求N次回应，这中间本身就是一个持续的过程，而同步交互更多简单粗暴的就是请求了得尽快拿到回应数据，而异步就是在N次发送请求中，不断的发送，服务端然后逐步的返回来结果信息。既然是异步，也就是离不开协程，而协程本身在计算机的层面是不存在的，计算机里面更多的是进程和线程，而协程更多是用户行为控制的一种线程，它的优势是遇到IO就切换，这很符合异步的交互。不管是同步交互还是同步通信，它最大的缺陷就是当有一个请求存在占用太多的计算能力以及逻辑存在问题，这就导致队列堵塞，而造成消息（任务）的积压，从而形成TimeOut的超时情况。
下面通过具体的案例来说明gRPC的协议中四种模式，如helloworld.proto的文件为：
syntax = "proto3";
service Greeter {
rpc SayHello(HelloRequest) returns (HelloReply) {}
rpc Register (stream RegisterRequest) returns (RegisterReply) {}
rpc Login (LoginRequest) returns (stream LoginReply) {}
rpc Profile (stream ProfileRequest) returns (stream ProfileReply) {}
}
message HelloRequest {
string name = 1;
}
message HelloReply {
string message = 1;
}
message RegisterRequest {
string username = 1;
string password = 2;
}
message RegisterReply{
string token =1;
}
message LoginRequest{
string username = 1;
string password = 2;
}
message LoginReply{
string nick = 1;
string username = 2;
string address = 3;
}
message ProfileRequest{
string nick = 1;
}
message ProfileReply{
string username = 1;
string phone = 2;
}在上面的proto文件中，可以看到Register的接口是请求流，Login是应答流，而Profile是双向流，流在proto的关键字是stream，针对流的这种建议使用异步编程的方式来进行。执行如下命令生成pb2和pb2_grpc的文件，如下所示:
下面通过具体的案例来具体说下针对不同的方式编写测试代码，单向在之前的文章体系里面有介绍，也是最简单的一种，这里就不再重复，首先来看Register，也就是请求流，也就是N次请求，一次返回结果信息，那么请求的数据我们需要写成可迭代的对象，这地方也就使用到生成器和迭代器了。其实保持一个原则，就是可迭代的函数都是可以循环输出的，如下函数：
def func():
data=[x for x in range(3)]
for item in data:
yield item首先它是一个生成器的函数，然后针对生成器的函数对象依次进行next()输出，具体如下所示：
到最后一个对象不存在的时候就会报StopIteration的错误信息，这是正常的，生成器结合迭代器是非常强大的，比如举例，使用一个一百万的列表对象进行循环，获取到里面的值进行处理，如果使用常规的方式计算机的资源首先存在问题，但是使用生成器和迭代器，需要那个就获取那个，很轻松的解决了这个问题，其实针对生成器的函数，我个人理解最简单的验证方式就是它是可以被循环的，直接编写for循环来验证，如下所示:
def func():
data=[x for x in range(3)]
for item in data:
yield item
for item in func():
print(item)理解了生成器和迭代器，再来看请求流，我们使用生成器和迭代器，可以很轻松的实现客户端的代码，具体为：
#!/usr/bin/env python
#!coding:utf-8
import grpc
import helloworld_pb2
import helloworld_pb2_grpc
def generateRegister():
datas=[
{"username":"wuya","password":"asd888"},
{"username":"无涯","password":"asd888"}
]
for item in datas:
msg=helloworld_pb2.RegisterRequest(
username=item['username'],
password=item['password'])
yield msg
async def send_register():
channel=grpc.insecure_channel('gRPC服务端地址')
stub=helloworld_pb2_grpc.GreeterStub(channel=channel)
response=stub.Register(generateRegister())
#拿到服务端的数据
return response.token针对应答流相对而言比较简单，但是使用异步的方式来拿回结果，这地方已Login接口为案例，具体实现代码如下：
#!/usr/bin/env python
#!coding:utf-8
import grpc
import helloworld_pb2
import helloworld_pb2_grpc
async def send_login():
channel=grpc.insecure_channel('gRPC服务端地址')
stub=helloworld_pb2_grpc.GreeterStub(channel=channel)
response=stub.LoginRequest(username='wuya',password='asd888')
datas=[]
async for item in response:
data={
"nick":item.nick,
"usernam":item.username,
"address":item.address
}
datas.append(data)
return datas拿回结果的过程中，一定要使用async的方式，否则就会报
TypeError: 'UnaryStreamCall' object is not iterable的错误信息，这点需要特别注意。
最后来演示下双向流的部分，也就是说N次请求，N次回应，测试案例代码如下：
#!/usr/bin/env python
#!coding:utf-8
import grpc
import helloworld_pb2
import helloworld_pb2_grpc
def generateProfile():
datas=[
{"nick":"wuya"},
{"nick":"无涯"}
]
for item in datas:
msg=helloworld_pb2.ProfileRequest(
nick=item['nick'])
yield msg
async def send_profile():
channel=grpc.insecure_channel('gRPC服务端地址')
stub=helloworld_pb2_grpc.GreeterStub(channel=channel)
response=stub.Profile(generateProfile())
datas=[]
async for item in response:
data={
"nick":item.nick,
"usernam":item.username,
"phone":item.phone
}
datas.append(data)
return datas如上主要演示了在gRPC的协议里面针对不同模式的案例和应用实战，后续主要演示下异步编程里面针对事件循环的任务管理和事件循环如何跳出循环，以及怎么和主流的测试框架Pytest整合起来。
感谢您的阅读，后续会持续更新！
腾讯云开发者
