用Python进行websocket接口测试

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

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

Socket

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

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

socket.socket ( family ，type ，proto)

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

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

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

on_message方法

1. def on_message(ws, message):

2. print(message)

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

on_error方法：

1. def on_error(ws, error):

2. print(error)

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

on_open方法：

1. def on_open(ws):

2. def run(*args):

3. for i in range(30):

4. # send the message, then wait

5. # so thread doesn't exit and socket

6. # isn't closed

7. ws.send("Hello %d" % i)

8. time.sleep(1)

9. time.sleep(1)

10. ws.close()

11. print("Thread terminating...")

12. Thread(target=run).start()

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

on_close方法：

1. def on_close(ws):

2. print("### closed ###")

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

如何创建一个websocket应用：

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

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

1. ws = websocket.WebSocketApp("ws://echo.websocket.org/",

2. on_message=on_message,

3. on_error=on_error,

4. on_close=on_close)

完整代码：

1. import websocket

2. from threading import Thread

3. import time

4. import sys

5. def on_message(ws, message):

6. print(message)

7. def on_error(ws, error):

8. print(error)

9. def on_close(ws):

10. print("### closed ###")

11. def on_open(ws):

12. def run(*args):

13. for i in range(3):

14. # send the message, then wait

15. # so thread doesn't exit and socket

16. # isn't closed

17. ws.send("Hello %d" % i)

18. time.sleep(1)

19. time.sleep(1)

20. ws.close()

21. print("Thread terminating...")

22. Thread(target=run).start()

23. if __name__ == "__main__":

24. websocket.enableTrace(True)

25. host = "ws://echo.websocket.org/"

26. ws = websocket.WebSocketApp(host,

27. on_message=on_message,

28. on_error=on_error,

29. on_close=on_close)

30. ws.on_open = on_open

31. ws.run_forever()

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

1. # -*- coding:utf-8 -*-

2. # __author__ == 'chenmingle'

3. import websocket

4. import time

5. import threading

6. import json

7. import multiprocessing

8. import uuid

9. from threadpool import ThreadPool, makeRequests

10. # 修改成自己的websocket地址

11. WS_URL = "xxxx"

12. # 定义进程数

13. processes = 4

14. # 定义线程数（每个文件可能限制1024个，可以修改fs.file等参数）

15. thread_num = 700

16. index = 1

17. def on_message(ws, message):

18. # print(message)

19. pass

20. def on_error(ws, error):

21. print(error)

22. pass

23. def on_close(ws):

24. # print("### closed ###")

25. pass

26. def on_open(ws):

27. global index

28. index = index + 1

29. def send_thread():

30. # 设置你websocket的内容

31. # 每隔10秒发送一下数据使链接不中断

32. while True:

33. ws.send(u'hello服务器')

34. time.sleep(10)

35. t = threading.Thread(target=send_thread)

36. t.start()

37. def on_start(num):

38. time.sleep(5)

39. # websocket.enableTrace(True)

40. ws = websocket.WebSocketApp(WS_URL + str(num),

41. on_message=on_message,

42. on_error=on_error,

43. on_close=on_close)

44. ws.on_open = on_open

45. ws.run_forever()

46. def thread_web_socket():

47. # 线程池

48. pool_list = ThreadPool(thread_num)

49. num = list()

50. # 设置开启线程的数量

51. for ir in range(thread_num):

52. num.append(ir)

53. requests = makeRequests(on_start, num)

54. [pool_list.putRequest(req) for req in requests]

55. pool_list.wait()

56. if __name__ == "__main__":

57. # 进程池

58. pool = multiprocessing.Pool(processes=processes)

59. # 设置开启进程的数量

60. for i in xrange(processes):

61. pool.apply_async(thread_web_socket)

62. pool.close()

63. pool.join()

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

感谢每一个认真阅读我文章的人，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走：

这些资料，对于【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴上万个测试工程师们走过最艰难的路程，希望也能帮助到你!有需要的小伙伴可以点击下方小卡片领取