zapretgui/tests/test_profile_setup_write_serialization.py
loop-uh 0d1299722c
All checks were successful
Zapret source guards / validate (push) Successful in 13s
Windows-сборка / Dev Dev 21.1.7.27 опубликован
Оценка стратегии — в меню по правой кнопке вместо кнопок под списком
Полоса с кнопками «Работает», «Не работает», «В избранное» под списком
готовых стратегий убрана: список занимает вкладку целиком. Оценку и
избранное ставят из меню стратегии (правая кнопка мыши или клавиша меню),
причём любой стратегии списка, а не только выбранной для профиля.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-07 15:49:00 +03:00

543 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
from profile.ui.profile_setup_page import ProfileSetupPageBase
class _Signal:
def connect(self, _callback) -> None:
return None
class _Worker:
saved = _Signal()
failed = _Signal()
finished = _Signal()
def __init__(self) -> None:
self.start = Mock()
self.deleteLater = Mock()
class _Runtime:
def __init__(self, *, running: bool = False) -> None:
self.running = running
def is_running(self) -> bool:
return self.running
def start_qthread_worker(self, *, worker_factory, **_kwargs):
worker = worker_factory(0)
worker.start()
return 0, worker
class ProfileSetupWriteSerializationTests(unittest.TestCase):
def test_profile_setup_write_queue_lives_in_queued_worker_state(self) -> None:
import inspect
from ui.queued_worker_state import QueuedWorkerState
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
module_source = inspect.getsource(
__import__("profile.ui.profile_setup_page", fromlist=[""])
)
init_source = inspect.getsource(ProfileSetupPageBase.__init__)
from profile.ui.profile_setup_save_controllers import ProfileSetupSaveController
queue_source = inspect.getsource(ProfileSetupSaveController._queue_profile_setup_write_operation)
start_next_source = inspect.getsource(ProfileSetupSaveController._start_next_profile_setup_write_operation)
schedule_source = inspect.getsource(ProfileSetupSaveController._schedule_profile_setup_write_operation_start)
cleanup_source = inspect.getsource(ProfileSetupPageBase.cleanup)
self.assertIsInstance(ProfileSetupPageBase._profile_setup_write_state_obj(page), QueuedWorkerState)
self.assertIn("from ui.queued_worker_state import QueuedWorkerState", module_source)
self.assertIn("_profile_setup_write_state = QueuedWorkerState", init_source)
self.assertIn("_profile_setup_write_state_obj()", queue_source)
self.assertIn("state.pop_next()", start_next_source)
self.assertIn("state.start_scheduled", schedule_source)
self.assertIn("_profile_setup_write_state_obj().reset()", cleanup_source)
self.assertNotIn("self._pending_profile_setup_write_operations: list", init_source)
def test_user_profile_write_queue_lives_in_queued_worker_state(self) -> None:
import inspect
from ui.queued_worker_state import QueuedWorkerState
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
module_source = inspect.getsource(
__import__("profile.ui.profile_setup_page", fromlist=[""])
)
init_source = inspect.getsource(ProfileSetupPageBase.__init__)
from profile.ui.profile_user_profile_controller import ProfileUserProfileController
queue_source = inspect.getsource(ProfileUserProfileController._queue_user_profile_write_operation)
pop_source = inspect.getsource(ProfileUserProfileController._pop_next_pending_user_profile_write_operation)
has_pending_source = inspect.getsource(ProfileUserProfileController._has_pending_user_profile_write_operation)
schedule_source = inspect.getsource(ProfileUserProfileController._schedule_next_pending_user_profile_write_operation_start)
cleanup_source = inspect.getsource(ProfileSetupPageBase.cleanup)
self.assertIsInstance(ProfileSetupPageBase._user_profile_write_state_obj(page), QueuedWorkerState)
self.assertIn("from ui.queued_worker_state import QueuedWorkerState", module_source)
self.assertIn("_user_profile_write_state = QueuedWorkerState", init_source)
self.assertIn("_user_profile_write_state_obj()", queue_source)
self.assertIn("state.pop_next()", pop_source)
self.assertIn("_user_profile_write_state_obj().has_pending()", has_pending_source)
self.assertIn("state.start_scheduled", schedule_source)
self.assertIn("_user_profile_write_state_obj().reset()", cleanup_source)
def test_raw_profile_save_waits_while_settings_save_runs(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._settings_save_runtime = _Runtime(running=True)
page._raw_profile_save_runtime = _Runtime(running=False)
page._raw_profile_save_request_id = 0
page._pending_settings_save = None
page._pending_raw_profile_save = None
page._pending_profile_setup_write_operations = []
page._profile_key = "profile-1"
page._raw_profile_save_button = None
page.create_profile_raw_text_save_worker = Mock(return_value=_Worker())
ProfileSetupPageBase._request_raw_profile_save(page, "profile-1", "--new\n--lua-desync=split")
page.create_profile_raw_text_save_worker.assert_not_called()
self.assertEqual(
page._pending_profile_setup_write_operations,
[
{
"kind": "raw_profile_save",
"profile_key": "profile-1",
"text": "--new\n--lua-desync=split",
}
],
)
page._settings_save_runtime.running = False
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_settings_save_worker_finished(
page, SimpleNamespace(_request_id=0)
)
page.create_profile_raw_text_save_worker.assert_not_called()
self.assertEqual(len(callbacks), 1)
callbacks[0]()
page.create_profile_raw_text_save_worker.assert_called_once_with(
1,
"profile-1",
"--new\n--lua-desync=split",
parent=page,
)
def test_repeated_raw_profile_save_keeps_only_latest_pending_text(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._settings_save_runtime = _Runtime(running=True)
page._raw_profile_save_runtime = _Runtime(running=False)
page._pending_settings_save = None
page._pending_raw_profile_save = None
page._pending_profile_setup_write_operations = []
page._profile_key = "profile-1"
page.create_profile_raw_text_save_worker = Mock(return_value=_Worker())
ProfileSetupPageBase._request_raw_profile_save(page, "profile-1", "--lua-desync=split")
ProfileSetupPageBase._request_raw_profile_save(page, "profile-1", "--lua-desync=fake")
page.create_profile_raw_text_save_worker.assert_not_called()
self.assertEqual(
page._pending_profile_setup_write_operations,
[
{
"kind": "raw_profile_save",
"profile_key": "profile-1",
"text": "--lua-desync=fake",
}
],
)
def test_user_profile_update_queue_keeps_latest_update_for_same_profile(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._user_profile_update_runtime = _Runtime(running=True)
page._user_profile_delete_runtime = _Runtime(running=False)
page._pending_user_profile_operations = []
page._pending_user_profile_updates = []
page._pending_user_profile_deletes = []
page._user_profile_update_request_id = 0
page._set_user_profile_actions_enabled = Mock()
page.create_profile_user_update_worker = Mock(return_value=_Worker())
ProfileSetupPageBase._request_user_profile_update(
page,
"user-1",
name="Old",
protocol="tcp",
ports="443",
)
ProfileSetupPageBase._request_user_profile_update(
page,
"user-1",
name="Latest",
protocol="udp",
ports="443,500",
)
page.create_profile_user_update_worker.assert_not_called()
self.assertEqual(
page._pending_user_profile_operations,
[
{
"action": "update",
"profile_id": "user-1",
"name": "Latest",
"protocol": "udp",
"ports": "443,500",
}
],
)
self.assertEqual(
page._pending_user_profile_updates,
[
{
"profile_id": "user-1",
"name": "Latest",
"protocol": "udp",
"ports": "443,500",
}
],
)
def test_stale_user_profile_update_worker_finished_does_not_start_pending_operation(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._user_profile_update_runtime = _Runtime(running=False)
page._user_profile_delete_runtime = _Runtime(running=False)
page._user_profile_update_runtime_worker = object()
page._user_profile_write_operation_start_scheduled = False
page._pending_user_profile_operations = [
{"action": "delete", "profile_id": "user-2", "name": "", "protocol": "", "ports": ""}
]
page._pending_user_profile_updates = []
page._pending_user_profile_deletes = ["user-2"]
page._set_user_profile_buttons_enabled = Mock()
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_user_profile_update_worker_finished(page, object())
self.assertEqual(callbacks, [])
self.assertEqual(
page._pending_user_profile_operations,
[{"action": "delete", "profile_id": "user-2", "name": "", "protocol": "", "ports": ""}],
)
page._set_user_profile_buttons_enabled.assert_not_called()
def test_stale_user_profile_delete_worker_finished_does_not_start_pending_operation(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._user_profile_update_runtime = _Runtime(running=False)
page._user_profile_delete_runtime = _Runtime(running=False)
page._user_profile_delete_runtime_worker = object()
page._user_profile_write_operation_start_scheduled = False
page._pending_user_profile_operations = [
{
"action": "update",
"profile_id": "user-1",
"name": "Latest",
"protocol": "tcp",
"ports": "443",
}
]
page._pending_user_profile_updates = [
{"profile_id": "user-1", "name": "Latest", "protocol": "tcp", "ports": "443"}
]
page._pending_user_profile_deletes = []
page._set_user_profile_buttons_enabled = Mock()
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_user_profile_delete_worker_finished(page, object())
self.assertEqual(callbacks, [])
self.assertEqual(
page._pending_user_profile_operations,
[
{
"action": "update",
"profile_id": "user-1",
"name": "Latest",
"protocol": "tcp",
"ports": "443",
}
],
)
page._set_user_profile_buttons_enabled.assert_not_called()
def test_raw_profile_save_waits_while_next_write_start_is_scheduled(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._profile_setup_write_operation_start_scheduled = True
page._raw_profile_save_runtime = _Runtime(running=False)
page._raw_profile_save_request_id = 0
page._pending_raw_profile_save = None
page._pending_profile_setup_write_operations = []
page._profile_key = "profile-1"
page._raw_profile_save_button = None
page.create_profile_raw_text_save_worker = Mock(return_value=_Worker())
ProfileSetupPageBase._request_raw_profile_save(page, "profile-1", "--lua-desync=fake")
page.create_profile_raw_text_save_worker.assert_not_called()
self.assertEqual(page._pending_raw_profile_save, ("profile-1", "--lua-desync=fake"))
self.assertEqual(
page._pending_profile_setup_write_operations,
[
{
"kind": "raw_profile_save",
"profile_key": "profile-1",
"text": "--lua-desync=fake",
}
],
)
def test_scheduled_raw_profile_save_uses_latest_request_before_worker_starts(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._raw_profile_save_runtime = _Runtime(running=False)
page._raw_profile_save_request_id = 0
page._pending_raw_profile_save = None
page._pending_profile_setup_write_operations = []
page._profile_key = "profile-1"
page._raw_profile_save_button = None
page.create_profile_raw_text_save_worker = Mock(return_value=_Worker())
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._schedule_profile_setup_write_operation_start(
page,
{
"kind": "raw_profile_save",
"profile_key": "profile-1",
"text": "--lua-desync=old",
},
)
ProfileSetupPageBase._request_raw_profile_save(page, "profile-1", "--lua-desync=latest")
page.create_profile_raw_text_save_worker.assert_not_called()
self.assertEqual(len(callbacks), 1)
callbacks[0]()
page.create_profile_raw_text_save_worker.assert_called_once_with(
1,
"profile-1",
"--lua-desync=latest",
parent=page,
)
self.assertEqual(page._pending_profile_setup_write_operations, [])
self.assertIsNone(page._pending_raw_profile_save)
def test_strategy_apply_waits_while_raw_profile_save_runs(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._raw_profile_save_runtime = _Runtime(running=True)
page._strategy_apply_runtime = _Runtime(running=False)
page._strategy_apply_request_id = 0
page._strategy_apply_runtime_strategy_id = ""
page._pending_strategy_apply = None
page._pending_profile_setup_write_operations = []
page._profile_key = "profile-1"
page._payload = SimpleNamespace(item=SimpleNamespace(strategy_id="none"))
page.create_profile_strategy_apply_worker = Mock(return_value=_Worker())
ProfileSetupPageBase._request_strategy_apply(page, "tls_fake")
page.create_profile_strategy_apply_worker.assert_not_called()
self.assertEqual(
page._pending_profile_setup_write_operations,
# Операция помнит профиль, открытый при щелчке.
[{"kind": "strategy_apply", "strategy_id": "tls_fake", "profile_key": "profile-1"}],
)
page._raw_profile_save_runtime.running = False
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_raw_profile_save_worker_finished(
page, SimpleNamespace(_request_id=0)
)
page.create_profile_strategy_apply_worker.assert_not_called()
self.assertEqual(len(callbacks), 1)
callbacks[0]()
page.create_profile_strategy_apply_worker.assert_called_once_with(
1,
profile_key="profile-1",
strategy_id="tls_fake",
parent=page,
)
def test_enabled_noop_does_not_enter_write_queue_while_other_save_runs(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._loading = False
page._profile_key = "profile-1"
page._settings_save_runtime = _Runtime(running=True)
page._enabled_save_runtime = _Runtime(running=False)
page._enabled_save_runtime_enabled = None
page._pending_enabled_save = None
page._pending_profile_setup_write_operations = []
page._payload = SimpleNamespace(item=SimpleNamespace(enabled=True))
page.create_profile_enabled_save_worker = Mock(return_value=_Worker())
ProfileSetupPageBase._on_enabled_changed(page, 2)
page.create_profile_enabled_save_worker.assert_not_called()
self.assertEqual(page._pending_profile_setup_write_operations, [])
def test_legacy_pending_settings_save_restarts_later_after_worker_finished(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._settings_save_runtime = _Runtime(running=False)
page._settings_save_request_id = 0
page._pending_profile_setup_write_operations = []
page._pending_settings_save = {
"profile_key": "profile-1",
"filter_kind": "hostlist",
"filter_value": "example.com",
"in_range": "x",
"out_range": "a",
}
page.create_profile_settings_save_worker = Mock(return_value=_Worker())
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_settings_save_worker_finished(
page, SimpleNamespace(_request_id=0)
)
page.create_profile_settings_save_worker.assert_not_called()
self.assertEqual(len(callbacks), 1)
callbacks[0]()
page.create_profile_settings_save_worker.assert_called_once_with(
1,
profile_key="profile-1",
filter_kind="hostlist",
filter_value="example.com",
in_range="x",
out_range="a",
parent=page,
)
def test_legacy_pending_raw_profile_save_restarts_later_after_worker_finished(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._raw_profile_save_runtime = _Runtime(running=False)
page._raw_profile_save_request_id = 0
page._raw_profile_save_button = None
page._pending_profile_setup_write_operations = []
page._pending_raw_profile_save = ("profile-1", "--new\n--lua-desync=split")
page.create_profile_raw_text_save_worker = Mock(return_value=_Worker())
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_raw_profile_save_worker_finished(
page, SimpleNamespace(_request_id=0)
)
page.create_profile_raw_text_save_worker.assert_not_called()
self.assertEqual(len(callbacks), 1)
callbacks[0]()
page.create_profile_raw_text_save_worker.assert_called_once_with(
1,
"profile-1",
"--new\n--lua-desync=split",
parent=page,
)
def test_legacy_pending_strategy_apply_restarts_later_after_worker_finished(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._strategy_apply_runtime = _Runtime(running=False)
page._strategy_apply_request_id = 0
page._strategy_apply_runtime_strategy_id = "old"
page._pending_profile_setup_write_operations = []
page._pending_strategy_apply = "tls_fake"
page._profile_key = "profile-1"
page.create_profile_strategy_apply_worker = Mock(return_value=_Worker())
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_strategy_apply_worker_finished(
page, SimpleNamespace(_request_id=0)
)
page.create_profile_strategy_apply_worker.assert_not_called()
self.assertEqual(len(callbacks), 1)
callbacks[0]()
page.create_profile_strategy_apply_worker.assert_called_once_with(
1,
profile_key="profile-1",
strategy_id="tls_fake",
parent=page,
)
def test_pending_strategy_feedback_save_restarts_later_after_worker_finished(self) -> None:
page = ProfileSetupPageBase.__new__(ProfileSetupPageBase)
page._strategy_feedback_save_runtime = _Runtime(running=False)
page._strategy_feedback_save_request_id = 0
page._pending_strategy_feedback_save = {"strategy_id": "tls_fake", "rating": "work", "favorite": True}
page._profile_key = "profile-1"
# Запись идёт для стратегии из запроса, а не для выбранной в профиле.
page._payload = SimpleNamespace(item=SimpleNamespace(strategy_id="other_strategy"))
page.create_profile_strategy_feedback_save_worker = Mock(return_value=_Worker())
callbacks = []
with patch(
"profile.ui.profile_setup_page.QTimer.singleShot",
side_effect=lambda _delay, callback: callbacks.append(callback),
):
ProfileSetupPageBase._on_strategy_feedback_save_worker_finished(
page, SimpleNamespace(_request_id=0)
)
page.create_profile_strategy_feedback_save_worker.assert_not_called()
self.assertEqual(len(callbacks), 1)
callbacks[0]()
page.create_profile_strategy_feedback_save_worker.assert_called_once_with(
1,
profile_key="profile-1",
strategy_id="tls_fake",
rating="work",
favorite=True,
parent=page,
)
if __name__ == "__main__":
unittest.main()