[test] Skip source address tests if the address cannot be bound to (#8900)

Fixes https://github.com/yt-dlp/yt-dlp/issues/8890

Authored by: coletdjnz

diff --git a/test/helper.py b/test/helper.py
index e5ace8fe2ceb49e32bc3714cdf40f12c7571d1d2..4aca47025e8a6ffbd4a46a392dd56a0d0e02514e 100644 (file)
--- a/test/helper.py
+++ b/test/helper.py
@@ -10,7 +10,7 @@
 import yt_dlp.extractor
 from yt_dlp import YoutubeDL
 from yt_dlp.compat import compat_os_name
 import yt_dlp.extractor
 from yt_dlp import YoutubeDL
 from yt_dlp.compat import compat_os_name

-from yt_dlp.utils import preferredencoding, try_call, write_string
+from yt_dlp.utils import preferredencoding, try_call, write_string, find_available_port

 
 if 'pytest' in sys.modules:
     import pytest
 
 if 'pytest' in sys.modules:
     import pytest


@@ -329,3 +329,8 @@ def http_server_port(httpd):
     else:
         sock = httpd.socket
     return sock.getsockname()[1]
     else:
         sock = httpd.socket
     return sock.getsockname()[1]

+
+
+def verify_address_availability(address):
+    if find_available_port(address) is None:
+        pytest.skip(f'Unable to bind to source address {address} (address may not exist)')

diff --git a/test/test_networking.py b/test/test_networking.py
index dc60ca699457f08b46168153305d56b2357bdddc..62325aa8e04e934201fb76f423119e768814d252 100644 (file)
--- a/test/test_networking.py
+++ b/test/test_networking.py
@@ -26,7 +26,7 @@
 from email.message import Message
 from http.cookiejar import CookieJar
 
 from email.message import Message
 from http.cookiejar import CookieJar
 

-from test.helper import FakeYDL, http_server_port
+from test.helper import FakeYDL, http_server_port, verify_address_availability

 from yt_dlp.cookies import YoutubeDLCookieJar
 from yt_dlp.dependencies import brotli, requests, urllib3
 from yt_dlp.networking import (
 from yt_dlp.cookies import YoutubeDLCookieJar
 from yt_dlp.dependencies import brotli, requests, urllib3
 from yt_dlp.networking import (


@@ -538,6 +538,9 @@ def test_timeout(self, handler):
     @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
     def test_source_address(self, handler):
         source_address = f'127.0.0.{random.randint(5, 255)}'
     @pytest.mark.parametrize('handler', ['Urllib', 'Requests'], indirect=True)
     def test_source_address(self, handler):
         source_address = f'127.0.0.{random.randint(5, 255)}'

+        # on some systems these loopback addresses we need for testing may not be available
+        # see: https://github.com/yt-dlp/yt-dlp/issues/8890
+        verify_address_availability(source_address)

         with handler(source_address=source_address) as rh:
             data = validate_and_send(
                 rh, Request(f'http://127.0.0.1:{self.http_port}/source_address')).read().decode()
         with handler(source_address=source_address) as rh:
             data = validate_and_send(
                 rh, Request(f'http://127.0.0.1:{self.http_port}/source_address')).read().decode()

diff --git a/test/test_socks.py b/test/test_socks.py
index 71f783e132dcccbac52d34b16535e1ddbf47c92a..cb22b61dc8a5ae6189bcaf1b8d8da10a7d2f6687 100644 (file)
--- a/test/test_socks.py
+++ b/test/test_socks.py
@@ -25,7 +25,7 @@
     ThreadingTCPServer,
 )
 
     ThreadingTCPServer,
 )
 

-from test.helper import http_server_port
+from test.helper import http_server_port, verify_address_availability

 from yt_dlp.networking import Request
 from yt_dlp.networking.exceptions import ProxyError, TransportError
 from yt_dlp.socks import (
 from yt_dlp.networking import Request
 from yt_dlp.networking.exceptions import ProxyError, TransportError
 from yt_dlp.socks import (


@@ -326,6 +326,7 @@ def test_socks4a_domain_target(self, handler, ctx):
     def test_ipv4_client_source_address(self, handler, ctx):
         with ctx.socks_server(Socks4ProxyHandler) as server_address:
             source_address = f'127.0.0.{random.randint(5, 255)}'
     def test_ipv4_client_source_address(self, handler, ctx):
         with ctx.socks_server(Socks4ProxyHandler) as server_address:
             source_address = f'127.0.0.{random.randint(5, 255)}'

+            verify_address_availability(source_address)

             with handler(proxies={'all': f'socks4://{server_address}'},
                          source_address=source_address) as rh:
                 response = ctx.socks_info_request(rh)
             with handler(proxies={'all': f'socks4://{server_address}'},
                          source_address=source_address) as rh:
                 response = ctx.socks_info_request(rh)


@@ -441,6 +442,7 @@ def test_ipv6_socks5_proxy(self, handler, ctx):
     def test_ipv4_client_source_address(self, handler, ctx):
         with ctx.socks_server(Socks5ProxyHandler) as server_address:
             source_address = f'127.0.0.{random.randint(5, 255)}'
     def test_ipv4_client_source_address(self, handler, ctx):
         with ctx.socks_server(Socks5ProxyHandler) as server_address:
             source_address = f'127.0.0.{random.randint(5, 255)}'

+            verify_address_availability(source_address)

             with handler(proxies={'all': f'socks5://{server_address}'}, source_address=source_address) as rh:
                 response = ctx.socks_info_request(rh)
                 assert response['client_address'][0] == source_address
             with handler(proxies={'all': f'socks5://{server_address}'}, source_address=source_address) as rh:
                 response = ctx.socks_info_request(rh)
                 assert response['client_address'][0] == source_address

diff --git a/test/test_websockets.py b/test/test_websockets.py
index af6142ea3b3434fe9629c83239fab3f362005ab8..91bac3442e1c96196efce0f0f73ec6dedea25818 100644 (file)
--- a/test/test_websockets.py
+++ b/test/test_websockets.py
@@ -6,6 +6,8 @@
 
 import pytest
 
 
 import pytest
 

+from test.helper import verify_address_availability
+

 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 import http.client
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 import http.client


@@ -227,6 +229,7 @@ def test_cookies(self, handler):
     @pytest.mark.parametrize('handler', ['Websockets'], indirect=True)
     def test_source_address(self, handler):
         source_address = f'127.0.0.{random.randint(5, 255)}'
     @pytest.mark.parametrize('handler', ['Websockets'], indirect=True)
     def test_source_address(self, handler):
         source_address = f'127.0.0.{random.randint(5, 255)}'

+        verify_address_availability(source_address)

         with handler(source_address=source_address) as rh:
             ws = validate_and_send(rh, Request(self.ws_base_url))
             ws.send('source_address')
         with handler(source_address=source_address) as rh:
             ws = validate_and_send(rh, Request(self.ws_base_url))
             ws.send('source_address')