Как замокировать метод только для чтения с помощью pytest-mock

Вопрос или проблема

Я хочу протестировать свои клиентские и серверные реализации, которые используют сокет для связи. Поэтому я пытался использовать 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, мы сталкиваемся с некоторыми ограничениями, связанными с тем, что эти методы являются в некоторых случаях «только для чтения». Однако есть способ обойти эту проблему.

Подход к решению

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

  2. Использование 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'

    # ... другие тесты

Объяснение кода

  1. Создание MockSocket. Этот класс наследуется от socket.socket и переопределяет методы recv и sendall, чтобы они были управляемы deque, который служит временным хранилищем для передаваемых данных.

  2. Имитация поведения. Метод recv проверяет, есть ли данные в очереди, а метод sendall добавляет данные в очередь. Если попытаться получить данные, когда их нет, генерируется исключение socket.timeout.

  3. Тестируемая логика. В тесте создается экземпляр MockSocket, который используется для тестирования средств связи. Таким образом, вам не нужно устанавливать реальные соединения, и вы можете тестировать логику вашей программы.

Заключение

Такой подход позволяет вам мокировать поведение сокетов и легко транспортировать данные между сервером и клиентом в тестах. Используя данный метод, вы сможете избежать ошибок, связанных с попытками изменения «только для чтения» методов оригинального класса socket.

Оцените материал
Добавить комментарий

Капча загружается...