zapretgui/tests/test_user_preset_write_serialization.py

from __future__ import annotations

import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch

from presets.ui.common.user_presets_page import UserPresetsPageBase


class _Signal:
    def connect(self, _callback) -> None:
        return None


class _Worker:
    completed = _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 = bool(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 _ActivationWorker(_Worker):
    activated = _Signal()


class UserPresetWriteSerializationTests(unittest.TestCase):
    def test_preset_write_queue_lives_in_queued_worker_state(self) -> None:
        import inspect
        from ui.queued_worker_state import QueuedWorkerState

        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        module_source = inspect.getsource(
            __import__("presets.ui.common.user_presets_page", fromlist=[""])
        )
        init_source = inspect.getsource(UserPresetsPageBase.__init__)
        queue_source = inspect.getsource(UserPresetsPageBase._queue_preset_write_action)
        pop_source = inspect.getsource(UserPresetsPageBase._pop_next_preset_write_action)
        has_pending_source = inspect.getsource(UserPresetsPageBase._has_pending_preset_write_action)
        schedule_source = inspect.getsource(UserPresetsPageBase._schedule_preset_write_action_start)
        cleanup_source = inspect.getsource(UserPresetsPageBase._stop_action_workers_for_cleanup)

        self.assertIsInstance(UserPresetsPageBase._preset_write_state_obj(page), QueuedWorkerState)
        self.assertIn("from ui.queued_worker_state import QueuedWorkerState", module_source)
        self.assertIn("_preset_write_state = QueuedWorkerState", init_source)
        self.assertIn("_preset_write_state_obj()", queue_source)
        self.assertIn("state.pop_next()", pop_source)
        self.assertIn("_preset_write_state_obj().has_pending()", has_pending_source)
        self.assertIn("state.start_scheduled", schedule_source)
        self.assertIn("_preset_write_state_obj().reset()", cleanup_source)
        self.assertNotIn("self._pending_preset_write_actions: list", init_source)

    def test_write_finish_uses_queued_state_finish_guard(self) -> None:
        import inspect

        helper_source = inspect.getsource(UserPresetsPageBase._schedule_next_preset_write_action_after_finish)
        finished_source = inspect.getsource(UserPresetsPageBase._on_preset_activate_worker_finished)
        folder_finished_source = inspect.getsource(UserPresetsPageBase._on_preset_folder_action_worker_finished)

        self.assertIn("schedule_next_after_finish", helper_source)
        self.assertIn("_accept_current_preset_worker_request_finished", helper_source)
        self.assertIn("_preset_folder_action_state_obj().schedule_next_after_finish", helper_source)
        self.assertIn("_preset_write_state_obj().schedule_next_after_finish", helper_source)
        self.assertIn("_schedule_next_preset_write_action_after_finish", finished_source)
        self.assertIn("_schedule_next_preset_write_action_after_finish", folder_finished_source)
        self.assertNotIn("_schedule_pending_preset_activation_start", finished_source)
        self.assertFalse(hasattr(UserPresetsPageBase, "_schedule_pending_preset_activation_start"))

    def test_storage_action_waits_while_item_action_worker_runs(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_item_action_runtime = _Runtime(running=True)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._pending_preset_storage_actions = []
        page._preset_storage_action_request_id = 0
        page.create_preset_storage_action_worker = Mock(return_value=_Worker())

        UserPresetsPageBase._request_preset_storage_action(
            page,
            "rating",
            name="Preset.txt",
            display_name="Preset",
            rating=7,
        )

        page.create_preset_storage_action_worker.assert_not_called()
        self.assertEqual(
            page._pending_preset_write_actions,
            [
                {
                    "kind": "storage",
                    "action": "rating",
                    "name": "Preset.txt",
                    "display_name": "Preset",
                    "rating": 7,
                    "direction": 0,
                    "cached_metadata": None,
                    "source_kind": "",
                    "source_id": "",
                    "destination_kind": "",
                    "destination_id": "",
                    "destination_folder_key": "",
                    "file_name": "",
                    "file_path": "",
                    "source_url": "",
                    "auto_update": False,
                }
            ],
        )

    def test_storage_action_waits_while_folder_action_worker_runs(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_folder_action_runtime = _Runtime(running=True)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._pending_preset_storage_actions = []
        page._preset_storage_action_request_id = 0
        page.create_preset_storage_action_worker = Mock(return_value=_Worker())

        UserPresetsPageBase._request_preset_storage_action(
            page,
            "rating",
            name="Preset.txt",
            display_name="Preset",
            rating=7,
        )

        page.create_preset_storage_action_worker.assert_not_called()
        self.assertEqual(page._pending_preset_write_actions[0]["kind"], "storage")
        self.assertEqual(page._pending_preset_storage_actions[0]["action"], "rating")

    def test_folder_action_waits_while_storage_action_worker_runs(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_storage_action_runtime = _Runtime(running=True)
        page._preset_folder_action_runtime = _Runtime(running=False)
        page._preset_folder_action_pending = []
        page._preset_folder_action_request_id = 0
        page.create_preset_folder_action_worker = Mock(return_value=_Worker())

        UserPresetsPageBase._request_preset_folder_action(
            page,
            "set_collapsed",
            folder_key="games",
            collapsed=True,
        )

        page.create_preset_folder_action_worker.assert_not_called()
        self.assertEqual(
            page._preset_folder_action_pending,
            [
                {
                    "action": "set_collapsed",
                    "folder_key": "games",
                    "name": "",
                    "direction": 0,
                    "collapsed": True,
                    "context_extra": {},
                }
            ],
        )

    def test_folder_action_result_ignored_when_new_folder_action_is_pending(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_folder_action_request_id = 5
        page._preset_folder_action_pending = [
            {
                "action": "rename",
                "folder_key": "games",
                "name": "Games",
                "direction": 0,
                "collapsed": False,
                "context_extra": {},
            }
        ]
        page._runtime_service = Mock()
        page._show_folder_menu_with_state = Mock()
        page._refresh_presets_view_from_cache = Mock()

        UserPresetsPageBase._on_preset_folder_action_finished(
            page,
            5,
            "rename",
            True,
            {"folder_state": {"folders": {}, "items": {}}},
        )

        page._runtime_service.update_cached_folder_state.assert_not_called()
        page._show_folder_menu_with_state.assert_not_called()
        page._refresh_presets_view_from_cache.assert_not_called()

    def test_folder_collapse_result_updates_visible_rows_without_full_refresh(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_folder_action_request_id = 6
        page._preset_folder_action_pending = []
        page._runtime_service = Mock()
        page._presets_model = Mock()
        page._presets_model.set_folder_collapsed.return_value = True
        page._update_presets_view_height = Mock()
        page._schedule_layout_resync = Mock()
        page._refresh_presets_view_from_cache = Mock(
            side_effect=AssertionError("folder collapse must not refresh the whole preset list")
        )

        UserPresetsPageBase._on_preset_folder_action_finished(
            page,
            6,
            "set_collapsed",
            True,
            {"folder_key": "games", "collapsed": True, "folder_state": {"folders": {}, "items": {}}},
        )

        page._runtime_service.update_cached_folder_state.assert_called_once_with({"folders": {}, "items": {}})
        page._presets_model.set_folder_collapsed.assert_called_once_with("games", True)
        page._update_presets_view_height.assert_called_once_with()
        page._schedule_layout_resync.assert_called_once_with()
        page._refresh_presets_view_from_cache.assert_not_called()

    def test_folder_expand_result_updates_cached_rows_without_full_refresh(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_folder_action_request_id = 6
        page._preset_folder_action_pending = []
        page._runtime_service = Mock()
        page._presets_model = Mock()
        page._presets_model.set_folder_collapsed.return_value = True
        page._update_presets_view_height = Mock()
        page._schedule_layout_resync = Mock()
        page._refresh_presets_view_from_cache = Mock(
            side_effect=AssertionError("folder expand must not refresh the whole preset list when rows are cached")
        )

        UserPresetsPageBase._on_preset_folder_action_finished(
            page,
            6,
            "set_collapsed",
            True,
            {"folder_key": "games", "collapsed": False, "folder_state": {"folders": {}, "items": {}}},
        )

        page._runtime_service.update_cached_folder_state.assert_called_once_with({"folders": {}, "items": {}})
        page._presets_model.set_folder_collapsed.assert_called_once_with("games", False)
        page._update_presets_view_height.assert_called_once_with()
        page._schedule_layout_resync.assert_called_once_with()
        page._refresh_presets_view_from_cache.assert_not_called()

    def test_storage_action_waits_while_next_write_start_is_scheduled(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_write_action_start_scheduled = True
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._pending_preset_storage_actions = []
        page._preset_storage_action_request_id = 0
        page.create_preset_storage_action_worker = Mock(return_value=_Worker())

        UserPresetsPageBase._request_preset_storage_action(
            page,
            "rating",
            name="Preset.txt",
            display_name="Preset",
            rating=7,
        )

        page.create_preset_storage_action_worker.assert_not_called()
        self.assertEqual(page._pending_preset_write_actions[0]["kind"], "storage")
        self.assertEqual(page._pending_preset_storage_actions[0]["action"], "rating")

    def test_item_action_waits_while_storage_action_worker_runs(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=True)
        page._pending_preset_write_actions = []
        page._preset_item_action_pending = []
        page._preset_item_action_request_id = 0
        page.create_preset_item_action_worker = Mock(return_value=_Worker())

        UserPresetsPageBase._request_preset_item_action(
            page,
            "delete",
            file_name="Preset.txt",
            display_name="Preset",
        )

        page.create_preset_item_action_worker.assert_not_called()
        self.assertEqual(
            page._pending_preset_write_actions,
            [
                {
                    "kind": "item",
                    "action": "delete",
                    "name": "",
                    "display_name": "Preset",
                    "rating": 0,
                    "direction": 0,
                    "cached_metadata": None,
                    "source_kind": "",
                    "source_id": "",
                    "destination_kind": "",
                    "destination_id": "",
                    "destination_folder_key": "",
                    "file_name": "Preset.txt",
                    "file_path": "",
                    "source_url": "",
                    "auto_update": False,
                }
            ],
        )

    def test_activation_waits_while_item_action_worker_runs(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_item_action_runtime = _Runtime(running=True)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._preset_activate_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._pending_preset_activation = None
        page._restore_preset_activation_marker_file_name = ""
        page._preset_activate_request_id = 0
        page._runtime_service = Mock()
        page.create_preset_activate_worker = Mock(return_value=_ActivationWorker())

        UserPresetsPageBase._request_preset_activation(page, "Next.txt", "Next")

        page.create_preset_activate_worker.assert_not_called()
        page._runtime_service.apply_active_preset_marker_for_file.assert_not_called()
        self.assertEqual(
            page._pending_preset_write_actions,
            [
                {
                    "kind": "activate",
                    "action": "",
                    "name": "",
                    "display_name": "Next",
                    "rating": 0,
                    "direction": 0,
                    "cached_metadata": None,
                    "source_kind": "",
                    "source_id": "",
                    "destination_kind": "",
                    "destination_id": "",
                    "destination_folder_key": "",
                    "file_name": "Next.txt",
                    "file_path": "",
                    "source_url": "",
                    "auto_update": False,
                }
            ],
        )

        page._preset_item_action_runtime._running = False
        callbacks = []
        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_item_action_worker_finished(page, SimpleNamespace(_request_id=0))

        page.create_preset_activate_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_activate_worker.assert_called_once_with(
            1,
            file_name="Next.txt",
            display_name="Next",
        )
        page._runtime_service.apply_active_preset_marker_for_file.assert_called_once_with("Next.txt")

    def test_stale_item_action_worker_finished_does_not_start_pending_write(self) -> None:
        old_worker = object()
        current_worker = object()
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_item_action_runtime_worker = current_worker
        page._preset_activate_runtime = _Runtime(running=False)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._preset_folder_action_runtime = _Runtime(running=False)
        page._preset_folder_action_pending = []
        page._preset_write_action_start_scheduled = False
        page._preset_folder_action_start_scheduled = False
        page._cleanup_in_progress = False
        page._pending_preset_write_actions = [
            {
                "kind": "activate",
                "file_name": "Next.txt",
                "display_name": "Next",
            }
        ]
        callbacks = []

        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_item_action_worker_finished(page, old_worker)

        self.assertIs(page._preset_item_action_runtime_worker, current_worker)
        self.assertEqual(len(callbacks), 0)
        self.assertEqual(page._pending_preset_write_actions[0]["file_name"], "Next.txt")

    def test_cleared_item_action_worker_finished_does_not_start_pending_write(self) -> None:
        old_worker = object()
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_item_action_runtime_worker = None
        page._pending_preset_write_actions = [
            {
                "kind": "activate",
                "file_name": "Next.txt",
                "display_name": "Next",
            }
        ]
        callbacks = []

        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_item_action_worker_finished(page, old_worker)

        self.assertIsNone(page._preset_item_action_runtime_worker)
        self.assertEqual(callbacks, [])
        self.assertEqual(page._pending_preset_write_actions[0]["file_name"], "Next.txt")

    def test_scheduled_activation_uses_latest_request_before_worker_starts(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=False)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._preset_activate_request_id = 0
        page._pending_preset_activation = None
        page._pending_preset_write_actions = []
        page._restore_preset_activation_marker_file_name = ""
        page._runtime_service = Mock()
        page.create_preset_activate_worker = Mock(return_value=_ActivationWorker())
        callbacks = []

        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._schedule_preset_write_action_start(
                page,
                {
                    "kind": "activate",
                    "file_name": "Old.txt",
                    "display_name": "Old",
                },
            )
            UserPresetsPageBase._request_preset_activation(page, "Latest.txt", "Latest")

        page.create_preset_activate_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_activate_worker.assert_called_once_with(
            1,
            file_name="Latest.txt",
            display_name="Latest",
        )
        page._runtime_service.apply_active_preset_marker_for_file.assert_called_once_with("Latest.txt")
        self.assertEqual(page._pending_preset_write_actions, [])
        self.assertIsNone(page._pending_preset_activation)

    def test_pending_activation_queue_keeps_only_latest_preset_click(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=True)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._pending_preset_activation = None
        page.create_preset_activate_worker = Mock(return_value=_ActivationWorker())

        UserPresetsPageBase._request_preset_activation(page, "Old.txt", "Old")
        UserPresetsPageBase._request_preset_activation(page, "Latest.txt", "Latest")

        page.create_preset_activate_worker.assert_not_called()
        self.assertEqual(
            [
                (operation["kind"], operation["file_name"], operation["display_name"])
                for operation in page._pending_preset_write_actions
            ],
            [("activate", "Latest.txt", "Latest")],
        )
        self.assertEqual(page._pending_preset_activation, ("Latest.txt", "Latest"))

    def test_edit_action_waits_while_activation_worker_runs(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=True)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._preset_edit_action_pending = []
        page._preset_edit_action_request_id = 0
        page.create_preset_edit_action_worker = Mock(return_value=_Worker())

        UserPresetsPageBase._request_preset_edit_action(
            page,
            "rename",
            current_name="Old.txt",
            new_name="New.txt",
        )

        page.create_preset_edit_action_worker.assert_not_called()
        self.assertEqual(page._pending_preset_write_actions[0]["kind"], "edit")

        page._preset_activate_runtime._running = False
        callbacks = []
        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_activate_worker_finished(page, SimpleNamespace(_request_id=0))

        page.create_preset_edit_action_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_edit_action_worker.assert_called_once_with(
            1,
            action="rename",
            name="",
            current_name="Old.txt",
            new_name="New.txt",
            from_current=False,
        )

    def test_bulk_action_waits_while_edit_action_worker_runs(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=False)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=True)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._preset_bulk_action_pending = []
        page._preset_bulk_action_request_id = 0
        page._preset_bulk_action_kind = ""
        page.create_preset_bulk_action_worker = Mock(return_value=_Worker())

        self.assertTrue(
            UserPresetsPageBase._request_preset_bulk_action(
                page,
                "import",
                file_path="C:/Temp/Preset.txt",
            )
        )

        page.create_preset_bulk_action_worker.assert_not_called()
        self.assertEqual(page._pending_preset_write_actions[0]["kind"], "bulk")

        page._preset_edit_action_runtime._running = False
        callbacks = []
        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_edit_action_worker_finished(page, SimpleNamespace(_request_id=0))

        page.create_preset_bulk_action_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_bulk_action_worker.assert_called_once_with(
            1,
            action="import",
            file_path="C:/Temp/Preset.txt",
            source_url="",
            auto_update=False,
        )

    def test_folder_action_restarts_later_after_storage_worker_finished(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=False)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._preset_folder_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._preset_folder_action_pending = [
            {
                "action": "set_collapsed",
                "folder_key": "games",
                "name": "",
                "direction": 0,
                "collapsed": True,
                "context_extra": {},
            }
        ]
        page._preset_folder_action_request_id = 0
        page.create_preset_folder_action_worker = Mock(return_value=_Worker())
        callbacks = []

        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_storage_action_worker_finished(page, SimpleNamespace(_request_id=0))

        page.create_preset_folder_action_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_folder_action_worker.assert_called_once_with(
            1,
            action="set_collapsed",
            folder_key="games",
            name="",
            direction=0,
            collapsed=True,
            context_extra={},
        )

    def test_storage_action_restarts_later_after_folder_worker_finished(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=False)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._preset_folder_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = [
            {
                "kind": "storage",
                "action": "rating",
                "name": "Default.txt",
                "display_name": "Default",
                "rating": 5,
                "direction": 0,
                "cached_metadata": None,
                "source_kind": "",
                "source_id": "",
                "destination_kind": "",
                "destination_id": "",
                "destination_folder_key": "",
                "file_name": "",
                "file_path": "",
                "source_url": "",
                "auto_update": False,
            }
        ]
        page._pending_preset_storage_actions = [
            {
                "action": "rating",
                "name": "Default.txt",
                "display_name": "Default",
                "rating": 5,
                "direction": 0,
                "cached_metadata": None,
                "source_kind": "",
                "source_id": "",
                "destination_kind": "",
                "destination_id": "",
                "destination_folder_key": "",
            }
        ]
        page._preset_folder_action_pending = []
        page._preset_storage_action_request_id = 0
        page.create_preset_storage_action_worker = Mock(return_value=_Worker())
        callbacks = []

        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_folder_action_worker_finished(page, SimpleNamespace(_request_id=0))

        page.create_preset_storage_action_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_storage_action_worker.assert_called_once_with(
            1,
            action="rating",
            name="Default.txt",
            display_name="Default",
            rating=5,
            direction=0,
            cached_metadata=None,
            source_kind="",
            source_id="",
            destination_kind="",
            destination_id="",
            destination_folder_key="",
        )

    def test_write_action_errors_ignored_when_next_write_is_pending(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_edit_action_request_id = 2
        page._preset_bulk_action_request_id = 3
        page._preset_item_action_request_id = 4
        page._pending_preset_write_actions = [{"kind": "activate", "file_name": "Next.txt"}]
        page._pending_preset_storage_actions = []
        page._preset_item_action_pending = []
        page._preset_bulk_action_pending = []
        page._preset_edit_action_pending = []
        page._pending_preset_activation = None
        page._preset_folder_action_pending = []
        page._config = Mock(tr_prefix="page.user_presets")
        page._ui_language = "ru"
        page.window = Mock(return_value=None)

        with (
            patch("presets.ui.common.user_presets_page.InfoBar.error") as error,
            patch("presets.ui.common.user_presets_page.log") as log_mock,
        ):
            UserPresetsPageBase._on_preset_edit_action_failed(page, 2, "rename", "old edit", {})
            # Импорт — отдельный файл, его ошибка показывается всегда;
            # откладывается только результат «Сбросить все».
            UserPresetsPageBase._on_preset_bulk_action_failed(page, 3, "reset_all", "old bulk", {})
            UserPresetsPageBase._on_preset_item_action_failed(page, 4, "delete", "old item")

        error.assert_not_called()
        log_mock.assert_not_called()

    def test_write_action_results_ignored_when_next_write_is_pending(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_edit_action_request_id = 2
        page._preset_bulk_action_request_id = 3
        page._preset_storage_action_request_id = 4
        page._preset_item_action_request_id = 5
        page._pending_preset_write_actions = [{"kind": "activate", "file_name": "Next.txt"}]
        page._pending_preset_storage_actions = []
        page._preset_item_action_pending = []
        page._preset_bulk_action_pending = []
        page._preset_edit_action_pending = []
        page._pending_preset_activation = None
        page._preset_folder_action_pending = []
        page._runtime_service = Mock()
        page._config = Mock(tr_prefix="page.user_presets")
        page._ui_language = "ru"
        page.window = Mock(return_value=None)
        page._refresh_presets_view_from_cache = Mock()
        page._update_cached_preset_rating = Mock()
        page._apply_preset_move_locally = Mock(return_value=False)

        edit_result = SimpleNamespace(
            ok=True,
            structure_changed=True,
            preset_file_name="Renamed.txt",
            preset_display_name="Renamed",
            log_message="old edit",
            log_level="INFO",
        )
        bulk_result = SimpleNamespace(
            ok=True,
            structure_changed=True,
            actual_file_name="Imported.txt",
            actual_name="Imported",
            infobar_level="success",
            infobar_title="Imported",
            infobar_content="Done",
            log_message="old bulk",
            log_level="INFO",
        )
        item_result = SimpleNamespace(
            ok=True,
            structure_changed=True,
            error_code="",
            preset_file_name="Copy.txt",
            preset_display_name="Copy",
            infobar_level="success",
            infobar_title="Copied",
            infobar_content="Done",
            log_message="old item",
            log_level="INFO",
        )

        with (
            patch("presets.ui.common.user_presets_page.InfoBar.success") as success,
            patch("presets.ui.common.user_presets_page.InfoBar.warning") as warning,
            patch("presets.ui.common.user_presets_page.InfoBar.error") as error,
            patch("presets.ui.common.user_presets_page.log") as log_mock,
        ):
            UserPresetsPageBase._on_preset_edit_action_finished(
                page,
                2,
                "rename",
                edit_result,
                {"current_name": "Old.txt", "new_name": "Renamed.txt"},
            )
            UserPresetsPageBase._on_preset_bulk_action_finished(page, 3, "reset_all", bulk_result, {})
            UserPresetsPageBase._on_preset_storage_action_finished(
                page,
                4,
                "rating",
                True,
                {"name": "Old.txt", "rating": 9, "folder_state": {"folders": {}, "items": {}}},
            )
            UserPresetsPageBase._on_preset_item_action_finished(
                page,
                5,
                "duplicate",
                item_result,
                {"file_name": "Old.txt"},
            )

        log_mock.assert_not_called()
        success.assert_not_called()
        warning.assert_not_called()
        error.assert_not_called()
        page._runtime_service.add_created_preset_locally.assert_not_called()
        page._runtime_service.rename_preset_locally.assert_not_called()
        page._runtime_service.mark_presets_structure_changed.assert_not_called()
        page._runtime_service.update_cached_folder_state.assert_not_called()
        page._refresh_presets_view_from_cache.assert_not_called()
        page._update_cached_preset_rating.assert_not_called()
        page._apply_preset_move_locally.assert_not_called()

    def test_rating_result_updates_single_visible_row_without_full_refresh(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_storage_action_request_id = 4
        page._pending_preset_write_actions = []
        page._pending_preset_storage_actions = []
        page._preset_item_action_pending = []
        page._preset_bulk_action_pending = []
        page._preset_edit_action_pending = []
        page._pending_preset_activation = None
        page._preset_folder_action_pending = []
        page._runtime_service = Mock()
        page._runtime_service.cached_presets_metadata.return_value = {
            "Preset.txt": {"name": "Preset", "rating": 1}
        }
        page._presets_model = Mock()
        page._presets_model.update_preset_row.return_value = True
        page._refresh_presets_view_from_cache = Mock(
            side_effect=AssertionError("rating must not refresh the whole preset list")
        )
        page._update_presets_view_height = Mock()
        page._schedule_layout_resync = Mock()

        with patch("presets.ui.common.user_presets_page.log") as log_mock:
            UserPresetsPageBase._on_preset_storage_action_finished(
                page,
                4,
                "rating",
                True,
                {"name": "Preset.txt", "rating": 7, "folder_state": {"folders": {}, "items": {}}},
            )

        page._presets_model.update_preset_row.assert_called_once_with("Preset.txt", rating=7)
        page._refresh_presets_view_from_cache.assert_not_called()
        page._update_presets_view_height.assert_not_called()
        page._schedule_layout_resync.assert_not_called()
        log_mock.assert_not_called()

    def test_move_step_result_updates_visible_row_without_full_refresh_when_context_is_known(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_storage_action_request_id = 4
        page._pending_preset_write_actions = []
        page._pending_preset_storage_actions = []
        page._preset_item_action_pending = []
        page._preset_bulk_action_pending = []
        page._preset_edit_action_pending = []
        page._pending_preset_activation = None
        page._preset_folder_action_pending = []
        page._runtime_service = Mock()
        page._apply_preset_move_locally = Mock(return_value=True)
        page._refresh_presets_view_from_cache = Mock(
            side_effect=AssertionError("move_step must not refresh the whole preset list")
        )

        UserPresetsPageBase._on_preset_storage_action_finished(
            page,
            4,
            "move_step",
            True,
            {
                "name": "Preset.txt",
                "destination_kind": "preset_after",
                "destination_id": "Other.txt",
                "destination_folder_key": "common",
                "folder_state": {"folders": {}, "items": {}},
            },
        )

        page._apply_preset_move_locally.assert_called_once_with(
            "Preset.txt",
            "preset_after",
            "Other.txt",
            "common",
        )
        page._refresh_presets_view_from_cache.assert_not_called()

    def test_pinned_move_skips_incompatible_optimistic_folder_model(self) -> None:
        from ui.presets_menu.model import PresetListModel

        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._presets_model = PresetListModel()
        page._presets_model.set_rows(
            [
                {
                    "kind": "folder",
                    "folder_key": "pinned",
                    "name": "Закрепленные",
                    "is_service": True,
                    "count": 1,
                },
                {
                    "kind": "preset",
                    "file_name": "Pinned.txt",
                    "name": "Pinned",
                    "folder_key": "common",
                    "is_pinned": True,
                },
            ]
        )
        page._runtime_service = Mock()
        page._update_presets_view_height = Mock()
        page._schedule_layout_resync = Mock()

        self.assertFalse(
            UserPresetsPageBase._apply_preset_move_locally(
                page,
                "Pinned.txt",
                "preset_after",
                "Other.txt",
                "common",
            )
        )

        page._runtime_service.capture_presets_view_state.assert_not_called()
        page._update_presets_view_height.assert_not_called()
        page._schedule_layout_resync.assert_not_called()

    def test_pending_move_step_result_updates_local_order_and_folder_cache(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_storage_action_request_id = 4
        page._pending_preset_write_actions = [
            {
                "kind": "storage",
                "action": "move_step",
                "name": "Next.txt",
                "direction": 1,
            }
        ]
        page._preset_folder_action_pending = []
        page._runtime_service = Mock()
        page._apply_preset_move_locally = Mock(return_value=True)
        page._refresh_presets_view_from_cache = Mock(
            side_effect=AssertionError("successful queued move must apply locally")
        )
        folder_state = {"folders": {}, "items": {"Preset.txt": {"order": 1}}}

        UserPresetsPageBase._on_preset_storage_action_finished(
            page,
            4,
            "move_step",
            True,
            {
                "name": "Preset.txt",
                "destination_kind": "preset_after",
                "destination_id": "Other.txt",
                "destination_folder_key": "common",
                "folder_state": folder_state,
            },
        )

        page._runtime_service.update_cached_folder_state.assert_called_once_with(folder_state)
        page._apply_preset_move_locally.assert_called_once_with(
            "Preset.txt",
            "preset_after",
            "Other.txt",
            "common",
        )

    def test_repeated_pending_folder_moves_are_kept_as_separate_steps(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_folder_action_runtime = _Runtime(running=True)
        page._preset_folder_action_pending = []

        payload = {
            "action": "move",
            "folder_key": "games",
            "name": "",
            "direction": 1,
            "collapsed": False,
            "context_extra": {},
        }
        UserPresetsPageBase._queue_preset_folder_action(page, payload)
        UserPresetsPageBase._queue_preset_folder_action(page, payload)

        self.assertEqual(page._preset_folder_action_pending, [payload, payload])

    def test_legacy_pending_edit_action_restarts_later_after_worker_finished(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=False)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._preset_edit_action_pending = [
            {
                "action": "rename",
                "name": "Old.txt",
                "current_name": "Old.txt",
                "new_name": "New.txt",
                "from_current": False,
            }
        ]
        page._preset_edit_action_request_id = 0
        page.create_preset_edit_action_worker = Mock(return_value=_Worker())
        callbacks = []

        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_edit_action_worker_finished(page, SimpleNamespace(_request_id=0))

        page.create_preset_edit_action_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_edit_action_worker.assert_called_once_with(
            1,
            action="rename",
            name="Old.txt",
            current_name="Old.txt",
            new_name="New.txt",
            from_current=False,
        )

    def test_legacy_pending_bulk_action_restarts_later_after_worker_finished(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._preset_activate_runtime = _Runtime(running=False)
        page._preset_item_action_runtime = _Runtime(running=False)
        page._preset_bulk_action_runtime = _Runtime(running=False)
        page._preset_edit_action_runtime = _Runtime(running=False)
        page._preset_storage_action_runtime = _Runtime(running=False)
        page._pending_preset_write_actions = []
        page._preset_bulk_action_pending = [{"action": "import", "file_path": "C:/Temp/Preset.txt"}]
        page._preset_bulk_action_request_id = 0
        page._preset_bulk_action_kind = ""
        page._bulk_reset_running = False
        page.create_preset_bulk_action_worker = Mock(return_value=_Worker())
        callbacks = []

        with patch(
            "presets.ui.common.user_presets_page.QTimer.singleShot",
            side_effect=lambda _delay, callback: callbacks.append(callback),
        ):
            UserPresetsPageBase._on_preset_bulk_action_worker_finished(page, SimpleNamespace(_request_id=0))

        page.create_preset_bulk_action_worker.assert_not_called()
        self.assertEqual(len(callbacks), 1)

        callbacks[0]()

        page.create_preset_bulk_action_worker.assert_called_once_with(
            1,
            action="import",
            file_path="C:/Temp/Preset.txt",
            source_url="",
            auto_update=False,
        )

    def test_cleanup_clears_pending_write_actions(self) -> None:
        page = UserPresetsPageBase.__new__(UserPresetsPageBase)
        page._pending_preset_activation = ("Preset.txt", "Preset")
        page._restore_preset_activation_marker_file_name = "Old.txt"
        page._pending_preset_write_actions = [{"kind": "item", "action": "delete"}]
        page._pending_preset_storage_actions = [{"action": "rating"}]
        page._preset_item_action_pending = [{"action": "delete"}]
        page._preset_folder_action_pending = []
        page._preset_open_folder_pending = True
        page._preset_bulk_action_kind = "reset_all"
        page._bulk_reset_running = True
        for attr in (
            "_preset_activate_request_id",
            "_preset_item_action_request_id",
            "_preset_bulk_action_request_id",
            "_preset_edit_action_request_id",
            "_preset_storage_action_request_id",
            "_preset_folder_action_request_id",
            "_preset_open_folder_request_id",
            "_preset_link_action_request_id",
        ):
            setattr(page, attr, 0)

        UserPresetsPageBase._stop_action_workers_for_cleanup(page)

        self.assertEqual(page._pending_preset_write_actions, [])
        self.assertEqual(page._pending_preset_storage_actions, [])
        self.assertEqual(page._preset_item_action_pending, [])


if __name__ == "__main__":
    unittest.main()