WebSocket — это тип полнодуплексной связи по одному TCP-соединению (полный дуплекс — это термин, обозначающий передачу данных. Связь позволяет передавать данные в обоих направлениях одновременно, и по своим возможностям она эквивалентна комбинации двух методов симплексной связи. Полнодуплексный режим относится к протоколу, который может передавать сигналы в обоих направлениях (A→B и B→A) одновременно (мгновенно). Это относится к A→B и B→A одновременно, что является мгновенным и быстрым. синхронно).

Протокол связи WebSocket был установлен IETF в 2011 году как стандарт RFC 6455 и дополнен RFC7936. WebSocket API (WebSocket API — это интерфейс, который использует протокол WebSocket для создания полнодуплексного канала для отправки и получения сообщений) также был признан W3C стандартом.

Протокол HTTP не поддерживает постоянные соединения. Хотя в HTTP 1.1 были внесены улучшения, в HTTP-соединении можно отправлять несколько запросов и получать несколько ответов.

Но в HTTP Request = Response всегда истинно, а это значит, что на запрос может быть только один ответ. Более того, эта реакция также пассивна и не может быть инициирована активно.

Websocket обычно используется для социальных сетей/подписки, многопользовательских игр, совместного офиса/редактирования, котировок фондов фондового рынка, прямых спортивных трансляций, аудио- и видеочата/видеоконференций/онлайн-обучения, умного дома и приложений на основе определения местоположения.

Интерфейс веб-сокета не может использовать запросы для прямого вызова интерфейса. Для реализации вызова можно использовать стороннюю библиотеку. В следующем материале рассказывается, как вызвать стороннюю библиотеку для реализации автоматического тестирования интерфейса веб-сокета.

Настоящий бой

Реализация автоматизации интерфейса веб-сокета с использованием языка Python.

Экологическая подготовка

1. Установите среду pyhton3 и загрузите необходимые библиотеки времени выполнения.

2. Загрузите необходимую библиотеку времени выполнения.

pip install websocket-client

Настоящий бой Демо

Подключиться к веб-серверу

import logging

from websocket import create_connection

logger = logging.getLogger(name)

url = 'ws://echo.websocket.org/' #Интернет-интерфейс веб-сокета с обратной связью, доступ к которому необходимо получить после подключения через веб-сокет. Доступ к нему нельзя получить напрямую, введя адрес на веб-странице.

wss = create_connection(url, timeout=timeout)

Отправить сообщение через веб-сокет

wss.send('Hello World')

Получать сообщения веб-сокета

res = wss.recv()

logger.info(res)

Закрыть соединение через веб-сокет

wss.close()

Вызов сторонней библиотеки веб-сокета не поддерживает прямую отправку типов данных, отличных от строк, поэтому перед отправкой запроса вам необходимо преобразовать структурированный формат Python в строковый тип или строку json, а затем инициировать запрос интерфейса веб-сокета. .

#Формат тела данных для отправки:

data= {

"a" : "abcd",

"b" : 123

}

Перед отправкой данные необходимо преобразовать в строку json.

new_data=json.dumps(data,ensure_ascii=False)

wss.send(new_data)

Обработка полученного тела данных: Если интерфейс определен как json, поскольку данные передаются в строковом формате, полученное тело данных необходимо преобразовать

Формат тела полученных данных также является строкой.

logger.info(type(res)) # <class 'str'>

Обработка преобразования формата содержимого ответа:

def load_json(base_str):

if isinstance(base_str, str):

    try:

        res = json.loads(base_str)

        return load_json(res)

    except JSONDecodeError:

        return base_str

elif isinstance(base_str, list):

    res = []

    for i in base_str:

        res.append(load_json(i))

    return res

elif isinstance(base_str, dict):

    for key, value in base_str.items():

        base_str[key] = load_json(value)

    return base_str

return base_str

Автоматическое тестирование интерфейса Websocket, демонстрация вторичной упаковки

web_socket_util.py инкапсулирует общие операции интерфейса веб-сокета:

import logging

import json

from websocket import create_connection

logger = logging.getLogger(name)

class WebsocketUtil():

def conn(self, uri, timeout=3):

    '''

    соединятьwebсервер

    :param uri: URL-адрес службы

    :param timeout: тайм-аут

    :return:

    '''

    self.wss = create_connection(uri, timeout=timeout)

def send(self, message):

    '''

    отправить тело данных запроса

    :param message: Информация о данных для отправки

    :return:

    '''

    if not isinstance(message, str):

        message = json.dumps(message)

    return self.wss.send(message)

def load_json(self, base_str):

    '''

    Обработка тела данных

    :param base_str: Тело данных, подлежащих обработке

    :return:

    '''

    if isinstance(base_str, str):

        try:

            res = json.loads(base_str)

            return self.load_json(res)

        except JSONDecodeError:

            return base_str

    elif isinstance(base_str, list):

        res = []

        for i in base_str:

            res.append(self.load_json(i))

        return res

    elif isinstance(base_str, dict):

        for key, value in base_str.items():

            base_str[key] = self.load_json(value)

        return base_str

    return base_str

def recv(self, timeout=3):

    '''

    перенимать информацию тела данных и вызывать метод обработки тела данных для обработки тела ответа

    :param timeout: тайм-аут

    :return:

    '''

    if isinstance(timeout, dict):

        timeout = timeout["timeout"]

    try:

        self.settimeout(timeout)

        recv_json = self.wss.recv()

        all_json_recv = self.load_json(recv_json)

        self._set_response(all_json_recv)

        return all_json_recv

    except WebSocketTimeoutException:

        logger.error(f"Нет данных для перенимания более {timeout} секунд")

def settimeout(self, timeout):

    '''

    настраиватьтайм-аут

    :param timeout: тайм-аут

    :return:

    '''

    self.wss.settimeout(timeout)

def recv_all(self, timeout=3):

    '''

перенимать Множественную информацию тела данных,И вызовите метод обработки тела данных для обработки тела ответа.

    :param timeout: тайм-аут

    :return:

    '''

    if isinstance(timeout, dict):

        timeout = timeout["timeout"]

    recv_list = []

    while True:

        try:

            self.settimeout(timeout)

            recv_json = self.wss.recv()

            all_json_recv = self.load_json(recv_json)

            recv_list.append(all_json_recv)

            logger.info(f"all::::: {all_json_recv}")

        except WebSocketTimeoutException:

            logger.error(f"Нет данных для перенимания более {timeout} секунд")

            break

    self._set_response(recv_list)

    return recv_list

def close(self):

    '''

    закрытиесоединять

    :return:

    '''

    return self.wss.close()

def _set_response(self, response):

    self.response = response

def _get_response(self) -> list:

    return self.response

test_case.py автоматизированный тестовый пример интерфейса веб-сокета:

class TestWsDemo:

def setup(self):

    url = 'ws://echo.websocket.org/'

    self.wss = WebsocketUtil()

    self.wss.conn(url)

def teardown(self):

    self.wss.close()

def test_demo(self):

    data = {"a": "hello", "b": "world"}

    self.wss.send(data)

    res = self.wss.recv()

    assert 'hello' == res['a']

WebSocket упрощает обмен данными между клиентом и сервером, позволяя серверу активно передавать данные клиенту.

В API WebSocket браузеру и серверу нужно только завершить рукопожатие, и между ними может быть создано постоянное соединение для двунаправленной передачи данных.