Вопрос или проблема
Я хочу протестировать свои клиентские и серверные реализации, которые используют сокет для связи. Поэтому я пытался использовать pytest-mock
, чтобы заменить методы recv
и sendall
у socket
, чтобы не нужно было устанавливать реальное соединение.
Я написал следующий код:
from collections import deque
from threading import Lock
import socket
from time import sleep
from typing import Deque
import pytest
from pytest_mock import MockFixture
from comms import Comms
def mocked_socket(mocker: MockFixture):
connection = socket.socket()
backing_store: Deque[int] = deque()
lock = Lock()
def receiver(n: int):
with lock:
# ЗАМЕТКА: Чтобы лучше смоделировать поведение сокета, мы позволяяем отправителю пополнять резервное хранилище на время connection.timeout.
bl = len(backing_store)
if bl == 0 and connection.timeout is not None:
lock.release()
sleep(connection.timeout)
lock.acquire()
# ИСПРАВИТЬ: Мы вызываем тайм-аут здесь, даже если connection.timeout равно None (т.е. сокет блокирующий). Это связано с тем, что возврат пустого объекта байтов интерпретируется как закрытие соединения другой стороной. Если мы хотим еще точнее смоделировать сокет, нам следует иметь условную переменную для принятия, на которую этот поток может ждать.
bl = len(backing_store)
if bl == 0:
raise socket.timeout
n_cap = min(len(backing_store), n)
return bytes(backing_store.popleft() for _ in range(n_cap))
def sender(buffer: bytes):
with lock:
backing_store.extend(buffer)
mocker.patch.object(connection, "recv", wraps=receiver)
mocker.patch.object(connection, "sendall", wraps=sender)
return connection
def test(mocker: MockFixture):
connection = mocked_socket(mocker)
server = Comms(connection)
client = Comms(connection)
# ... что угодно, что я хочу протестировать
Однако, когда я пытаюсь запустить pytest
, я получаю множество таких ошибок:
self.temp_original = original
self.is_local = local
self._exit_stack = contextlib.ExitStack()
try:
> setattr(self.target, self.attribute, new_attr)
E AttributeError: 'socket' object attribute 'recv' is read-only
/usr/lib/python3.12/unittest/mock.py:1567: AttributeError
Так что, похоже, то, что я пытаюсь сделать, невозможно, потому что подмена recv
/sendall
требует удаления их оригинальных реализаций, что не работает, так как они помечены как только для чтения.
Я нашел Как подменить свойство только для чтения с помощью mock?, но это касается только подмены свойств, а не методов.
Я никогда раньше не использовал pytest
, unittest.mock
и т.д., так что есть ли простое решение, которое я упускаю? Правильный ли подход, который я пытаюсь использовать?
Спасибо за вашу помощь.
Ответ или решение
Для имитации методов, таких как recv
и sendall
у объектов типа socket
с использованием библиотеки pytest-mock
, мы сталкиваемся с некоторыми ограничениями, связанными с тем, что эти методы являются в некоторых случаях «только для чтения». Однако есть способ обойти эту проблему.
Подход к решению
-
Создание класса-наследника. Вместо того, чтобы пытаться модифицировать методы существующего объекта
socket
, вы можете создать подклассsocket.socket
, который переопределяет методыrecv
иsendall
. -
Использование
pytest-mock
для создания экземпляра этого класса. Мы можем применятьmocker.patch
для подмены методовrecv
иsendall
.
В качестве примера, можно использовать следующий код:
from collections import deque
from threading import Lock
import socket
from time import sleep
from typing import Deque
import pytest
from pytest_mock import MockFixture
# Создаем подкласс для сокета
class MockSocket(socket.socket):
def __init__(self):
super().__init__()
self.backing_store: Deque[bytes] = deque()
self.lock = Lock()
def recv(self, n: int) -> bytes:
with self.lock:
if not self.backing_store:
raise socket.timeout
n_cap = min(len(self.backing_store), n)
data = b''.join(self.backing_store.popleft() for _ in range(n_cap))
return data
def sendall(self, buffer: bytes):
with self.lock:
self.backing_store.append(buffer)
def test(mocker: MockFixture):
# Создаем экземпляр нашего мок-сокета
connection = MockSocket()
server = Comms(connection)
client = Comms(connection)
# Пример отправки и получения данных
client.sendall(b'Hello')
data = server.recv(5)
assert data == b'Hello'
# ... другие тесты
Объяснение кода
-
Создание
MockSocket
. Этот класс наследуется отsocket.socket
и переопределяет методыrecv
иsendall
, чтобы они были управляемыdeque
, который служит временным хранилищем для передаваемых данных. -
Имитация поведения. Метод
recv
проверяет, есть ли данные в очереди, а методsendall
добавляет данные в очередь. Если попытаться получить данные, когда их нет, генерируется исключениеsocket.timeout
. -
Тестируемая логика. В тесте создается экземпляр
MockSocket
, который используется для тестирования средств связи. Таким образом, вам не нужно устанавливать реальные соединения, и вы можете тестировать логику вашей программы.
Заключение
Такой подход позволяет вам мокировать поведение сокетов и легко транспортировать данные между сервером и клиентом в тестах. Используя данный метод, вы сможете избежать ошибок, связанных с попытками изменения «только для чтения» методов оригинального класса socket
.