
Привіт, друзі! У вас часто буває, що ви читаєте про нову технологію і думаєте: «Оце круто! Я б дуже хотів сам з таким попрацювати!»?
А потім ніколи в роботі її не зустрічаєте, бо у вас багато легасі, а ключові системи взагалі ще динозаври на коболі писали. У мене така історія з веб сокетами — часто про них чую, читаю, але в основній роботі ніколи не зустрічав (на відміну від звичайних TCP сокетів).
Сама технологія проста як табурет, і мені, як практикуючому спеціалісту, цікаво — якщо завтра вона з’явиться в моїй роботі — як мені з нею взаємодіяти?
Для HTTP в нас є ціла купа інструментів — від curl до Postman. Можна навіть JMeter використати як клієнт, чи написати свій на Python чи JS (прикладів є повно).
Перше, що я роблю, коли чогось не знаю — гуглю. Здається, знайшов кілька тулів (забігаючи на перед, скажу, що найбільше мені сподобався цей — Websocket King).
Тепер треба інструмент протестувати. І тут мені закортіло не просто підключитись та відправити повідомлення, а подивитись, як клієнт працює в різних режимах зв’язку.
Що я маю на увазі? Працюючи з HTTP, ми звикли, що на кожен запит сервер надсилає відповідь. 200 OK, 400 НЕ ОК, 500 ДУЖЕ НЕ ОК і т.д.
Веб сокети ж дозволяють пересилати дані в будь-якому порядку:
- запит — відповідь
- запит — дві+ відповіді
- запит без відповіді
- відповідь без запита
Я пошукав приклади веб сокет серверу на Python, взяв приклад тут, трохи його допрацював і в мене вийшов наступний код:
import asyncio
import websockets
import logging
from websockets import WebSocketServerProtocol
logging.basicConfig(level=logging.INFO)
class Server:
clients = set()
counter = 0
async def periodic_message(self, period: int):
while True:
await asyncio.sleep(period)
if self.clients:
await self.send_2_clients(f'hello world {self.counter}')
logging.info(f'periodic message sent {self.counter} times')
self.counter += 1
async def register(self, ws: WebSocketServerProtocol):
self.clients.add(ws)
logging.info(f'{ws.remote_address} connected')
async def unregister(self, ws: WebSocketServerProtocol):
self.clients.remove(ws)
logging.info(f'{ws.remote_address} disconnected')
async def send_2_clients(self, message: str):
await asyncio.wait([client.send(message) for client in self.clients])
async def distribute(self, ws: WebSocketServerProtocol):
async for message in ws:
if 'ignore' in message:
logging.info('ignore response')
elif 'double' in message:
logging.info('double response')
await self.send_2_clients(f'{message} #1')
await self.send_2_clients(f'{message} #2')
else:
await self.send_2_clients(message)
async def ws_handler(self, ws: WebSocketServerProtocol, uri: str):
await self.register(ws)
try:
await self.distribute(ws)
finally:
await self.unregister(ws)
server = Server()
start_srv = websockets.serve(server.ws_handler, 'localhost', 4000)
loop1 = asyncio.get_event_loop()
loop2 = asyncio.get_event_loop()
loop1.run_until_complete(start_srv)
loop2.run_until_complete(server.periodic_message(5))
loop1.run_forever()
loop2.run_forever()
Щоб його запустити, треба мати Python 3.8+ а також виконати команду
pip install websockets
Терер відкриваємо клієнт, підключаємось за посиланням ws://localhost:4000 та починаємо посилати повідомлення.
Працює!

Але це ще не все.
Уявимо, що в нас вже є фронтенд, що взаємодіє з бекендом через веб сокети. І нам, як тестувальникам, цікаво, що ж там передається?
Для то, щоб це перевірити, зайдемо в браузері Chrome на сайт, де є вбудований тест.
- Перевіримо, що ваш браузер підтримує технологію
- Відкриємо Dev Tools (F12), вкладинку Network
- Вкажемо тип фільтру WS
- Введемо ws://localhost:4000 в Location та натиснемо Connect
- Тепер маємо побачити в Dev Tools, що в нас відкрито 1 з’єднання з сервером
- Натискаємо на нього, відкриваємо вкладинку Messages
Дивіться, всі повідомлення тут!

Власне, це все, чим я хотів поділитись. Пишіть, чи ви тестуєте ПЗ з веб сокетами? Чим користуєтесь? Які цікаві випадки у вас з ними були?