Как правильно смоделировать TCP socket.socket.accept () блокирующий вызов - PullRequest
Новая
       11

Как правильно смоделировать TCP socket.socket.accept () блокирующий вызов

0 голосов
/ 30 сентября 2019

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

Но socket.socket.accept() - это метод блокировки.

Я пыталсяимитировать socket.socket.accept(), возвращая (socket, address) кортеж макета при первом вызове, а затем поднять socket.error Exception, чтобы я мог разорвать цикл в моем приложении, которое вызывает socket.socket.accept()

Как ни странновсе мои модульные тесты пройдены, но py.test процесс остается активным (что делает ?? Отладчик не помогает: он не останавливается ни на каких точках останова после второго вызова, который выходит из цикла, как ожидалось ...), перебирая ресурсы на бесконечномцикл, пока не произойдет сбой всей системы.

Итак, какой здесь правильный подход?

TcpService.py: 

import socket
import threading

class TcpService:

    _bind_ip = '127.0.0.1'
    _bind_port = 0
    _max_tcp_conn = 20

    def _handle_client_connection(self):
        pass

    def listen_tcp(self):

        self._server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._server.bind((self._bind_ip, self._tcp_bind_port))
        self._server.listen(self._max_tcp_conn)

        self._tcp_listen = True

        while self._tcp_listen:
            client_sock, address = self._server.accept() # socket.error is supposed to be raised on the second iteration during the unit test, breaking the loop
            client_handler = threading.Thread(
                target=self._handle_client_connection,
                args=(client_sock, address)
            )
            client_handler.start()

test_TcpService.py: 

import unittest
import socket
from time import sleep
import mock
import TcpService

def accept_gen():
    for i in range(1):
        mock_socket = mock.MagicMock(name='socket.socket', spec=socket.socket)
        sleep(1)
        yield (mock_socket, ['0.0.0.0', 1234])
    while True:
        sleep(1) # so I have a chance to kill the process before the OS becomes unresponsive
        yield socket.error()



class test_TcpService(unittest.TestCase):

    @mock.patch('socket.socket', autospec=True)
    def test_listen_tcp(self, mock_socket):
        mocked_socket = mock_socket.return_value
        mocked_socket.accept.side_effect = accept_gen()
        sts = TcpService()
        with self.assertRaises(socket.error):
            sts.listen_tcp()
            mock_socket.assert_called_once() # trivial POC check, final test would be more thorough...
        sts.close_tcp()

1 Ответ

0 голосов
/ 01 октября 2019

[РЕДАКТИРОВАТЬ]: Тест был в порядке. Проблема заключалась в том, что обслуживающие потоки были не демоническими, поэтому родительский поток ожидал их окончания в течение неопределенного времени, прежде чем завершить работу сокета прослушивания. Просто установив поток client_handler в качестве демона перед его запуском, он исправляет его: 

client_handler = threading.Thread(
                target=self._handle_client_connection,
                args=(client_sock, address)
            )
client_handler.setDaemon(True) # There. Fixed.
client_handler.start()

До того, как я это понял, лучшим подходом, который я нашел, было: не глушить сокет и использовать localhost и порт 0, чтобы вы моглиразверните и подключитесь к своему серверу через произвольно доступный локальный порт в любой среде, в которой вы запускаете свой набор тестов:

Python: модульное тестирование кода на основе сокетов?

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

...