Быстрое обнаружение или моделирование WSAECONNREFUSED

У сокетов Windows есть странное поведение, когда дело доходит до WSAECONNREFUSED (что означает, что полное заполнение или порт недоступен, см. qaru.site/info/492418/...). Если Windows обнаруживает одно из этих условий, оно повторяет (до) два раза с интервалом 0,5 с. Это означает, что для обнаружения WSAECONNREFUSED при попытке подключения к сокету требуется не менее 1 секунды (http://support.microsoft.com/kb/175523/en-us).

Есть ли способ ускорить это обнаружение, не испортив значения реестра? Мне нужно моделировать отказ от соединений сокетов в unittests. Обходной путь, например, имитация отказавшего соединения с сырыми сокетами, также будет приемлемым.

Вот простой Python script, демонстрирующий проблему:

import errno
import socket
import time

PORT = 50123


def main():
    s = socket.socket()
    s.bind(('127.0.0.1', PORT))
    s.listen(0)
    client = socket.socket()
    client.connect(('127.0.0.1', PORT))

    client2 = socket.socket()
    start = time.time()

    try:
        client2.connect(('127.0.0.1', PORT))
    except socket.error as e:
        assert e.errno == errno.WSAECONNREFUSED
        print 'connection attempt took', time.time() - start
    finally:
        client2.close()
        client.close()
        s.close()


if __name__ == '__main__':
    main()

Ответ 1

Это не совсем то, о чем вы просили. Но если вам нужно это только в unittests, полезно будет mock библиотека.

import errno
import socket
import time
import mock

PORT = 50123


def connect_mock(*agrs):
    raise socket.error(errno.WSAECONNREFUSED, "Testing")


def main():
    s = socket.socket()
    s.bind(('127.0.0.1', PORT))
    s.listen(0)
    client = socket.socket()
    client.connect(('127.0.0.1', PORT))

    client2 = socket.socket()
    start = time.time()

    with mock.patch('socket.socket.connect', connect_mock):
        try:
            client2.connect(('127.0.0.1', PORT))
            print "done"
        except socket.error as e:
            assert e.errno == errno.WSAECONNREFUSED
            print 'connection attempt took', time.time() - start
        finally:
            client2.close()
            client.close()
            s.close()


if __name__ == '__main__':
    main()

Ответ 2

Вот мое решение, основанное на dmitry-vakhrushev answer, которое исправляет метод подключения более интеллектуальным:

if sys.platform == 'win32':
    n_calls = [0]
    org_connect = socket.socket.connect

    def refusing_connect(*args):
        if n_calls[0] < 2:
            n_calls[0] += 1
            raise socket.error(errno.WSAECONNREFUSED, "Testing")
        return org_connect(*args)

    # patch socket.connect to speed up WSAECONNREFUSED detection
    patcher = mock.patch('socket.socket.connect', refusing_connect)
    patcher.start()
    self.addCleanup(patcher.stop)