ChatGPT使用的SSE协议接口怎么做测试

SSE是server-sent events协议简称，SSE协议主要是依托于HTTP链接用来从服务端将消息、信息、事件推动给客户端的协议。

SSE最近突然的被很多人关注还有一个主要原因就是ChatGPT等大模型的聊天类系统就采用了SSE协议。在使用ChatGPT的时候，输入Prompt后的反馈是逐渐的显示在聊天区域的，这部分的实现就是主要基于EventStream的事件流，类似打字机一样输出的，这就是SSE的实现，其实类似一下股票行情推送、期货行情推送都可以使用SSE实现。

SSE采用了long-polling的机制，Client端发送了一个HTTP的请求到Server端，Server端就保持了这个HTTP的链接并周期性的发送消息给Client端。SSE主要解决了Server端不断向Client端发送实时数据的场景需求，因为SSE是基于HTTP协议的所以对于浏览器兼容性非常好。在解决实时传输场景的技术方案中，WebSocket是目前更加广泛的解决方案，虽然SSE和WebSocket都是为了实现实时性传输、低延时通信而出现的协议，但是它们一些区别：

SSE是基于HTTP协议的长连接，客户端发起请求后会保持连接不断直到服务器主动关闭或者网络通信故障导致断开为止，在连接之上也只是Server端发送数据给Client端的单向通信，在大量高频小数据传输或即时交互的应用中并没有WebSocket高效。
SSE是基于HTTP协议的，每次传输都是需要消耗一定的带宽，而WebSocket是二进制传输协议，对于带宽消耗就很小。
最大的区别就是SSE是单向通信的，WebSocket是双向通信的。所以SSE比较适合解决一些推送类的场景例如股票行情、IoT等，WebSocket就更加适合聊天室、实时协作应用等。

SSE协议开始也是从客户端发送请求开始的，客户端发送一个Get请求到服务端，其中HTTP部分头如下：

GET /api/v1/live-scores 
Accept: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

Accept为text/event-stream就是Client端要和Server端说明它期望获得一个event stream，Connection设置成keep-alive表示这个是一个”长连接“请求。后续服务端就在这个保持的链接上不断地给Client端推动Event，传输的示意图如下：

SSE协议的通信示意图

原理就讲完了，如果作为一名测试工程师，面对一个SSE协议的接口服务，应该在怎么完成这个接口测试呢？下面就回详细的说一下。

SSE 服务端代码#

首先我们需要一个被测试系统，能够支持SSE协议的传输，源代码如下：

import json
import time

from flask import Flask, request
from flask import Response
from flask import render_template

app = Flask(__name__)


def get_message():
    """this could be any function that blocks until data is ready"""
    time.sleep(1)
    s = time.ctime(time.time())
    return json.dumps(['当前时间：' + s , 'sse server'], ensure_ascii=False)


@app.route('/')
def hello_world():
    return render_template('index.html')


@app.route('/stream')
def stream():
    user_id = request.args.get('user_id')
    print(user_id)
    def eventStream():
        id = 0
        while True:
            id +=1
            # wait for source data to be available, then push it

            yield 'id: {}\nevent: add\ndata: {}\n\n'.format(id,get_message())


    return Response(eventStream(), mimetype="text/event-stream")


if __name__ == '__main__':
    app.run()

如上代码是基于flask框架的Web服务Demo，实现了一个简单的SSE协议的接口。访问http://127.0.0.1:5000/stream就得建立的SSE链接，并且Server端不断地给客户端推动当前系统时间。

SSE客户端#

单纯运行了如上SSE的Server还并不能很好的体验SSE协议的奇妙，也就无法理解为什么类似ChatGPT会选择这个协议来完成LLM反馈的传输。我们现在写一个客户端程序，可以正确接受上面SSE的Server发送给Client端的Event。

import json
import pprint
import sseclient
import requests

def event_handler(message):
    '''
    @des  ：打印接收到的数据
    @params  :message世界收到的event
    '''
    pprint.pprint(f"Data: {message.data}")
def with_requests(url, headers):
    '''
    @des  :访问sse的服务器，建立链接
    @params  :url 服务器地址和端口，还有路由拼凑的地址，
    @params  :headers SSE是在http基础之上建立的链接，因此需要设置请求头Accept为text/event-stream
    @return  :response 建立的链接
    '''
    
    
    return requests.get(url, stream=True, headers=headers)
def test_sse_api(api_url):
    '''
    @des  : 访问sse的服务器，建立链接，然后开始接收数据
    @params  :api_url SSE是在http基础之上建立的链接
    '''
    headers = {'Accept': 'text/event-stream'}
    response = with_requests(api_url, headers)  

    #检查请求返回是否成功 
    if response.status_code == 200:
        # 如果成功，创建一个SSEClient对象
        client = sseclient.SSEClient(response)

        try:
            # 开始监听SSE的流
            for event in client.events():
                event_handler(event)
        except KeyboardInterrupt:
            print("Stopping SSE stream.")
    else:
        print(f"Failed to connect. Status code: {response.status_code}")

if __name__ == "__main__":
    api_url = 'http://localhost:5000/stream'
    
    test_sse_api(api_url)

先运行Server端，在运行上面Client端代码，你可以在IDE的控制太不断地看到刷新点额服务器端传输回来的时间，这就是在SSE协议的好处，Server端不断地推送数据给Client端，Clinet端仅仅发起了一次请求就可以不断地接受Server端返回的数据。

SSE接口的测试脚本#

那么如果有一个SSE接口，接口自动化就可以在如上Client端的代码基础之上，引入pytest类的测试框架，完成测试脚本的开发了，如上的接口测试代码如下：

import pytest
import sseclient
import requests

@pytest.fixture
def with_requests(url='http://localhost:5000/stream', headers= {'Accept': 'text/event-stream'}):
     return requests.get(url, stream=True, headers=headers)
    
def test_test(with_requests):
   
    response = with_requests 
    if response.status_code == 200:
 
        data=""
        for event in client.events():
       
            data=event.data
            break
        assert "sse" in data
    else:
        print(f"Failed to connect. Status code: {response.status_code}")

if __name__ == '__main__':
    pytest.main()