"""DECODO 프록시 제공자 테스트 (포트 기반 sticky, 순수·네트워크 불필요).""" from config.config_models import DecodoConfig from services.search.proxy import DecodoProxy def _p(**kw): base = dict(host="gate.decodo.com", username="user1", password="pw", port_start=10001, port_end=10010, session_minutes=10) base.update(kw) return DecodoProxy(DecodoConfig(**base)) def test_disabled_when_credentials_missing(): assert DecodoProxy(DecodoConfig()).enabled is False # 전부 비어있음 assert _p(password="").enabled is False assert _p(port_end=0).enabled is False # 포트 미설정 assert _p().enabled is True def test_playwright_proxy_shape(): cfg = _p().playwright_proxy() assert cfg["username"] == "user1" and cfg["password"] == "pw" # 고정 자격증명 assert cfg["server"].startswith("http://gate.decodo.com:") port = int(cfg["server"].rsplit(":", 1)[1]) assert 10001 <= port <= 10010 # 포트 범위 안 def test_disabled_returns_none(): assert DecodoProxy(DecodoConfig()).playwright_proxy() is None def test_port_selected_within_range_and_stable_in_window(): p = _p() port = p._port() assert 10001 <= port <= 10010 assert p._port() == port # 같은 시간창에서는 동일 포트(동일 sticky IP) def test_single_port_range(): p = _p(port_start=10001, port_end=10001) assert p._port() == 10001 # 포트 1개면 항상 그 포트 def test_rotate_advances_port_immediately(): p = _p(port_start=10001, port_end=10003) # 포트 3개 before = p._port() p.rotate() after = p._port() assert after != before # 즉시 다음 포트(새 IP) assert 10001 <= after <= 10003 p.rotate(); p.rotate() # 3번 회전하면 한 바퀴 → 원위치 assert p._port() == before def test_burned_port_is_skipped(): p = _p(port_start=10001, port_end=10003) burned = p._port() p.mark_burned(burned) assert p._port() != burned # 쿨다운 중인 포트는 건너뜀 assert p.available_ports() == 2 def test_burned_port_returns_after_cooldown(): p = _p(port_start=10001, port_end=10003) burned = p._port() p.mark_burned(burned, cooldown_sec=0) # 즉시 만료 assert p._port() == burned # 만료 후엔 다시 사용 가능 assert p.available_ports() == 3 def test_all_burned_falls_back_to_earliest_expiry(): p = _p(port_start=10001, port_end=10002) # 포트 2개 first, second = 10001, 10002 p.mark_burned(first, cooldown_sec=10) # 먼저 만료 p.mark_burned(second, cooldown_sec=999) assert p._port() == first # 전 포트 쿨다운 — 만료 임박 포트 사용(가용성 우선) def test_mark_burned_ignores_none_port(): p = _p() p.mark_burned(None) # 프록시 미사용 세션 — no-op assert p.available_ports() == 10 def test_cooldown_default_at_least_30min(): assert _p(session_minutes=10).cooldown_sec == 1800 # max(sticky 600, 1800) assert _p(session_minutes=60).cooldown_sec == 3600 # sticky 가 더 길면 sticky 만큼