fix: resolve core runtime bugs in tester, scheduler and proxy parsing (#229)

- utils/proxy: fix ValueError on multi-colon proxy strings in the str branch of convert_proxy_or_proxies, which made /random return 500 (#218)

- scheduler: guard None process handles in finally so ENABLE_*=false no longer raises AttributeError (#62)

- tester: replace deprecated get_event_loop()/asyncio.wait() with a fresh loop + asyncio.gather to fix 'Passing coroutines is forbidden' on Python 3.10+ (#191, #201)

- tester: reuse one aiohttp session with a bounded connector instead of one per proxy, reducing memory growth and fd exhaustion (#119, #155)

- tester: catch ClientResponseError/ContentTypeError + return_exceptions so bad httpbin responses no longer raise 'Task exception was never retrieved' (#208)

Author: Germey

proxypool/processors/tester.py

@@ -5,7 +5,8 @@
 from proxypool.storages.redis import RedisClient
 from proxypool.setting import TEST_TIMEOUT, TEST_BATCH, TEST_URL, TEST_VALID_STATUS, TEST_ANONYMOUS, \
     TEST_DONT_SET_MAX_SCORE
-from aiohttp import ClientProxyConnectionError, ServerDisconnectedError, ClientOSError, ClientHttpProxyError
+from aiohttp import ClientProxyConnectionError, ServerDisconnectedError, ClientOSError, ClientHttpProxyError, \
+    ClientResponseError, ContentTypeError
 from asyncio import TimeoutError
 from proxypool.testers import __all__ as testers_cls
 
@@ -16,6 +17,8 @@
     ServerDisconnectedError,
     ClientOSError,
     ClientHttpProxyError,
+    ClientResponseError,
+    ContentTypeError,
     AssertionError
 )
 
@@ -30,83 +33,102 @@ def __init__(self):
         init redis
         """
         self.redis = RedisClient()
-        self.loop = asyncio.get_event_loop()
         self.testers_cls = testers_cls
         self.testers = [tester_cls() for tester_cls in self.testers_cls]
 
-    async def test(self, proxy: Proxy):
+    async def test(self, proxy: Proxy, session: aiohttp.ClientSession):
         """
         test single proxy
         :param proxy: Proxy object
+        :param session: shared aiohttp session
         :return:
         """
-        async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) as session:
-            try:
-                logger.debug(f'testing {proxy.string()}')
-                # if TEST_ANONYMOUS is True, make sure that
-                # the proxy has the effect of hiding the real IP
-                # logger.debug(f'TEST_ANONYMOUS {TEST_ANONYMOUS}')
-                if TEST_ANONYMOUS:
-                    url = 'https://httpbin.org/ip'
-                    async with session.get(url, timeout=TEST_TIMEOUT) as response:
-                        resp_json = await response.json()
-                        origin_ip = resp_json['origin']
-                        # logger.debug(f'origin ip is {origin_ip}')
-                    async with session.get(url, proxy=f'http://{proxy.string()}', timeout=TEST_TIMEOUT) as response:
-                        resp_json = await response.json()
-                        anonymous_ip = resp_json['origin']
-                        logger.debug(f'anonymous ip is {anonymous_ip}')
-                    assert origin_ip != anonymous_ip
-                    assert proxy.host == anonymous_ip
-                async with session.get(TEST_URL, proxy=f'http://{proxy.string()}', timeout=TEST_TIMEOUT,
-                                       allow_redirects=False) as response:
-                    if response.status in TEST_VALID_STATUS:
-                        if TEST_DONT_SET_MAX_SCORE:
-                            logger.debug(
-                                f'proxy {proxy.string()} is valid, remain current score')
-                        else:
-                            self.redis.max(proxy)
-                            logger.debug(
-                                f'proxy {proxy.string()} is valid, set max score')
+        try:
+            logger.debug(f'testing {proxy.string()}')
+            # if TEST_ANONYMOUS is True, make sure that
+            # the proxy has the effect of hiding the real IP
+            # logger.debug(f'TEST_ANONYMOUS {TEST_ANONYMOUS}')
+            if TEST_ANONYMOUS:
+                url = 'https://httpbin.org/ip'
+                async with session.get(url, timeout=TEST_TIMEOUT) as response:
+                    resp_json = await response.json()
+                    origin_ip = resp_json['origin']
+                    # logger.debug(f'origin ip is {origin_ip}')
+                async with session.get(url, proxy=f'http://{proxy.string()}', timeout=TEST_TIMEOUT) as response:
+                    resp_json = await response.json()
+                    anonymous_ip = resp_json['origin']
+                    logger.debug(f'anonymous ip is {anonymous_ip}')
+                assert origin_ip != anonymous_ip
+                assert proxy.host == anonymous_ip
+            async with session.get(TEST_URL, proxy=f'http://{proxy.string()}', timeout=TEST_TIMEOUT,
+                                   allow_redirects=False) as response:
+                if response.status in TEST_VALID_STATUS:
+                    if TEST_DONT_SET_MAX_SCORE:
+                        logger.debug(
+                            f'proxy {proxy.string()} is valid, remain current score')
                     else:
-                        self.redis.decrease(proxy)
+                        self.redis.max(proxy)
                         logger.debug(
-                            f'proxy {proxy.string()} is invalid, decrease score')
-                # if independent tester class found, create new set of storage and do the extra test
-                for tester in self.testers:
-                    key = tester.key
-                    if self.redis.exists(proxy, key):
-                        test_url = tester.test_url
-                        headers = tester.headers()
-                        cookies = tester.cookies()
-                        async with session.get(test_url, proxy=f'http://{proxy.string()}',
-                                               timeout=TEST_TIMEOUT,
-                                               headers=headers,
-                                               cookies=cookies,
-                                               allow_redirects=False) as response:
-                            resp_text = await response.text()
-                            is_valid = await tester.parse(resp_text, test_url, proxy.string())
-                            if is_valid:
-                                if tester.test_dont_set_max_score:
-                                    logger.info(
-                                        f'key[{key}] proxy {proxy.string()} is valid, remain current score')
-                                else:
-                                    self.redis.max(
-                                        proxy, key, tester.proxy_score_max)
-                                    logger.info(
-                                        f'key[{key}] proxy {proxy.string()} is valid, set max score')
+                            f'proxy {proxy.string()} is valid, set max score')
+                else:
+                    self.redis.decrease(proxy)
+                    logger.debug(
+                        f'proxy {proxy.string()} is invalid, decrease score')
+            # if independent tester class found, create new set of storage and do the extra test
+            for tester in self.testers:
+                key = tester.key
+                if self.redis.exists(proxy, key):
+                    test_url = tester.test_url
+                    headers = tester.headers()
+                    cookies = tester.cookies()
+                    async with session.get(test_url, proxy=f'http://{proxy.string()}',
+                                           timeout=TEST_TIMEOUT,
+                                           headers=headers,
+                                           cookies=cookies,
+                                           allow_redirects=False) as response:
+                        resp_text = await response.text()
+                        is_valid = await tester.parse(resp_text, test_url, proxy.string())
+                        if is_valid:
+                            if tester.test_dont_set_max_score:
+                                logger.info(
+                                    f'key[{key}] proxy {proxy.string()} is valid, remain current score')
                             else:
-                                self.redis.decrease(
-                                    proxy, tester.key, tester.proxy_score_min)
+                                self.redis.max(
+                                    proxy, key, tester.proxy_score_max)
                                 logger.info(
-                                    f'key[{key}] proxy {proxy.string()} is invalid, decrease score')
+                                    f'key[{key}] proxy {proxy.string()} is valid, set max score')
+                        else:
+                            self.redis.decrease(
+                                proxy, tester.key, tester.proxy_score_min)
+                            logger.info(
+                                f'key[{key}] proxy {proxy.string()} is invalid, decrease score')
 
-            except EXCEPTIONS:
-                self.redis.decrease(proxy)
-                [self.redis.decrease(proxy, tester.key, tester.proxy_score_min)
-                 for tester in self.testers]
+        except EXCEPTIONS:
+            self.redis.decrease(proxy)
+            [self.redis.decrease(proxy, tester.key, tester.proxy_score_min)
+             for tester in self.testers]
+            logger.debug(
+                f'proxy {proxy.string()} is invalid, decrease score')
+
+    async def run_tests(self):
+        """
+        test all proxies in batches, reusing a single aiohttp session
+        :return:
+        """
+        count = self.redis.count()
+        logger.debug(f'{count} proxies to test')
+        cursor = 0
+        connector = aiohttp.TCPConnector(ssl=False, limit=TEST_BATCH)
+        async with aiohttp.ClientSession(connector=connector) as session:
+            while True:
                 logger.debug(
-                    f'proxy {proxy.string()} is invalid, decrease score')
+                    f'testing proxies use cursor {cursor}, count {TEST_BATCH}')
+                cursor, proxies = self.redis.batch(cursor, count=TEST_BATCH)
+                if proxies:
+                    tasks = [self.test(proxy, session) for proxy in proxies]
+                    await asyncio.gather(*tasks, return_exceptions=True)
+                if not cursor:
+                    break
 
     @logger.catch
     def run(self):
@@ -116,26 +138,29 @@ def run(self):
         """
         # event loop of aiohttp
         logger.info('stating tester...')
-        count = self.redis.count()
-        logger.debug(f'{count} proxies to test')
-        cursor = 0
-        while True:
-            logger.debug(
-                f'testing proxies use cursor {cursor}, count {TEST_BATCH}')
-            cursor, proxies = self.redis.batch(cursor, count=TEST_BATCH)
-            if proxies:
-                tasks = [self.loop.create_task(
-                    self.test(proxy)) for proxy in proxies]
-                self.loop.run_until_complete(asyncio.wait(tasks))
-            if not cursor:
-                break
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            loop.run_until_complete(self.run_tests())
+        finally:
+            loop.close()
 
 
 def run_tester():
     host = '96.113.165.182'
     port = '3128'
-    tasks = [tester.test(Proxy(host=host, port=port))]
-    tester.loop.run_until_complete(asyncio.wait(tasks))
+    tester = Tester()
+
+    async def _test():
+        async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) as session:
+            await tester.test(Proxy(host=host, port=port), session)
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(_test())
+    finally:
+        loop.close()
 
 
 if __name__ == '__main__':

proxypool/scheduler.py

@@ -129,12 +129,15 @@ def run(self):
             tester_process and tester_process.join()
             getter_process and getter_process.join()
             server_process and server_process.join()
-            logger.info(
-                f'tester is {"alive" if tester_process.is_alive() else "dead"}')
-            logger.info(
-                f'getter is {"alive" if getter_process.is_alive() else "dead"}')
-            logger.info(
-                f'server is {"alive" if server_process.is_alive() else "dead"}')
+            if tester_process:
+                logger.info(
+                    f'tester is {"alive" if tester_process.is_alive() else "dead"}')
+            if getter_process:
+                logger.info(
+                    f'getter is {"alive" if getter_process.is_alive() else "dead"}')
+            if server_process:
+                logger.info(
+                    f'server is {"alive" if server_process.is_alive() else "dead"}')
             logger.info('proxy terminated')
 
 

proxypool/utils/proxy.py

@@ -63,7 +63,7 @@ def convert_proxy_or_proxies(data):
         if is_auth_proxy(data):
             host, port = extract_auth_proxy(data)
         else:
-            host, port = data.split(':')
+            host, port, *_ = data.split(':')
         return Proxy(host=host, port=int(port))