tester for testing proxies in queue
| 24 | |
| 25 | |
| 26 | class Tester(object): |
| 27 | """ |
| 28 | tester for testing proxies in queue |
| 29 | """ |
| 30 | |
| 31 | def __init__(self): |
| 32 | """ |
| 33 | init redis |
| 34 | """ |
| 35 | self.redis = RedisClient() |
| 36 | self.testers_cls = testers_cls |
| 37 | self.testers = [tester_cls() for tester_cls in self.testers_cls] |
| 38 | |
| 39 | async def test(self, proxy: Proxy, session: aiohttp.ClientSession): |
| 40 | """ |
| 41 | test single proxy |
| 42 | :param proxy: Proxy object |
| 43 | :param session: shared aiohttp session |
| 44 | :return: |
| 45 | """ |
| 46 | try: |
| 47 | logger.debug(f'testing {proxy.string()}') |
| 48 | # if TEST_ANONYMOUS is True, make sure that |
| 49 | # the proxy has the effect of hiding the real IP |
| 50 | # logger.debug(f'TEST_ANONYMOUS {TEST_ANONYMOUS}') |
| 51 | if TEST_ANONYMOUS: |
| 52 | url = TEST_ANONYMOUS_URL |
| 53 | async with session.get(url, timeout=TEST_TIMEOUT) as response: |
| 54 | resp_json = await response.json() |
| 55 | origin_ip = resp_json['origin'] |
| 56 | # logger.debug(f'origin ip is {origin_ip}') |
| 57 | async with session.get(url, proxy=f'http://{proxy.string()}', timeout=TEST_TIMEOUT) as response: |
| 58 | resp_json = await response.json() |
| 59 | anonymous_ip = resp_json['origin'] |
| 60 | logger.debug(f'anonymous ip is {anonymous_ip}') |
| 61 | assert origin_ip != anonymous_ip |
| 62 | assert proxy.host == anonymous_ip |
| 63 | async with session.get(TEST_URL, proxy=f'http://{proxy.string()}', timeout=TEST_TIMEOUT, |
| 64 | allow_redirects=False) as response: |
| 65 | if response.status in TEST_VALID_STATUS: |
| 66 | if TEST_DONT_SET_MAX_SCORE: |
| 67 | logger.debug( |
| 68 | f'proxy {proxy.string()} is valid, remain current score') |
| 69 | else: |
| 70 | self.redis.max(proxy) |
| 71 | logger.debug( |
| 72 | f'proxy {proxy.string()} is valid, set max score') |
| 73 | else: |
| 74 | self.redis.decrease(proxy) |
| 75 | logger.debug( |
| 76 | f'proxy {proxy.string()} is invalid, decrease score') |
| 77 | # if independent tester class found, create new set of storage and do the extra test |
| 78 | for tester in self.testers: |
| 79 | key = tester.key |
| 80 | if self.redis.exists(proxy, key): |
| 81 | test_url = tester.test_url |
| 82 | headers = tester.headers() |
| 83 | cookies = tester.cookies() |
no outgoing calls