loop-uh
9b2860b851
Some checks failed
Windows project source guards / test (push) Failing after 26s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
531 lines
23 KiB
Python
531 lines
23 KiB
Python
from __future__ import annotations
|
||
|
||
"""Сигнал сервера о новой версии: слушатель и его связка с окном."""
|
||
|
||
import unittest
|
||
from types import SimpleNamespace
|
||
from unittest.mock import Mock, patch
|
||
|
||
from xray_fluent.updates import release_watch
|
||
from xray_fluent.updates.release_watch import ReleaseWatcher
|
||
from zapret_update import screen
|
||
|
||
|
||
def _quiet(version: str = "0.8.19") -> dict:
|
||
return {"channel": "stable", "version": version, "changed": False}
|
||
|
||
|
||
def _news(version: str) -> dict:
|
||
return {"channel": "stable", "version": version, "changed": True}
|
||
|
||
|
||
class _Harness:
|
||
"""Сервер и часы слушателя: ответы идут по сценарию, паузы не ждут."""
|
||
|
||
def __init__(self, script: list, **options) -> None:
|
||
self.script = list(script)
|
||
self.urls: list[str] = []
|
||
self.pauses: list[float] = []
|
||
self.released: list[str] = []
|
||
self.queued: list[str] = []
|
||
self.now = 0.0
|
||
options.setdefault("is_busy", lambda: False)
|
||
self.watcher = ReleaseWatcher(
|
||
current_version="0.8.19",
|
||
on_release=self.released.append,
|
||
on_queued=self.queued.append,
|
||
ask=self._ask,
|
||
clock=lambda: self.now,
|
||
**options,
|
||
)
|
||
self.watcher._pause = self._pause
|
||
|
||
def _ask(self, url: str, timeout: float) -> dict:
|
||
self.urls.append(url)
|
||
if not self.script:
|
||
self.watcher.stop()
|
||
raise OSError("сценарий кончился")
|
||
step = self.script.pop(0)
|
||
took, answer = step if isinstance(step, tuple) else (0.2 if "hold=0" in url else 300.0, step)
|
||
self.now += took
|
||
if isinstance(answer, BaseException):
|
||
raise answer
|
||
return answer
|
||
|
||
def _pause(self, seconds: float) -> bool:
|
||
self.pauses.append(float(seconds))
|
||
return self.watcher._stop.is_set()
|
||
|
||
|
||
class ReleaseWatcherTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
patcher = patch.object(release_watch, "_log")
|
||
patcher.start()
|
||
self.addCleanup(patcher.stop)
|
||
|
||
def test_first_question_is_a_probe_then_questions_wait(self) -> None:
|
||
harness = _Harness([_quiet(), _quiet(), _quiet()])
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertIn("hold=0", harness.urls[0])
|
||
self.assertIn("known=0.8.19", harness.urls[0])
|
||
self.assertIn("channel=stable", harness.urls[0])
|
||
# Сразу после запуска вопрос короче, дальше — в полный срок.
|
||
self.assertIn(f"hold={release_watch.WARMUP_HOLD_S}", harness.urls[1])
|
||
self.assertNotIn("hold=", harness.urls[2])
|
||
self.assertIs(harness.watcher.reachable, False) # сценарий кончился ошибкой
|
||
|
||
def test_permission_is_reported_once_and_version_becomes_known(self) -> None:
|
||
harness = _Harness([_quiet(), _news("0.8.20"), _quiet("0.8.20")])
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertEqual(harness.released, ["0.8.20"])
|
||
self.assertIn("known=0.8.20", harness.urls[2])
|
||
|
||
def test_older_or_strange_version_is_not_a_permission(self) -> None:
|
||
harness = _Harness([_news("0.8.18"), _news("мусор"), _news("0.8.19")])
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertEqual(harness.released, [])
|
||
|
||
def test_queue_ticket_is_kept_and_named_in_next_questions(self) -> None:
|
||
queued = lambda ticket: {"version": "0.8.20", "changed": False, "queued": True, "ticket": ticket} # noqa: E731
|
||
harness = _Harness([queued(1791570000), queued(1791579999), _news("0.8.20"), _quiet("0.8.20")])
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertNotIn("ticket=", harness.urls[0])
|
||
# Талон — время первой постановки: программа держится за него.
|
||
self.assertIn("ticket=1791570000", harness.urls[1])
|
||
self.assertIn("ticket=1791570000", harness.urls[2])
|
||
self.assertNotIn("ticket=", harness.urls[3])
|
||
self.assertEqual(harness.queued, ["0.8.20"])
|
||
|
||
def test_answer_wait_for_your_stage_is_not_a_permission(self) -> None:
|
||
held = {"version": "0.8.20", "changed": False, "held": "stage"}
|
||
harness = _Harness([held, held, {"version": "0.8.20", "changed": False, "held": "halted"}, _news("0.8.20")])
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertEqual(harness.queued, [])
|
||
self.assertEqual(harness.released, ["0.8.20"])
|
||
self.assertNotIn(release_watch.QUICK_ANSWER_PAUSE_S, harness.pauses)
|
||
|
||
def test_busy_program_says_so_and_asks_shorter(self) -> None:
|
||
busy = [True, True, True, False]
|
||
harness = _Harness([_quiet()] * 4, is_busy=lambda: busy.pop(0) if busy else False)
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertIn("busy=1", harness.urls[0])
|
||
self.assertIn("busy=1", harness.urls[2])
|
||
self.assertIn(f"hold={release_watch.BUSY_HOLD_S}", harness.urls[2])
|
||
self.assertNotIn("busy=", harness.urls[3])
|
||
|
||
def test_program_tells_what_it_is_doing_and_nothing_else(self) -> None:
|
||
harness = _Harness([_quiet()], activity=lambda: {"act": "tray", "run": "1", "server": "secret.example"})
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertIn("act=tray", harness.urls[0])
|
||
self.assertIn("run=1", harness.urls[0])
|
||
self.assertNotIn("secret", harness.urls[0])
|
||
|
||
def test_update_outcome_is_reported_until_the_server_hears_it(self) -> None:
|
||
pending = [{"prev": "0.8.18", "took": "24"}]
|
||
harness = _Harness(
|
||
[OSError("нет сети"), _quiet(), _quiet()],
|
||
pending_report=lambda: dict(pending[0]) if pending else {},
|
||
report_delivered=lambda _report: pending.clear(),
|
||
)
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertIn("prev=0.8.18", harness.urls[0])
|
||
self.assertIn("took=24", harness.urls[1])
|
||
self.assertNotIn("prev=", harness.urls[2])
|
||
|
||
def test_pauses_grow_when_the_server_is_away_and_program_lives_on_its_own(self) -> None:
|
||
harness = _Harness([OSError("нет"), OSError("нет"), OSError("нет")])
|
||
|
||
harness.watcher.run()
|
||
|
||
# Паузы растут, но не дольше трёх минут: вернулась сеть — программа узнаёт быстро.
|
||
self.assertEqual(harness.pauses[:3], [60.0, 120.0, 180.0])
|
||
self.assertIs(harness.watcher.reachable, False)
|
||
|
||
def test_quick_empty_answer_is_followed_by_a_pause(self) -> None:
|
||
harness = _Harness([_quiet(), (0.3, _quiet())])
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertIn(release_watch.QUICK_ANSWER_PAUSE_S, harness.pauses)
|
||
|
||
def test_switched_off_updates_keep_no_connection(self) -> None:
|
||
enabled = [False, False]
|
||
harness = _Harness([_quiet()], is_enabled=lambda: enabled.pop(0) if enabled else True)
|
||
|
||
harness.watcher.run()
|
||
|
||
self.assertEqual(harness.pauses[:2], [release_watch.DISABLED_RECHECK_S] * 2)
|
||
self.assertEqual(len(harness.urls), 2)
|
||
|
||
|
||
class ScreenBusyTests(unittest.TestCase):
|
||
"""«Экран занят» — по двум независимым признакам, а не по одному ответу Windows."""
|
||
|
||
def _busy(self, state: int, foreground: str) -> bool:
|
||
# Определение экрана — общее ядро zapret_update.screen; память об игре
|
||
# каждый случай начинает пустой.
|
||
with patch.object(screen, "_last_fullscreen", 0.0), patch.object(
|
||
screen, "_notification_state", return_value=state
|
||
), patch.object(screen, "_foreground", return_value=foreground):
|
||
return release_watch.fullscreen_app_active()
|
||
|
||
def test_game_and_presentation_mean_busy_whatever_is_in_front(self) -> None:
|
||
self.assertTrue(self._busy(3, release_watch.FOREGROUND_SHELL))
|
||
self.assertTrue(self._busy(4, release_watch.FOREGROUND_PART))
|
||
|
||
def test_vague_answer_needs_a_real_fullscreen_window_in_front(self) -> None:
|
||
self.assertTrue(self._busy(2, release_watch.FOREGROUND_FULL))
|
||
for foreground in (release_watch.FOREGROUND_PART, release_watch.FOREGROUND_SHELL, release_watch.FOREGROUND_OWN, ""):
|
||
self.assertFalse(self._busy(2, foreground), foreground)
|
||
|
||
def test_ordinary_work_is_not_busy(self) -> None:
|
||
self.assertFalse(self._busy(5, release_watch.FOREGROUND_FULL))
|
||
self.assertFalse(self._busy(0, ""))
|
||
|
||
def test_raw_state_is_told_as_code_and_foreground(self) -> None:
|
||
with patch.object(screen, "_notification_state", return_value=2), patch.object(
|
||
screen, "_foreground", return_value="p"
|
||
):
|
||
self.assertEqual(release_watch.screen_state(), "2p")
|
||
|
||
def test_borderless_game_and_alt_tab_keep_the_update_waiting(self) -> None:
|
||
# Игра «окно без рамки»: Windows говорит «можно», выдаёт само окно.
|
||
self.assertTrue(self._busy(5, release_watch.FOREGROUND_BARE))
|
||
with patch.object(screen, "_last_fullscreen", 0.0):
|
||
with patch.object(screen, "_notification_state", return_value=5), patch.object(
|
||
screen, "_foreground", return_value=release_watch.FOREGROUND_BARE
|
||
):
|
||
self.assertTrue(screen.fullscreen_active(now=1000.0))
|
||
# Свернул игру, ушёл в Discord — обновление всё равно ждёт.
|
||
with patch.object(screen, "_notification_state", return_value=5), patch.object(
|
||
screen, "_foreground", return_value=release_watch.FOREGROUND_PART
|
||
):
|
||
self.assertTrue(screen.fullscreen_active(now=1300.0))
|
||
self.assertFalse(screen.fullscreen_active(now=1000.0 + screen.GAME_MEMORY_SECONDS + 1))
|
||
|
||
|
||
class UpdateSignalTests(unittest.TestCase):
|
||
def _signal(self, **options):
|
||
from xray_fluent.updates.update_signal import UpdateSignal
|
||
|
||
self.watcher = SimpleNamespace(reachable=None, start=Mock(), stop=Mock())
|
||
captured: dict = {}
|
||
|
||
def factory(**kwargs):
|
||
captured.update(kwargs)
|
||
return self.watcher
|
||
|
||
signal = UpdateSignal(
|
||
is_enabled=lambda: True,
|
||
window_shown=options.get("window_shown", lambda: True),
|
||
connected=options.get("connected", lambda: False),
|
||
watcher_factory=factory,
|
||
)
|
||
return signal, captured
|
||
|
||
def test_permission_is_remembered_with_its_moment(self) -> None:
|
||
signal, hooks = self._signal()
|
||
heard: list[str] = []
|
||
signal.granted.connect(heard.append)
|
||
|
||
self.assertFalse(signal.is_granted("0.8.20"))
|
||
hooks["on_release"]("0.8.20")
|
||
|
||
self.assertTrue(signal.is_granted("0.8.20"))
|
||
self.assertGreater(signal.granted_at("0.8.20"), 0)
|
||
self.assertEqual(signal.granted_at("0.8.21"), 0.0)
|
||
self.assertEqual(heard, ["0.8.20"])
|
||
|
||
def test_lost_download_is_told_to_the_server_without_hiding_an_older_report(self) -> None:
|
||
signal, hooks = self._signal()
|
||
|
||
signal.note_miss("0.8.22", "hash")
|
||
self.assertEqual(hooks["pending_report"](), {"miss": "0.8.22", "why": "hash"})
|
||
hooks["report_delivered"]({})
|
||
self.assertEqual(hooks["pending_report"](), {})
|
||
|
||
# Неотправленное «не вышло» прошлой установки важнее.
|
||
signal.set_report({"fail": "0.8.21", "why": "crash"})
|
||
signal.note_miss("0.8.22", "net")
|
||
self.assertEqual(hooks["pending_report"](), {"fail": "0.8.21", "why": "crash"})
|
||
|
||
hooks["report_delivered"]({})
|
||
signal.note_miss("0.8.22", "../x")
|
||
self.assertEqual(hooks["pending_report"](), {})
|
||
|
||
def test_download_failure_codes(self) -> None:
|
||
import errno
|
||
import zipfile
|
||
|
||
from xray_fluent.updates.app_updater import DownloadFailed, UpdateRejected, download_failure_code
|
||
|
||
self.assertEqual(download_failure_code(UpdateRejected("сумма", "hash")), "hash")
|
||
self.assertEqual(download_failure_code(zipfile.BadZipFile()), "badzip")
|
||
self.assertEqual(download_failure_code(OSError(errno.ENOSPC, "места нет")), "disk")
|
||
self.assertEqual(download_failure_code(DownloadFailed("нет сети")), "net")
|
||
self.assertEqual(download_failure_code(ValueError()), "other")
|
||
|
||
def test_queue_counts_only_when_the_server_answered(self) -> None:
|
||
signal, _hooks = self._signal()
|
||
|
||
self.assertFalse(signal.queue_reachable()) # ещё не знаем — живём как раньше
|
||
self.watcher.reachable = True
|
||
self.assertTrue(signal.queue_reachable())
|
||
|
||
def test_activity_names_window_tray_game_and_connection(self) -> None:
|
||
from xray_fluent.updates import update_signal
|
||
|
||
shown = [True]
|
||
signal, hooks = self._signal(window_shown=lambda: shown[0], connected=lambda: True)
|
||
with patch.object(update_signal, "fullscreen_app_active", return_value=False), patch.object(
|
||
update_signal, "screen_state", return_value="5p"
|
||
):
|
||
self.assertEqual(hooks["activity"](), {}) # окно о себе ещё не сказало
|
||
signal.refresh_presence()
|
||
self.assertEqual(hooks["activity"](), {"act": "window", "run": "1", "scr": "5p"})
|
||
shown[0] = False
|
||
signal.refresh_presence()
|
||
self.assertEqual(hooks["activity"](), {"act": "tray", "run": "1", "scr": "5p"})
|
||
with patch.object(update_signal, "fullscreen_app_active", return_value=True):
|
||
self.assertEqual(hooks["activity"]()["act"], "fullscreen")
|
||
self.assertTrue(signal.busy_reason())
|
||
|
||
def test_report_is_given_once(self) -> None:
|
||
signal, hooks = self._signal()
|
||
signal.set_report({"prev": "0.8.18", "took": "30"})
|
||
|
||
self.assertEqual(hooks["pending_report"](), {"prev": "0.8.18", "took": "30"})
|
||
hooks["report_delivered"]({"prev": "0.8.18"})
|
||
|
||
self.assertEqual(hooks["pending_report"](), {})
|
||
|
||
|
||
class BackgroundInstallGateTests(unittest.TestCase):
|
||
"""Фоновая установка ждёт разрешения сервера и простоя."""
|
||
|
||
def _reason(self, *, reachable: bool, granted: bool, busy: str = "", urgent: bool = False) -> str:
|
||
from xray_fluent.ui import main_window
|
||
from xray_fluent.ui.main_window import MainWindow
|
||
|
||
signal = SimpleNamespace(
|
||
queue_reachable=lambda: reachable,
|
||
is_granted=lambda _version: granted,
|
||
is_urgent=lambda _version: urgent,
|
||
busy_reason=lambda: busy,
|
||
)
|
||
window = SimpleNamespace(_update_signal=signal, _recheck_update_when_free=Mock())
|
||
with patch.object(main_window.QTimer, "singleShot") as self.single_shot:
|
||
return MainWindow._update_wait_reason(window, SimpleNamespace(version="0.8.20"))
|
||
|
||
def test_version_waits_for_the_server_permission(self) -> None:
|
||
self.assertIn("разрешения сервера", self._reason(reachable=True, granted=False))
|
||
self.assertEqual(self._reason(reachable=True, granted=True), "")
|
||
|
||
def test_without_a_queue_server_program_installs_as_before(self) -> None:
|
||
self.assertEqual(self._reason(reachable=False, granted=False), "")
|
||
|
||
def test_game_on_screen_postpones_the_update_and_plans_a_recheck(self) -> None:
|
||
reason = self._reason(reachable=True, granted=True, busy="на экране игра или видео на весь экран")
|
||
|
||
self.assertIn("игра", reason)
|
||
self.single_shot.assert_called_once()
|
||
|
||
def test_while_in_game_only_the_screen_is_watched_then_the_release_is_checked(self) -> None:
|
||
from xray_fluent.ui import main_window
|
||
from xray_fluent.ui.main_window import MainWindow
|
||
|
||
busy = ["на экране игра", ""]
|
||
window = SimpleNamespace(
|
||
_quitting=False,
|
||
_busy_recheck_pending=True,
|
||
controller=SimpleNamespace(state=SimpleNamespace(settings=SimpleNamespace(check_updates=True))),
|
||
_update_signal=SimpleNamespace(busy_reason=lambda: busy.pop(0)),
|
||
_check_updates=Mock(),
|
||
)
|
||
window._recheck_update_when_free = lambda: MainWindow._recheck_update_when_free(window)
|
||
with patch.object(main_window.QTimer, "singleShot") as single_shot:
|
||
MainWindow._recheck_update_when_free(window)
|
||
# Всё ещё в игре: через пять секунд взглянуть снова, выпуск не трогать.
|
||
window._check_updates.assert_not_called()
|
||
self.assertEqual(single_shot.call_args.args[0], main_window.UPDATE_BUSY_RECHECK_MS)
|
||
single_shot.call_args.args[1]()
|
||
# Освободился: теперь проверка выпуска и установка.
|
||
window._check_updates.assert_called_once_with(silent=True)
|
||
self.assertFalse(window._busy_recheck_pending)
|
||
|
||
def test_revoked_version_is_replaced_even_during_a_game(self) -> None:
|
||
reason = self._reason(reachable=True, granted=True, busy="на экране игра", urgent=True)
|
||
|
||
self.assertEqual(reason, "")
|
||
self.single_shot.assert_not_called()
|
||
|
||
def test_no_listener_means_no_extra_waiting(self) -> None:
|
||
from xray_fluent.ui.main_window import MainWindow
|
||
|
||
self.assertEqual(MainWindow._update_wait_reason(SimpleNamespace(_update_signal=None), SimpleNamespace(version="1")), "")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|
||
|
||
class RevokedVersionTests(unittest.TestCase):
|
||
"""Выключатель версии: отозванная сервером версия ставится сразу, мимо «занят»."""
|
||
|
||
def test_revoked_answer_takes_the_urgent_path(self) -> None:
|
||
released, revoked = [], []
|
||
watcher = ReleaseWatcher(current_version="0.8.25", on_release=released.append, on_revoked=revoked.append)
|
||
|
||
self.assertTrue(watcher._accept({"channel": "stable", "version": "0.8.26", "changed": True, "revoked": True}))
|
||
|
||
self.assertEqual((released, revoked), ([], ["0.8.26"]))
|
||
|
||
def test_urgent_version_skips_the_busy_wait(self) -> None:
|
||
from xray_fluent.updates.update_signal import UpdateSignal
|
||
|
||
captured = {}
|
||
|
||
def factory(**kwargs):
|
||
captured.update(kwargs)
|
||
return SimpleNamespace(reachable=True, start=Mock(), stop=Mock())
|
||
|
||
signal = UpdateSignal(is_enabled=lambda: True, window_shown=lambda: True, connected=lambda: False, watcher_factory=factory)
|
||
captured["on_revoked"]("0.8.26")
|
||
|
||
self.assertTrue(signal.is_urgent("0.8.26"))
|
||
self.assertTrue(signal.is_granted("0.8.26"))
|
||
self.assertFalse(signal.is_urgent("0.8.27"))
|
||
|
||
|
||
class AskServerTransportTests(unittest.TestCase):
|
||
"""Отключено — напрямую стеком Chromium; подключено — urllib внутри туннеля."""
|
||
|
||
def setUp(self) -> None:
|
||
# Имя сервера выбирается по сети; здесь проверяется только маршрут запроса.
|
||
from unittest.mock import patch
|
||
from xray_fluent.updates import site_names
|
||
|
||
patcher = patch.object(site_names, "resolve", side_effect=lambda url: url)
|
||
patcher.start()
|
||
self.addCleanup(patcher.stop)
|
||
|
||
def _service(self, *, status=200, body=b'{"version": "0.9.0"}'):
|
||
from xray_fluent.network import browser_fetch
|
||
|
||
calls = []
|
||
|
||
class _Service:
|
||
def available(self):
|
||
return True
|
||
|
||
def fetch(self, url, *, headers, timeout, max_bytes):
|
||
calls.append((url, dict(headers)))
|
||
return browser_fetch.BrowserResponse(status=status, body=body)
|
||
|
||
return _Service(), calls
|
||
|
||
def test_disconnected_poll_goes_through_the_browser_stack(self) -> None:
|
||
from unittest.mock import patch
|
||
from xray_fluent.network import browser_fetch
|
||
from xray_fluent.updates import release_watch
|
||
|
||
service, calls = self._service()
|
||
with patch.object(browser_fetch, "SERVICE", service), patch.object(release_watch, "urlopen") as legacy:
|
||
answer = release_watch._ask_server(release_watch.WAIT_URL + "?known=0.8.38&run=0", 5)
|
||
legacy.assert_not_called()
|
||
self.assertEqual(answer, {"version": "0.9.0"})
|
||
self.assertEqual(calls[0][1]["accept"], "application/json")
|
||
|
||
def test_connected_poll_stays_inside_the_tunnel(self) -> None:
|
||
import io
|
||
from unittest.mock import patch
|
||
from xray_fluent.network import browser_fetch
|
||
from xray_fluent.updates import release_watch
|
||
|
||
service, calls = self._service()
|
||
|
||
class _Response(io.BytesIO):
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *exc):
|
||
return False
|
||
|
||
with patch.object(browser_fetch, "SERVICE", service), \
|
||
patch.object(release_watch, "urlopen", return_value=_Response(b'{"version": "0.9.1"}')) as legacy:
|
||
answer = release_watch._ask_server(release_watch.WAIT_URL + "?known=0.8.38&run=1", 5)
|
||
self.assertEqual(calls, [])
|
||
legacy.assert_called_once()
|
||
self.assertEqual(answer, {"version": "0.9.1"})
|
||
|
||
def test_server_error_is_a_failure_not_an_answer(self) -> None:
|
||
from unittest.mock import patch
|
||
from xray_fluent.network import browser_fetch
|
||
from xray_fluent.updates import release_watch
|
||
|
||
service, _calls = self._service(status=502, body=b"")
|
||
with patch.object(browser_fetch, "SERVICE", service):
|
||
with self.assertRaises(OSError):
|
||
release_watch._ask_server(release_watch.WAIT_URL + "?run=0", 5)
|
||
|
||
|
||
class AskServerNameTests(unittest.TestCase):
|
||
"""Очередь обновлений спрашивается под открытым именем сервера, а не под режущимся основным."""
|
||
|
||
def test_poll_goes_to_the_open_name(self) -> None:
|
||
import io
|
||
from unittest.mock import patch
|
||
from xray_fluent.network import browser_fetch
|
||
from xray_fluent.updates import release_watch, site_names
|
||
|
||
asked: list[str] = []
|
||
|
||
class _Response(io.BytesIO):
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *exc):
|
||
return False
|
||
|
||
def fake_urlopen(request, timeout):
|
||
asked.append(request.full_url)
|
||
return _Response(b'{"changed": false}')
|
||
|
||
with patch.object(browser_fetch.SERVICE, "available", return_value=False), \
|
||
patch.object(site_names._NAMES, "choose", return_value="git2.zapret.moe"), \
|
||
patch.object(release_watch, "urlopen", side_effect=fake_urlopen):
|
||
release_watch._ask_server(release_watch.WAIT_URL + "?known=0.8.38&run=1", 5)
|
||
|
||
self.assertEqual(asked, ["https://git2.zapret.moe/api/zapret/kvn/wait?known=0.8.38&run=1"])
|
||
|
||
def test_foreign_address_is_not_rewritten_and_asks_nothing(self) -> None:
|
||
from unittest.mock import patch
|
||
from xray_fluent.updates import site_names
|
||
|
||
with patch.object(site_names._NAMES, "choose", side_effect=AssertionError("сеть не нужна")):
|
||
for url in ("https://example.com/x", "http://git.zapret.moe/x", "https://git2.zapret.moe:8443/x"):
|
||
self.assertEqual(site_names.resolve(url), url)
|
||
|
||
def test_release_client_shares_its_working_name_with_the_queue(self) -> None:
|
||
from xray_fluent.updates import site_names
|
||
|
||
site_names._NAMES.forget()
|
||
site_names.remember("git3.zapret.moe")
|
||
self.assertEqual(site_names.resolve("https://git2.zapret.moe/a"), "https://git3.zapret.moe/a")
|
||
site_names._NAMES.forget()
|