用Python进行websocket接口测试

待我称王封你为后i 2023-10-16 18:15 118阅读 0赞

![8e92d1c29d0c46b2b2cab4f16f549d5b.png][]

这篇文章主要介绍了用Python进行websocket接口测试，帮助大家更好的理解和使用python，感兴趣的朋友可以了解下

我们在做接口测试时，除了常见的http接口，还有一种比较多见，就是socket接口，今天讲解下怎么用Python进行websocket接口测试。

### Socket ###

Socket又称"套接字"，应用程序通常通过"套接字"向网络发出请求或者应答网络请求，使主机间或者一台计算机上的进程间可以通讯。

Python 中，我们使用 socket 模块的 socket 函数来创建一个 socket 对象。语法格式如下：

> socket.socket ( family ，type ，proto)

现在大多数用的都是websocket，那我们就先来安装一下websocket的安装包。

<table> 
 <tbody> 
  <tr> 
   <td> <p>1</p> </td> 
   <td> <p><code>pip&nbsp;</code><code>install</code>&nbsp;<code>websocket-client</code></p> </td> 
  </tr> 
 </tbody> 
</table>

![72f2f72991a34ef1bf0248cc340e6be4.png][]

安装完之后，我们就开始我们的websocket之旅了。

在websocket里，我们有常用的这几个方法：

**on\_message方法**

def on_message(ws, message):
      print(message)

on\_message是用来接受消息的，server发送的所有消息都可以用on\_message这个方法来收取。

**on\_error方法：**

def on_error(ws, error):
      print(error)

这个方法是用来处理错误异常的，如果一旦socket的程序出现了通信的问题，就可以被这个方法捕捉到。

**on\_open方法：**

def on_open(ws):
      def run(*args):
        for i in range(30):
          # send the message, then wait
          # so thread doesn't exit and socket
          # isn't closed
          ws.send("Hello %d" % i)
          time.sleep(1)
    
        time.sleep(1)
        ws.close()
        print("Thread terminating...")
    
      Thread(target=run).start()

on\_open方法是用来保持连接的，上面这样的一个例子，就是保持连接的一个过程，每隔一段时间就会来做一件事,他会在30s内一直发送hello。最后停止。

**on\_close方法：**

def on_close(ws):
      print("### closed ###")

onclose主要就是关闭socket连接的。

如何创建一个websocket应用：

ws = websocket.WebSocketApp("wss://echo.websocket.org")

括号里面就是你要连接的socket的地址，在WebSocketApp这个实例化的方法里面还可以有其他参数，这些参数就是我们刚刚介绍的这些方法。

ws = websocket.WebSocketApp("ws://echo.websocket.org/",
                  on_message=on_message,
                  on_error=on_error,
                  on_close=on_close)

**完整代码：**

import websocket
    from threading import Thread
    import time
    import sys
    
    def on_message(ws, message):
      print(message)
    
    def on_error(ws, error):
      print(error)
    
    def on_close(ws):
      print("### closed ###")
    
    def on_open(ws):
      def run(*args):
        for i in range(3):
          # send the message, then wait
          # so thread doesn't exit and socket
          # isn't closed
          ws.send("Hello %d" % i)
          time.sleep(1)
    
        time.sleep(1)
        ws.close()
        print("Thread terminating...")
    
      Thread(target=run).start()
    
    
    if __name__ == "__main__":
    
      websocket.enableTrace(True)
      host = "ws://echo.websocket.org/"
      ws = websocket.WebSocketApp(host,
                    on_message=on_message,
                    on_error=on_error,
                    on_close=on_close)
      ws.on_open = on_open
      ws.run_forever()

也可以使用python进行websocket的客户端压力测试。

# -*- coding:utf-8 -*-
    # __author__ == 'chenmingle'
     
    import websocket
    import time
    import threading
    import json
    import multiprocessing
    import uuid
    from threadpool import ThreadPool, makeRequests
     
    # 修改成自己的websocket地址
    WS_URL = "xxxx"
    # 定义进程数
    processes = 4
    # 定义线程数（每个文件可能限制1024个，可以修改fs.file等参数）
    thread_num = 700
    index = 1
     
     
    def on_message(ws, message):
     # print(message)
     pass
     
     
    def on_error(ws, error):
     print(error)
     pass
     
     
    def on_close(ws):
     # print("### closed ###")
     pass
     
     
    def on_open(ws):
     global index
     index = index + 1
     
     def send_thread():
      # 设置你websocket的内容
      # 每隔10秒发送一下数据使链接不中断
      while True:
       ws.send(u'hello服务器')
       time.sleep(10)
     
     t = threading.Thread(target=send_thread)
     t.start()
     
     
    def on_start(num):
     time.sleep(5)
     # websocket.enableTrace(True)
     ws = websocket.WebSocketApp(WS_URL + str(num),
            on_message=on_message,
            on_error=on_error,
            on_close=on_close)
     ws.on_open = on_open
     ws.run_forever()
     
     
    def thread_web_socket():
     # 线程池
     pool_list = ThreadPool(thread_num)
     num = list()
     # 设置开启线程的数量
     for ir in range(thread_num):
      num.append(ir)
     requests = makeRequests(on_start, num)
     [pool_list.putRequest(req) for req in requests]
     pool_list.wait()
     
     
    if __name__ == "__main__":
     # 进程池
     pool = multiprocessing.Pool(processes=processes)
     # 设置开启进程的数量
     for i in xrange(processes):
      pool.apply_async(thread_web_socket)
     pool.close()
     pool.join()

以上就是用Python进行websocket接口测试的详细内容。

**最后：**下方这份完整的软件测试视频教程已经整理上传完成，需要的朋友们可以自行领取**【保证100%免费】**

![8b0171187dce405e878f274a8cdd9ec0.png][]

#### **软件测试面试文档** ####

我们学习必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有字节大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。

![a3e4226c935c4242804cbdf26b571bc6.png][]

![58df6cd288094ad8874e814f2477620b.gif][]

[8e92d1c29d0c46b2b2cab4f16f549d5b.png]: https://img-blog.csdnimg.cn/8e92d1c29d0c46b2b2cab4f16f549d5b.png
[72f2f72991a34ef1bf0248cc340e6be4.png]: https://img-blog.csdnimg.cn/72f2f72991a34ef1bf0248cc340e6be4.png
[8b0171187dce405e878f274a8cdd9ec0.png]: https://img-blog.csdnimg.cn/8b0171187dce405e878f274a8cdd9ec0.png
[a3e4226c935c4242804cbdf26b571bc6.png]: https://img-blog.csdnimg.cn/a3e4226c935c4242804cbdf26b571bc6.png
[58df6cd288094ad8874e814f2477620b.gif]: https://img-blog.csdnimg.cn/58df6cd288094ad8874e814f2477620b.gif