Add new helper method to provide a list of N free ports (#21662)

* Add new helper method to provide a list of N free ports

* Fix changelog number

* Add test coverage

Author: AAraKKe

datadog_checks_dev/changelog.d/21662.added

@@ -0,0 +1 @@
+Add helper method to get a list of N free ports

datadog_checks_dev/datadog_checks/dev/utils.py

@@ -114,12 +114,35 @@ def mock_context_manager(obj=None):
     yield obj
 
 
-def find_free_port(ip):
+def __get_free_port_from_socket(_socket: socket.socket, ip: str) -> int:
+    _socket.bind((ip, 0))
+    _socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    return _socket.getsockname()[1]
+
+
+def find_free_port(ip: str) -> int:
     """Return a port available for listening on the given `ip`."""
     with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
-        s.bind((ip, 0))
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        return s.getsockname()[1]
+        return __get_free_port_from_socket(s, ip)
+
+
+def find_free_ports(ip: str, count: int) -> list[int]:
+    """Return `count` ports available for listening on the given `ip`."""
+    sockets: list[socket.socket] = []
+    ports: list[int] = []
+
+    try:
+        # Create and bind all sockets first to reserve the ports
+        for _ in range(count):
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sockets.append(s)
+            ports.append(__get_free_port_from_socket(s, ip))
+
+        return ports
+    finally:
+        # Close all sockets
+        for s in sockets:
+            s.close()
 
 
 def get_ip():

datadog_checks_dev/tests/test_utils.py

@@ -0,0 +1,48 @@
+# (C) Datadog, Inc. 2025-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import socket
+from contextlib import closing
+
+import pytest
+
+from datadog_checks.dev.utils import find_free_port, find_free_ports
+
+
+def assert_ports_valid(ip, ports):
+    assert all(isinstance(p, int) for p in ports)
+    assert all(0 < port <= 65535 for port in ports)
+
+    for port in ports:
+        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+            s.bind((ip, port))
+
+
+@pytest.mark.parametrize('ip', ['127.0.0.1', '0.0.0.0'], ids=['localhost', 'any_interface'])
+def test_find_free_port(ip):
+    port = find_free_port(ip)
+
+    assert_ports_valid(ip, [port])
+
+
+@pytest.mark.parametrize(
+    'ip,count',
+    [
+        ('127.0.0.1', 1),
+        ('127.0.0.1', 5),
+        ('127.0.0.1', 10),
+        ('0.0.0.0', 3),
+    ],
+    ids=['localhost_single', 'localhost_multiple', 'localhost_many', 'any_interface'],
+)
+def test_find_free_ports(ip, count):
+    ports = find_free_ports(ip, count)
+
+    assert len(ports) == count
+    assert len(ports) == len(set(ports))
+    assert_ports_valid(ip, ports)
+
+
+def test_find_free_ports_empty():
+    ports = find_free_ports('127.0.0.1', 0)
+    assert ports == []