jiazhizhong/MediaCrawler
1
0
Fork 0
mirror of https://github.com/NanmiCoder/MediaCrawler.git synced 2026-07-24 05:10:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #925 from kapil971390/fix/proxy-provider-unknown-name-error

Browse Source 
fix(proxy): raise ValueError for unknown IP_PROXY_PROVIDER_NAME instead of crashing with AttributeError
This commit is contained in:
程序员阿江-Relakkes
2026-07-01 22:03:09 +08:00
committed by GitHub
parent 2295208b2d e3861fe9be
commit 98a2830985
2 changed files with 26 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
proxy/proxy_ip_pool.py
View File

@@ -202,11 +202,17 @@ async def create_ip_pool(ip_pool_count: int, enable_validate_ip: bool) -> ProxyI
:param enable_validate_ip: Whether to enable IP proxy validation
:return:
"""
ip_provider = IpProxyProvider.get(config.IP_PROXY_PROVIDER_NAME)
if ip_provider is None:
raise ValueError(
f"Unknown proxy provider: '{config.IP_PROXY_PROVIDER_NAME}'. "
f"Valid options: {list(IpProxyProvider.keys())}"
)
is_static = config.IP_PROXY_PROVIDER_NAME == ProviderNameEnum.STATIC_PROVIDER.value
pool = ProxyIpPool(
ip_pool_count=ip_pool_count,
enable_validate_ip=False if is_static else enable_validate_ip,
ip_provider=IpProxyProvider.get(config.IP_PROXY_PROVIDER_NAME),
ip_provider=ip_provider,
)
await pool.load_proxies()
return pool

19
test/test_proxy_ip_pool.py
View File

@@ -268,4 +268,23 @@ class TestIpPool(IsolatedAsyncioTestCase):
print(f" Is expired after setting expired proxy: {is_expired}")
self.assertTrue(is_expired, msg="Expired proxy should return True")
async def test_create_ip_pool_unknown_provider_raises_value_error(self):
"""create_ip_pool() should raise ValueError with a helpful message when
IP_PROXY_PROVIDER_NAME is not in the IpProxyProvider registry, rather than
passing None silently and crashing with AttributeError in load_proxies()."""
from proxy.proxy_ip_pool import IpProxyProvider
from unittest.mock import patch
invalid_cases = ["kuaidalli", "", "KUAIDAILI", "unknown_provider"]
for bad_name in invalid_cases:
with self.subTest(provider=bad_name):
with patch("proxy.proxy_ip_pool.config") as mock_config:
mock_config.IP_PROXY_PROVIDER_NAME = bad_name
with self.assertRaises(ValueError) as ctx:
from proxy.proxy_ip_pool import create_ip_pool
await create_ip_pool(ip_pool_count=1, enable_validate_ip=False)
self.assertIn(bad_name, str(ctx.exception))
self.assertIn(str(list(IpProxyProvider.keys())), str(ctx.exception))
print("\n=== Standalone IP proxy expiration detection test completed ===\n")