# HG changeset patch # User RhodeCode Admin # Date 2023-01-31 07:55:12 # Node ID 6b029be8c7e4df4044875d95655fb75170c9dfac # Parent 2ecd5d81436b87096d8789d5947166e9b6444234 pick_port: unified code for testing/hooks diff --git a/rhodecode/lib/hooks_daemon.py b/rhodecode/lib/hooks_daemon.py --- a/rhodecode/lib/hooks_daemon.py +++ b/rhodecode/lib/hooks_daemon.py @@ -158,18 +158,8 @@ class HttpHooksCallbackDaemon(ThreadedHo return socket.gethostname() or '127.0.0.1' def get_available_port(self, min_port=20000, max_port=65535): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - hostname = self.get_hostname() - - for _ in range(min_port, max_port): - pick_port = random.randint(min_port, max_port) - try: - sock.bind((hostname, pick_port)) - sock.close() - del sock - return pick_port - except OSError: - pass + from rhodecode.lib.utils2 import get_available_port as _get_port + return _get_port(min_port, max_port) def _prepare(self, txn_id=None, host=None, port=None): if not host or host == "*": diff --git a/rhodecode/lib/utils2.py b/rhodecode/lib/utils2.py --- a/rhodecode/lib/utils2.py +++ b/rhodecode/lib/utils2.py @@ -35,6 +35,8 @@ import urllib import urlobject import uuid import getpass +import socket +import random from functools import update_wrapper, partial, wraps import pygments.lexers @@ -1171,3 +1173,21 @@ def user_agent_normalizer(user_agent_raw raise return ua + + +def get_available_port(min_port=40000, max_port=55555): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + hostname = '127.0.0.1' + pick_port = min_port + + for _ in range(min_port, max_port): + pick_port = random.randint(min_port, max_port) + try: + sock.bind((hostname, pick_port)) + sock.close() + break + except OSError: + pass + + del sock + return pick_port diff --git a/rhodecode/tests/pylons_plugin.py b/rhodecode/tests/pylons_plugin.py --- a/rhodecode/tests/pylons_plugin.py +++ b/rhodecode/tests/pylons_plugin.py @@ -199,18 +199,8 @@ def ini_settings(ini_config): def get_available_port(min_port=40000, max_port=55555): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - hostname = '127.0.0.1' - - for _ in range(min_port, max_port): - pick_port = random.randint(min_port, max_port) - try: - sock.bind((hostname, pick_port)) - sock.close() - del sock - return pick_port - except OSError: - pass + from rhodecode.lib.utils2 import get_available_port as _get_port + return _get_port(min_port, max_port) @pytest.fixture(scope='session')