# test_socks5_final.py """Тест SOCKS5 с правильным завершением.""" import asyncio import sys import os import time import random from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) def print_flush(*args, **kwargs): print(*args, **kwargs, flush=True) from managers.proxy_manager import ProxyManager from services.browser_service import BrowserService async def test(): print_flush("=" * 60) print_flush("🧦 ТЕСТ SOCKS5") print_flush("=" * 60) pm = ProxyManager() bs = BrowserService(pm) socks5 = pm.get_socks5_proxies() print_flush(f"\n📊 SOCKS5 прокси: {len(socks5)}") # 3 тестовых визита urls = [ "https://httpbin.org/ip", "https://httpbin.org/headers", "https://www.example.com", ] for i, url in enumerate(urls): print_flush(f"\n📍 Визит {i+1}/3: {url}") try: result = await bs.visit_page(url, reading_time=60) status = "✅" if result.success else "❌" print_flush(f" {status} Прокси: {result.proxy_info}") except Exception as e: print_flush(f" ❌ Ошибка: {e}") if i < 2: await asyncio.sleep(3) # Чистим с таймаутом print_flush(f"\n🧹 Очистка...") try: await asyncio.wait_for(bs.cleanup(), timeout=10) print_flush(" ✅ Очищено") except asyncio.TimeoutError: print_flush(" ⚠️ Таймаут очистки") print_flush(f"\n✅ Тест завершен!") return True if __name__ == "__main__": try: result = asyncio.run(test()) print_flush(f"\nРезультат: {result}") except KeyboardInterrupt: print_flush("\n⚠️ Прервано") except Exception as e: print_flush(f"\n❌ Ошибка: {e}") import traceback traceback.print_exc() finally: # Принудительный выход если зависло print_flush("Выход через 2 секунды...") time.sleep(2) print_flush("Готово!") os._exit(0) # Принудительный выход