loop-uh
54ac4b0251
Some checks failed
Windows project source guards / test (push) Has been cancelled
Требование незаметности: пачка соединений к разным серверам или ровный ритм проверок выдаёт провайдеру (ТСПУ) весь список подписки. - network/probe_gate.py — одно место с правилами: пауза прохода «все» 10–20 с со случайным разбросом, не больше 3 автоматических проверок других серверов в скользящий час, журнал всех проверок для тестов. - «Пинг всех» — одно соединение за раз (было до 8) и пауза шлюза между серверами; отмена будит паузу, одиночный пинг после прохода не ждёт. - «Скорость всех» — пауза 10–20 с между серверами вместо 0,4–1,6 с. - Умная замена мерит одного кандидата (было до 3 прямых рукопожатий с другими серверами каждые ~5 мин) и только в бюджете шлюза. - Тест-список сетевых вызовов: create_connection/getaddrinfo/urlopen и т. п. в новом модуле роняют тест, пока не решено, по какому правилу шлюза идёт проверка. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
41 lines
1.6 KiB
Python
41 lines
1.6 KiB
Python
"""Шлюз проверок (xray_fluent/network/probe_gate.py): бюджет и журнал."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import random
|
|
import unittest
|
|
|
|
from xray_fluent.network import probe_gate
|
|
|
|
|
|
class ProbeGateTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.now = 1000.0
|
|
self.gate = probe_gate.ProbeGate(clock=lambda: self.now)
|
|
|
|
def test_candidate_budget_is_a_sliding_hour(self) -> None:
|
|
for _ in range(probe_gate.CANDIDATE_PROBES_PER_HOUR):
|
|
self.assertTrue(self.gate.try_candidate("a.example"))
|
|
self.assertFalse(self.gate.try_candidate("b.example"))
|
|
self.now += 3599.0
|
|
self.assertFalse(self.gate.try_candidate("b.example"))
|
|
self.now += 2.0
|
|
self.assertTrue(self.gate.try_candidate("b.example"))
|
|
|
|
def test_refused_candidate_is_not_journaled(self) -> None:
|
|
for _ in range(probe_gate.CANDIDATE_PROBES_PER_HOUR + 2):
|
|
self.gate.try_candidate("a.example")
|
|
kinds = [record.kind for record in self.gate.journal()]
|
|
self.assertEqual(kinds, [probe_gate.CANDIDATE] * probe_gate.CANDIDATE_PROBES_PER_HOUR)
|
|
|
|
def test_sweep_pause_stays_in_range(self) -> None:
|
|
rng = random.Random(7)
|
|
low, high = probe_gate.SWEEP_PAUSE_SEC
|
|
self.assertGreaterEqual(low, 10.0)
|
|
pauses = [probe_gate.sweep_pause(rng) for _ in range(200)]
|
|
self.assertTrue(all(low <= pause <= high for pause in pauses))
|
|
self.assertGreater(max(pauses) - min(pauses), 5.0) # разброс, не ровный ритм
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|