Глупый сокет.recv ()

Question

Я хочу издеваться над функцией recv сокета и иметь проблемы.Получение ошибок, которые мне нужны для подключения к моей розетке, даже если это не нужно.

Вот весь соответствующий код:

Class A:
    def __init__.py(self):
        self.s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    def connect(self):
        self.s.connect((host,port))
    def data_collection(self, d):
        self.send_over_socket(self.cg.get_privilege_command())
        self.send_over_socket(self.status_string)          
        buf = ""
        while 1:
            buf = buf + self.s.recv(self.BUFFER_SIZE).decode()
            #Removed string manipulation below
            self.publish(data from string manipulation)

    def send_over_socket(self,data):
        return (self.s.send(bytes(data.encode())))

Код Pytest;

def test_data_collection():
with mock.patch('moduleA.classA.send_over_socket') as mock_send_over_socket:
    A = classA()
    A.send_over_socket.return_value = 1
    with mock.patch('socket.socket') as mock_socket:
        mock_socket.return_value.recv.decode.return_value = "packet_string"
        with mock.patch('moduleA.ClassA.publish') as mock_publish:
            d = {}
            A.data_collection(d)
            A.publish.assert_called_with("0","500","2","1")

Answer 1

Вам не хватает return_value для вашего mock_socket;вам нужно отработать то, что socket.socket собирается вернуть.Ваш тест должен выглядеть следующим образом:

def test_data_collection():
    with mock.patch('socket.socket') as mock_socket:
        A = classA()
        mock_socket.return_value.recv.decode.return_value = "ABC123"
        buf = A.data_collection()
        assert "ABC123" == buf

Обратите внимание, что мне пришлось исправить некоторые синтаксические ошибки в коде вашего класса, и я заставил data_collection вернуть переменную buf, чтобы ее можно было проверить.

РЕДАКТИРОВАТЬ: Файл с классом:

"""receiver.py"""
import socket

IP = 'localhost'
PORT = 80


class classA:
    def __init__(self):
        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    def connect(self):
        self.s.connect((IP, PORT))
    def data_collection(self):
        buf = self.s.recv.decode()
        return buf