# Copyright (C) 2024–present  Loren Eteval & contributors <loren.eteval@proton.me>
#
# This file is part of Furious.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""Verify subscription workflow ownership outside table widgets."""

import os
import sys
import threading
import time

os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')

from Furious.Repository import Storage
from Furious.Repository.Subscriptions import SubscriptionGroup
from Furious.Qt import gettext
from Furious.Service.SubscriptionManager import (
    SubscriptionManager,
    _SubscriptionBatchState,
    _parseSubscriptionUserInfo,
)
from Furious.Service.SubscriptionPreparation import SubscriptionPreparationOutcome
from Furious.Window.SubscriptionPage import SubscriptionPage
from Furious.Widget.SubscriptionTableView import SubscriptionTableView

from tests.support import application, processQtEvents, waitFor
from tests.fixtures.editor_lifetime_probe import runSubscriptionWorkerTeardownProbe

from PySide6 import QtCore, QtNetwork, QtTest, QtWidgets

from shiboken6 import isValid, delete as deleteQObject

from types import SimpleNamespace
from unittest import TestCase, mock


class _Reply(QtNetwork.QNetworkReply):
    """Supply in-memory data while preserving Qt's real header API contract."""

    def __init__(self, value=b'', error='request failed', headers=None):
        super().__init__()

        self._value = value
        self._error = error

        for name, headerValue in (headers or {}).items():
            self.setRawHeader(name, headerValue)

        self.open(QtCore.QIODevice.OpenModeFlag.ReadOnly)

    def readData(self, maxSize):
        data, self._value = self._value[:maxSize], self._value[maxSize:]

        return data

    def errorString(self):
        return self._error

    def abort(self):
        self.close()


class _AbortableReply:
    """Record cancellation of one exact service-owned network reply."""

    def __init__(self):
        self.aborted = False

    def abort(self):
        self.aborted = True


class SubscriptionManagerTest(TestCase):
    """Protect decoding, failure, and stable-ID scheduling boundaries."""

    @classmethod
    def setUpClass(cls):
        cls.application = application()

    @staticmethod
    def _manager(subscriptions=None):
        subscriptions = subscriptions if subscriptions is not None else {}

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            return SubscriptionManager()

    @staticmethod
    def _subscription(**overrides):
        """Return one enabled five-minute subscription definition."""
        subscription = {
            'remark': 'Group A',
            'webURL': 'https://invalid.test/a',
            'autoupdate': 'Every 5 mins',
            'proxy': '',
            'enabled': True,
        }
        subscription.update(overrides)

        return subscription

    def testSuccessfulAndInvalidPayloadsProduceSemanticBatchInputs(self):
        manager = self._manager()
        profile = SimpleNamespace(itemRemark='profile')
        imported = SimpleNamespace(
            decoderId='decoder',
            profiles=(profile,),
            rejectedItems=2,
        )
        manager.importer = SimpleNamespace(
            importPayload=mock.Mock(return_value=imported)
        )
        successful = []
        failed = []

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.SubscriptionGroup',
            return_value=None,
        ):
            manager.successCallback(
                _Reply(
                    b'payload',
                    headers={
                        b'Subscription-Userinfo': (
                            b'upload=1024; download=2048; total=8192; '
                            b'expire=1893456000'
                        )
                    },
                ),
                unique='group-a',
                remark='Group A',
                webURL='https://invalid.test/subscription',
                successArgs=successful,
                failureArgs=failed,
            )

        self.assertEqual(len(successful), 1)
        self.assertEqual(successful[0]['profiles'], (profile,))
        self.assertEqual(successful[0]['decoderId'], 'decoder')
        self.assertEqual(
            successful[0]['subscriptionInfo'],
            {
                'upload': 1024,
                'download': 2048,
                'total': 8192,
                'expire': 1893456000,
            },
        )
        self.assertEqual(failed, [])

        manager.importer = SimpleNamespace(importPayload=mock.Mock(return_value=None))

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.SubscriptionGroup',
            return_value=None,
        ):
            manager.successCallback(
                _Reply(b'invalid'),
                unique='group-a',
                successArgs=successful,
                failureArgs=failed,
            )

        self.assertEqual(failed[-1]['error'], 'UnsupportedSubscriptionFormat')
        manager.deleteLater()

    def testSubscriptionUserInfoParserRejectsInvalidAndUnboundedValues(self):
        """Treat provider quota headers as bounded advisory network input."""
        self.assertIsNone(_parseSubscriptionUserInfo(b''))
        self.assertIsNone(_parseSubscriptionUserInfo(b'upload=\xff'))
        self.assertEqual(
            _parseSubscriptionUserInfo(QtCore.QByteArray(b'Upload=1')),
            {'upload': 1, 'download': 0, 'total': 0, 'expire': 0},
        )
        self.assertEqual(
            _parseSubscriptionUserInfo(
                b'upload=1024; download=2048; total=8192; expire=1893456000; '
                b'ignored=value; upload=-1; total=999999999999999999999999'
            ),
            {
                'upload': 1024,
                'download': 2048,
                'total': 8192,
                'expire': 1893456000,
            },
        )

    def testBatchResponseCarriesSubscriptionInfoIntoWorkerContext(self):
        """Keep response metadata attached to the exact asynchronous request."""
        manager = self._manager()
        manager._isCurrentRequest = mock.Mock(return_value=True)
        manager.importer = SimpleNamespace(
            registry=SimpleNamespace(
                subscriptionDecoderWorkerSafe=mock.Mock(return_value=True)
            )
        )
        manager._startImportPreparation = mock.Mock()

        manager.successCallback(
            _Reply(
                b'payload',
                headers={
                    b'Subscription-Userinfo': (
                        b'upload=1024; download=2048; total=8192; ' b'expire=1893456000'
                    )
                },
            ),
            unique='group-a',
            batchId=7,
            lastDecoderId='decoder',
        )

        payload, context = manager._startImportPreparation.call_args.args
        self.assertEqual(payload, b'payload')
        self.assertEqual(
            context['subscriptionInfo'],
            {
                'upload': 1024,
                'download': 2048,
                'total': 8192,
                'expire': 1893456000,
            },
        )
        manager.deleteLater()

    def testSuccessfulCommitReplacesSubscriptionInfoButFailurePreservesIt(self):
        """Tie advisory metadata to the same successful group commit boundary."""
        group = SubscriptionGroup(
            id='group-a',
            subscriptionUpload=1,
            subscriptionDownload=2,
            subscriptionTotal=3,
            subscriptionExpire=4,
        )
        result = SimpleNamespace(profileIds=('profile-a', 'profile-b'))

        with (
            mock.patch.object(Storage, 'SubscriptionGroup', return_value=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroup') as upsert,
        ):
            SubscriptionManager._recordGroupSuccess(
                {
                    'unique': 'group-a',
                    'decoderId': 'decoder',
                    'subscriptionInfo': {
                        'upload': 1024,
                        'download': 2048,
                        'total': 8192,
                        'expire': 1893456000,
                    },
                },
                result,
            )

            SubscriptionManager._recordGroupFailure(
                {'unique': 'group-a', 'error': 'offline'}
            )
            metadataAfterFailure = (
                group.subscriptionUpload,
                group.subscriptionDownload,
                group.subscriptionTotal,
                group.subscriptionExpire,
            )

            SubscriptionManager._recordGroupSuccess(
                {'unique': 'group-a', 'subscriptionInfo': None},
                result,
            )

        self.assertEqual(metadataAfterFailure, (1024, 2048, 8192, 1893456000))

        self.assertEqual(group.subscriptionUpload, 0)
        self.assertEqual(group.subscriptionDownload, 0)
        self.assertEqual(group.subscriptionTotal, 0)
        self.assertEqual(group.subscriptionExpire, 0)
        self.assertEqual(group.lastSyncStatus, 'success')
        self.assertEqual(upsert.call_count, 3)

    def testRequestFailureIsDataForPresentationNotAWidgetSideEffect(self):
        manager = self._manager()
        failed = []

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.SubscriptionGroup',
            return_value=None,
        ):
            manager.failureCallback(
                _Reply(error='offline'),
                unique='group-a',
                remark='Group A',
                webURL='https://invalid.test/subscription',
                failureArgs=failed,
            )

        self.assertEqual(failed[-1]['error'], 'offline')
        self.assertFalse(hasattr(manager, 'table'))
        manager.deleteLater()

    def testSubscriptionDiagnosticsExcludeConfiguredURL(self):
        """Identify a request without logging its credential-bearing URL."""
        manager = self._manager()
        profile = SimpleNamespace(itemRemark='profile')
        manager.importer = SimpleNamespace(
            importPayload=mock.Mock(
                return_value=SimpleNamespace(
                    decoderId='decoder',
                    profiles=(profile,),
                    rejectedItems=0,
                )
            )
        )
        configuredURL = 'https://example.invalid/subscription?token=value'

        with mock.patch('Furious.Service.SubscriptionManager.logger.info') as infoLog:
            manager.successCallback(
                _Reply(b'payload'),
                unique='group-a',
                remark='Group A',
                webURL=configuredURL,
                successArgs=[],
                failureArgs=[],
            )

        with mock.patch('Furious.Service.SubscriptionManager.logger.error') as errorLog:
            manager.failureCallback(
                _Reply(error='offline'),
                unique='group-a',
                remark='Group A',
                webURL=configuredURL,
                failureArgs=[],
            )

        self.assertIn('Group A', infoLog.call_args.args[0])
        self.assertIn('Group A', errorLog.call_args.args[0])
        self.assertIn('group-a', infoLog.call_args.args[0])
        self.assertIn('group-a', errorLog.call_args.args[0])
        self.assertNotIn(configuredURL, infoLog.call_args.args[0])
        self.assertNotIn(configuredURL, errorLog.call_args.args[0])
        self.assertNotIn('token=value', infoLog.call_args.args[0])
        self.assertNotIn('token=value', errorLog.call_args.args[0])

        manager.deleteLater()

    def testStaleRequestCompletionCannotMutateCurrentSubscription(self):
        subscriptions = {
            'group-a': {
                'webURL': 'https://invalid.test/current',
                'enabled': True,
            }
        }
        manager = self._manager(subscriptions)
        manager._requestVersions['group-a'] = 2
        manager.importer = SimpleNamespace(importPayload=mock.Mock())

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            manager.successCallback(
                _Reply(b'stale'),
                unique='group-a',
                webURL='https://invalid.test/current',
                requestVersion=1,
                successArgs=[],
                failureArgs=[],
            )

        manager.importer.importPayload.assert_not_called()
        manager.deleteLater()

    def testStaleBatchEntriesAreNotReconciledOrReported(self):
        subscriptions = {
            'group-a': {
                'webURL': 'https://invalid.test/current',
                'enabled': True,
            }
        }
        manager = self._manager(subscriptions)
        manager._requestVersions['group-a'] = 2
        manager._synchronizeProfiles = mock.Mock()
        completed = []
        manager.updateCompleted.connect(completed.append)
        stale = {
            'unique': 'group-a',
            'webURL': 'https://invalid.test/current',
            'requestVersion': 1,
            'profiles': (),
        }

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            manager.handleSynchronizationResults(
                successArgs=[stale],
                failureArgs=[{'error': 'offline', **stale}],
            )

        manager._synchronizeProfiles.assert_not_called()
        self.assertEqual(completed, [])
        manager.deleteLater()

    def testStaleDecodedResultCannotCommitSubscriptionMetadata(self):
        """Delay group metadata writes until the final current-request check."""
        subscriptions = {
            'group-a': {
                'webURL': 'https://invalid.test/current',
                'enabled': True,
            }
        }

        manager = self._manager(subscriptions)
        manager._requestVersions['group-a'] = 1

        profile = SimpleNamespace(itemRemark='profile')
        manager.importer = SimpleNamespace(
            importPayload=mock.Mock(
                return_value=SimpleNamespace(
                    decoderId='decoder',
                    profiles=(profile,),
                    rejectedItems=0,
                )
            )
        )

        successful = []
        failed = []

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            manager.successCallback(
                _Reply(b'payload'),
                unique='group-a',
                webURL='https://invalid.test/current',
                requestVersion=1,
                successArgs=successful,
                failureArgs=failed,
            )

        manager._requestVersions['group-a'] = 2

        with (
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.UserSubs',
                return_value=subscriptions,
            ),
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.upsertSubscriptionGroup'
            ) as upsert,
        ):
            manager.handleSynchronizationResults(
                successArgs=successful,
                failureArgs=failed,
            )

        upsert.assert_not_called()

        manager.deleteLater()

    def testOneSynchronizationFailureDoesNotAbortOtherGroups(self):
        """Isolate one group's preparation failure from the rest of a batch."""
        manager = self._manager()
        committed = SimpleNamespace(profileIds=('profile-id',))
        manager._isCurrentRequest = mock.Mock(return_value=True)
        manager._synchronizeProfiles = mock.Mock(
            side_effect=(RuntimeError('injected failure'), committed)
        )

        completed = []
        committedSubscriptions = []
        stateChanges = []
        structuralChanges = []

        manager.updateCompleted.connect(completed.append)
        manager.subscriptionCommitted.connect(committedSubscriptions.append)
        manager.subscriptionStateChanged.connect(
            lambda uniques: stateChanges.append(tuple(uniques))
        )
        manager.subscriptionsChanged.connect(lambda: structuralChanges.append(True))

        failed = {'unique': 'group-a', 'profiles': ()}
        successful = {'unique': 'group-b', 'profiles': ()}

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.SubscriptionGroup',
            return_value=None,
        ):
            manager.handleSynchronizationResults(
                successArgs=[failed, successful],
                failureArgs=[],
            )

        self.assertEqual(manager._synchronizeProfiles.call_count, 2)
        self.assertEqual(len(completed), 1)
        self.assertEqual(completed[0].successful[0]['unique'], 'group-b')
        self.assertEqual(completed[0].failed[0]['unique'], 'group-a')
        self.assertIn('injected failure', completed[0].failed[0]['error'])

        self.assertEqual(committedSubscriptions, ['group-b'])
        self.assertEqual(stateChanges, [('group-b', 'group-a')])
        self.assertEqual(structuralChanges, [True])

        manager.deleteLater()

    def testFailedBatchPublishesStateWithoutStructuralChange(self):
        """Present terminal failure metadata without resetting profile consumers."""
        manager = self._manager()
        stateChanges = []
        structuralChanges = []
        completed = []

        manager.subscriptionStateChanged.connect(
            lambda uniques: stateChanges.append(tuple(uniques))
        )
        manager.subscriptionsChanged.connect(lambda: structuralChanges.append(True))
        manager.updateCompleted.connect(completed.append)

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.SubscriptionGroup',
            return_value=None,
        ):
            manager.handleSynchronizationResults(
                successArgs=[],
                failureArgs=[{'unique': 'group-a', 'error': 'offline'}],
            )

        self.assertEqual(stateChanges, [('group-a',)])
        self.assertEqual(structuralChanges, [])
        self.assertEqual(len(completed), 1)
        self.assertEqual(completed[0].failed[0]['error'], 'offline')

        manager.deleteLater()

    def testCommittedMetadataFailureDoesNotAbortOtherGroups(self):
        """Treat status metadata as post-commit and keep processing the batch."""
        manager = self._manager()
        result = SimpleNamespace(profileIds=('profile-id',))
        manager._isCurrentRequest = mock.Mock(return_value=True)
        manager._synchronizeProfiles = mock.Mock(return_value=result)
        completed = []
        manager.updateCompleted.connect(completed.append)

        with (
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.SubscriptionGroup',
                side_effect=(SimpleNamespace(), SimpleNamespace()),
            ),
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.upsertSubscriptionGroup',
                side_effect=(RuntimeError('metadata write failed'), None),
            ) as upsert,
        ):
            manager.handleSynchronizationResults(
                successArgs=[
                    {'unique': 'group-a', 'profiles': ()},
                    {'unique': 'group-b', 'profiles': ()},
                ],
                failureArgs=[],
            )

        self.assertEqual(upsert.call_count, 2)
        self.assertEqual(len(completed), 1)
        self.assertEqual(
            [item['unique'] for item in completed[0].successful],
            ['group-a', 'group-b'],
        )

        manager.deleteLater()

    def testPostCommitConnectionFailureDoesNotUndoSynchronization(self):
        """Keep a committed reconciliation successful if reconnect later fails."""
        manager = self._manager()
        active = SimpleNamespace(
            metadata=SimpleNamespace(profileId='active-profile'),
            itemSubscription='group-a',
            itemSubscriptionManaged=True,
        )
        servers = [active]
        result = SimpleNamespace(
            profileIds=('active-profile',),
            changedProfileIds=('active-profile',),
        )
        manager.synchronizer.reconcile = mock.Mock(return_value=result)
        controller = SimpleNamespace(
            isConnected=mock.Mock(return_value=True),
            startDisconnection=mock.Mock(),
            startReconnection=mock.Mock(side_effect=RuntimeError('reconnect failed')),
        )

        with (
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.UserServers',
                return_value=servers,
            ),
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.UserActivatedItemIndex',
                return_value=0,
            ),
            mock.patch(
                'Furious.Service.SubscriptionManager.AppConnectionController',
                return_value=controller,
            ),
            mock.patch(
                'Furious.Service.SubscriptionManager.AppSettings.set',
                side_effect=RuntimeError('settings write failed'),
            ),
        ):
            committed = manager._synchronizeProfiles('group-a', ())

        self.assertIs(committed, result)

        controller.startReconnection.assert_called_once_with()
        manager.deleteLater()

    def testCancellationInvalidatesAndAbortsOnlyTheSelectedSubscription(self):
        manager = self._manager()
        groupAReply = _AbortableReply()
        groupBReply = _AbortableReply()
        manager._requestVersions.update({'group-a': 1, 'group-b': 4})
        manager._activeReplies.update(
            {groupAReply: groupAReply, groupBReply: groupBReply}
        )
        manager._replySubscriptions.update(
            {groupAReply: 'group-a', groupBReply: 'group-b'}
        )

        manager.cancelUpdates('group-a')

        self.assertTrue(groupAReply.aborted)
        self.assertFalse(groupBReply.aborted)
        self.assertEqual(manager._requestVersions, {'group-a': 2, 'group-b': 4})

        manager._activeReplies.clear()
        manager._replySubscriptions.clear()
        manager.deleteLater()

    def testUnchangedAutoUpdateReconciliationPreservesDeadlineAndConnection(self):
        """Make repeated full reconciliation a true scheduler no-op."""
        subscriptions = {'group-a': self._subscription()}
        manager = self._manager(subscriptions)
        timer = manager._autoUpdateTimers['group-a']
        timerId = timer.timerId()

        QtTest.QTest.qWait(40)

        remainingBefore = timer.remainingTime()

        with (
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.UserSubs',
                return_value=subscriptions,
            ),
            mock.patch(
                'Furious.Service.SubscriptionManager.logger.info'
            ) as lifecycleLog,
        ):
            for _index in range(100):
                manager.refreshAutoUpdates()

        self.assertIs(manager._autoUpdateTimers['group-a'], timer)
        self.assertEqual(timer.timerId(), timerId)
        self.assertEqual(timer.property('subscriptionId'), 'group-a')
        self.assertEqual(len(manager._autoUpdateTimers), 1)
        self.assertLessEqual(timer.remainingTime(), remainingBefore + 5)
        lifecycleLog.assert_not_called()

        manager.configureHttpProxy = mock.Mock()
        manager.updateSubsByUnique = mock.Mock()

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            timer.timeout.emit()

        manager.updateSubsByUnique.assert_called_once_with(
            'group-a', showMessageBox=False
        )
        manager.deleteLater()

    def testAutoUpdatePolicyTransitionsReuseTimerAndLogOnlyRealChanges(self):
        """Start, reschedule, and stop exactly when policy state changes."""
        subscriptions = {
            'group-a': self._subscription(autoupdate='Never'),
        }
        manager = self._manager(subscriptions)
        timer = manager._autoUpdateTimers['group-a']

        with (
            mock.patch(
                'Furious.Service.SubscriptionManager.Storage.UserSubs',
                return_value=subscriptions,
            ),
            mock.patch(
                'Furious.Service.SubscriptionManager.logger.info'
            ) as lifecycleLog,
        ):
            manager.configureAutoUpdate('group-a')

            lifecycleLog.assert_not_called()

            subscriptions['group-a']['autoupdate'] = 'Every 5 mins'

            manager.configureAutoUpdate('group-a')

            self.assertTrue(timer.isActive())
            self.assertEqual(timer.interval(), 5 * 60 * 1000)
            self.assertIn('start auto update job', lifecycleLog.call_args.args[0])

            activeTimerId = timer.timerId()

            manager.configureAutoUpdate('group-a')

            self.assertEqual(timer.timerId(), activeTimerId)
            self.assertEqual(lifecycleLog.call_count, 1)

            subscriptions['group-a']['autoupdate'] = 'Every 10 mins'

            manager.configureAutoUpdate('group-a')

            self.assertIs(manager._autoUpdateTimers['group-a'], timer)
            self.assertEqual(timer.interval(), 10 * 60 * 1000)
            self.assertIn('reschedule auto update job', lifecycleLog.call_args.args[0])

            subscriptions['group-a']['enabled'] = False

            manager.configureAutoUpdate('group-a')

            self.assertFalse(timer.isActive())
            self.assertIn('stop auto update job', lifecycleLog.call_args.args[0])

            manager.configureAutoUpdate('group-a')

            self.assertEqual(lifecycleLog.call_count, 3)

            subscriptions['group-a']['enabled'] = True

            manager.configureAutoUpdate('group-a')

            self.assertTrue(timer.isActive())
            self.assertIn('start auto update job', lifecycleLog.call_args.args[0])

        self.assertEqual(lifecycleLog.call_count, 4)

        manager.deleteLater()

    def testUnrelatedSubscriptionEditDoesNotRestartItsTimer(self):
        """Keep the real table edit path outside unchanged timer deadlines."""
        subscriptions = {'group-a': self._subscription()}
        manager = self._manager(subscriptions)
        timer = manager._autoUpdateTimers['group-a']

        QtTest.QTest.qWait(40)

        remainingBefore = timer.remainingTime()
        timerId = timer.timerId()

        def group(unique):
            """Return the exact in-memory group edited by the table."""
            value = subscriptions.get(unique)

            return (
                SubscriptionGroup.fromMapping(unique, value)
                if value is not None
                else None
            )

        def upsert(value):
            """Persist the edited group into the isolated test repository."""
            subscriptions[value.id] = value.toMapping()

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'SubscriptionGroup', side_effect=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroup', side_effect=upsert),
        ):
            table = SubscriptionTableView(subscriptionManager=manager)
            table.appendNewItem(
                unique='group-a',
                remark='Renamed Group',
                webURL=subscriptions['group-a']['webURL'],
                enabled=True,
                autoupdate='Every 5 mins',
                proxy='',
                userAgent='',
                filter='',
                lastUpdated='',
            )

        self.assertEqual(subscriptions['group-a']['remark'], 'Renamed Group')
        self.assertIs(manager._autoUpdateTimers['group-a'], timer)
        self.assertEqual(timer.timerId(), timerId)
        self.assertLessEqual(timer.remainingTime(), remainingBefore + 5)
        table.deleteLater()
        manager.deleteLater()

    def testSubscriptionTableExposesPersistedSynchronizationStatus(self):
        """Present the existing sync state without inventing a UI-owned copy."""
        subscriptions = {
            'group-a': self._subscription(lastSyncStatus=''),
        }

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch(
                'Furious.Widget.SubscriptionTableView._',
                side_effect=lambda text: text,
            ),
        ):
            table = SubscriptionTableView()
            model = table.sourceModel
            statusColumn = table.ItemKey.index('lastSyncStatus')
            statusIndex = model.index(0, statusColumn)

            for persisted, expected in (
                ('', 'Never'),
                ('syncing', 'Updating...'),
                ('success', 'Updated'),
                ('error', 'Update Failed'),
                ('cancelled', 'Update Cancelled'),
                ('future-value', 'Never'),
            ):
                with self.subTest(persisted=persisted):
                    subscriptions['group-a']['lastSyncStatus'] = persisted

                    self.assertEqual(
                        model.data(statusIndex, QtCore.Qt.ItemDataRole.DisplayRole),
                        expected,
                    )

            self.assertFalse(
                model.flags(statusIndex) & QtCore.Qt.ItemFlag.ItemIsEditable
            )

            table.deleteLater()

        self.assertEqual(
            gettext('Sync Status', 'RU'),
            'Статус синхронизации',
        )
        self.assertEqual(gettext('Updating...', 'RU'), 'Обновление...')
        self.assertEqual(gettext('Updated', 'ZH'), '已更新')
        self.assertEqual(gettext('Update Failed', 'ZH'), '更新失败')
        self.assertEqual(gettext('Update Cancelled', 'ZH'), '更新已取消')
        self.assertEqual(gettext('Update Cancelled', 'RU'), 'Обновление отменено')
        self.assertEqual(gettext('Stop Updates', 'ZH'), '停止更新')
        self.assertEqual(gettext('Stop Updates', 'RU'), 'Остановить обновления')
        self.assertEqual(gettext('Usage / Expiry', 'RU'), 'Трафик / Срок')
        self.assertEqual(gettext('Usage / Expiry', 'ZH'), '用量 / 到期')

    def testSubscriptionTableShowsOptionalUsageAndExpiryMetadata(self):
        """Keep provider metadata compact and blank when it is unavailable."""
        subscriptions = {
            'group-a': self._subscription(
                subscriptionUpload=1024,
                subscriptionDownload=2048,
                subscriptionTotal=8192,
                subscriptionExpire=1893456000,
            ),
            'group-b': self._subscription(
                remark='Group B',
                webURL='https://invalid.test/b',
            ),
        }

        with mock.patch.object(Storage, 'UserSubs', return_value=subscriptions):
            table = SubscriptionTableView()
            model = table.sourceModel
            column = table.ItemKey.index('subscriptionInfo')

            self.assertEqual(
                model.data(model.index(0, column), QtCore.Qt.ItemDataRole.DisplayRole),
                '3 KiB / 8 KiB · 2030-01-01',
            )
            self.assertEqual(
                model.data(model.index(1, column), QtCore.Qt.ItemDataRole.DisplayRole),
                '',
            )
            self.assertFalse(
                model.flags(model.index(0, column)) & QtCore.Qt.ItemFlag.ItemIsEditable
            )
            table.deleteLater()

    def testSubscriptionStateNotificationRepaintsOnlyAffectedMetadataRow(self):
        """Resolve stable IDs at delivery and avoid a whole-table refresh."""
        subscriptions = {
            'group-a': self._subscription(),
            'group-b': self._subscription(
                remark='Group B',
                webURL='https://invalid.test/b',
            ),
            'group-c': self._subscription(
                remark='Group C',
                webURL='https://invalid.test/c',
            ),
        }

        with mock.patch.object(Storage, 'UserSubs', return_value=subscriptions):
            table = SubscriptionTableView()
            changes = []

            def changed(topLeft, bottomRight, _roles):
                """Capture one exact model repaint range."""
                changes.append(
                    (
                        (topLeft.row(), topLeft.column()),
                        (bottomRight.row(), bottomRight.column()),
                    )
                )

            table.sourceModel.dataChanged.connect(changed)

            with mock.patch.object(table, 'flushAll') as flushAll:
                table.refreshSubscriptionState(('group-b',))

            self.assertEqual(
                changes,
                [
                    (
                        (1, table.ItemKey.index('lastSyncStatus')),
                        (1, table.ItemKey.index('profiles')),
                    )
                ],
            )
            flushAll.assert_not_called()
            table.deleteLater()

    def testUpdateAllPublishesOneImmediateSyncingSnapshot(self):
        """Expose one metadata batch without publishing structural changes."""
        subscriptions = {
            'group-a': self._subscription(),
            'group-b': self._subscription(
                remark='Group B',
                webURL='https://invalid.test/b',
            ),
        }
        manager = self._manager(subscriptions)
        snapshots = []
        stateChanges = []
        structuralChanges = []

        def group(unique):
            """Return one isolated persisted group."""
            value = subscriptions.get(unique)

            return (
                SubscriptionGroup.fromMapping(unique, value)
                if value is not None
                else None
            )

        def upsert(groups):
            """Persist one status batch into the isolated repository."""
            for value in groups:
                subscriptions[value.id] = value.toMapping()

        def stateChanged(uniques):
            """Capture the stable IDs and their state at notification time."""
            stateChanges.append(tuple(uniques))
            snapshots.append(
                tuple(
                    subscriptions[unique].get('lastSyncStatus', '')
                    for unique in ('group-a', 'group-b')
                )
            )

        manager.subscriptionStateChanged.connect(stateChanged)
        manager.subscriptionsChanged.connect(lambda: structuralChanges.append(True))

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'SubscriptionGroup', side_effect=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroups', side_effect=upsert),
            mock.patch.object(Storage, 'persistSubscriptionGroups') as persist,
            mock.patch.object(manager, 'updateSubsByWebGET') as update,
        ):
            manager.updateSubs()

        self.assertEqual(snapshots, [('syncing', 'syncing')])
        self.assertEqual(stateChanges, [('group-a', 'group-b')])
        self.assertEqual(structuralChanges, [])
        persist.assert_called_once_with()
        self.assertEqual(update.call_count, 2)
        self.assertEqual(
            {call.kwargs['unique'] for call in update.call_args_list},
            {'group-a', 'group-b'},
        )

        manager.deleteLater()

    def testTargetedUpdatePublishesImmediateSyncingState(self):
        """Publish one narrow targeted state before network I/O."""
        subscriptions = {'group-a': self._subscription()}
        manager = self._manager(subscriptions)
        group = SubscriptionGroup.fromMapping('group-a', subscriptions['group-a'])
        snapshots = []
        stateChanges = []
        structuralChanges = []

        def stateChanged(uniques):
            """Capture the target and persisted state before network I/O."""
            stateChanges.append(tuple(uniques))
            snapshots.append(group.lastSyncStatus)

        manager.subscriptionStateChanged.connect(stateChanged)
        manager.subscriptionsChanged.connect(lambda: structuralChanges.append(True))

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'SubscriptionGroup', return_value=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroups'),
            mock.patch.object(Storage, 'persistSubscriptionGroups') as persist,
            mock.patch.object(manager, 'updateSubsByWebGET') as update,
        ):
            manager.updateSubsByUnique('group-a')

        self.assertEqual(snapshots, ['syncing'])
        self.assertEqual(stateChanges, [('group-a',)])
        self.assertEqual(structuralChanges, [])
        persist.assert_called_once_with()
        update.assert_called_once()

        manager.deleteLater()

    def testTargetedPolicyChangeDoesNotDisturbOtherSubscriptions(self):
        """Reconcile only the edited subscription's schedule."""
        subscriptions = {
            'group-a': self._subscription(),
            'group-b': self._subscription(
                remark='Group B',
                webURL='https://invalid.test/b',
                autoupdate='Every 10 mins',
            ),
        }
        manager = self._manager(subscriptions)
        groupATimer = manager._autoUpdateTimers['group-a']
        groupBTimer = manager._autoUpdateTimers['group-b']
        groupBTimerId = groupBTimer.timerId()

        QtTest.QTest.qWait(40)

        groupBRemainingBefore = groupBTimer.remainingTime()
        subscriptions['group-a']['autoupdate'] = 'Every 10 mins'

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            manager.configureAutoUpdate('group-a')

        self.assertIs(manager._autoUpdateTimers['group-a'], groupATimer)
        self.assertEqual(groupATimer.interval(), 10 * 60 * 1000)

        self.assertIs(manager._autoUpdateTimers['group-b'], groupBTimer)
        self.assertEqual(groupBTimer.timerId(), groupBTimerId)
        self.assertLessEqual(groupBTimer.remainingTime(), groupBRemainingBefore + 5)
        self.assertEqual(len(manager._autoUpdateTimers), 2)

        manager.deleteLater()

    def testRemovingSubscriptionDestroysOnlyItsTimerAndCancelsItsReply(self):
        """Release one removed subscription without disturbing its sibling."""
        subscriptions = {
            'group-a': self._subscription(),
            'group-b': self._subscription(
                remark='Group B',
                webURL='https://invalid.test/b',
            ),
        }

        manager = self._manager(subscriptions)
        groupATimer = manager._autoUpdateTimers['group-a']
        groupBTimer = manager._autoUpdateTimers['group-b']
        groupBTimerId = groupBTimer.timerId()

        groupAReply = _AbortableReply()
        groupBReply = _AbortableReply()
        destroyed = []
        groupATimer.destroyed.connect(lambda *_args: destroyed.append(True))

        manager._activeReplies.update(
            {groupAReply: groupAReply, groupBReply: groupBReply}
        )
        manager._replySubscriptions.update(
            {groupAReply: 'group-a', groupBReply: 'group-b'}
        )

        subscriptions.pop('group-a')

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            manager.removeAutoUpdate('group-a')

        self.assertTrue(groupAReply.aborted)
        self.assertFalse(groupBReply.aborted)
        self.assertNotIn('group-a', manager._autoUpdateTimers)
        self.assertIs(manager._autoUpdateTimers['group-b'], groupBTimer)
        self.assertEqual(groupBTimer.timerId(), groupBTimerId)
        self.assertTrue(groupBTimer.isActive())

        processQtEvents()

        self.assertEqual(destroyed, [True])
        self.assertFalse(isValid(groupATimer))

        manager._activeReplies.clear()
        manager._replySubscriptions.clear()
        manager.deleteLater()

    def testUpdateSelectedUsesOneOrderedCanonicalBatchCall(self):
        """Keep a real selected-update click narrow and event-loop responsive."""
        subscriptions = {
            'group-a': self._subscription(),
            'group-b': self._subscription(
                remark='Group B',
                webURL='https://invalid.test/b',
            ),
            'group-c': self._subscription(
                remark='Group C',
                webURL='https://invalid.test/c',
            ),
        }

        def group(unique):
            """Return one isolated persisted group."""
            value = subscriptions.get(unique)

            return (
                SubscriptionGroup.fromMapping(unique, value)
                if value is not None
                else None
            )

        def upsert(groups):
            """Persist one status batch into the isolated repository."""
            for value in groups:
                subscriptions[value.id] = value.toMapping()

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'SubscriptionGroup', side_effect=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroups', side_effect=upsert),
            mock.patch.object(Storage, 'persistSubscriptionGroups'),
        ):
            manager = SubscriptionManager()
            stateChanges = []
            structuralChanges = []

            def updateSubscriptions(uniques, _httpProxy, **kwargs):
                """Cross the production wrapper boundary into the real manager."""
                kwargs.pop('parent', None)
                manager.updateSubscriptions(uniques, **kwargs)

            serverTable = SimpleNamespace(subsManager=manager)
            serverTable.updateSubscriptions = mock.Mock(side_effect=updateSubscriptions)
            page = SubscriptionPage(serverTable)
            selection = page.table.selectionModel()
            flags = (
                QtCore.QItemSelectionModel.SelectionFlag.Select
                | QtCore.QItemSelectionModel.SelectionFlag.Rows
            )

            selection.select(page.table.sourceModel.index(0, 0), flags)
            selection.select(page.table.sourceModel.index(2, 0), flags)

            manager.subscriptionStateChanged.connect(
                lambda uniques: stateChanges.append(tuple(uniques))
            )
            manager.subscriptionsChanged.connect(lambda: structuralChanges.append(True))

            page.show()
            processQtEvents(1)

            handled = []
            QtCore.QTimer.singleShot(0, lambda: handled.append(True))

            with mock.patch.object(manager, 'updateSubsByWebGET') as update:
                QtTest.QTest.mouseClick(
                    page.updateSelectedButton,
                    QtCore.Qt.MouseButton.LeftButton,
                )

            processQtEvents(1)

            serverTable.updateSubscriptions.assert_called_once_with(
                ('group-a', 'group-c'),
                None,
                showMessageBox=True,
                parent=page,
            )
            self.assertEqual(stateChanges, [('group-a', 'group-c')])
            self.assertEqual(structuralChanges, [])
            self.assertEqual(update.call_count, 2)
            self.assertEqual(handled, [True])

            page.deleteLater()
            manager.deleteLater()

        processQtEvents()

    def testStopUpdatesClickPreservesCommitsSchedulesAndRejectsLateWork(self):
        """Cancel a partial batch through real Qt input and allow another update."""
        subscriptions = {
            unique: self._subscription(remark=unique)
            for unique in ('group-a', 'group-b')
        }

        def group(unique):
            return SubscriptionGroup.fromMapping(unique, subscriptions[unique])

        def upsert(groups):
            for value in groups:
                subscriptions[value.id] = value.toMapping()

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'SubscriptionGroup', side_effect=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroups', side_effect=upsert),
            mock.patch.object(Storage, 'persistSubscriptionGroups'),
        ):
            manager = SubscriptionManager()
            page = SubscriptionPage(SimpleNamespace(subsManager=manager))

            replies = []
            contexts = []
            completed = []
            manager.updateCompleted.connect(completed.append)

            timerIds = {
                key: timer.timerId() for key, timer in manager._autoUpdateTimers.items()
            }

            def request(**context):
                reply = _Reply()
                replies.append(reply)
                contexts.append(context)
                manager._activeReplies[reply] = reply
                manager._replySubscriptions[reply] = context['unique']

            try:
                page.show()
                processQtEvents()

                self.assertFalse(page.stopUpdatesButton.isEnabled())

                with mock.patch.object(
                    manager, 'updateSubsByWebGET', side_effect=request
                ):
                    manager.updateSubscriptions(('group-a', 'group-b'))

                    self.assertTrue(page.stopUpdatesButton.isEnabled())

                    subscriptions['group-a']['lastSyncStatus'] = 'success'
                    subscriptions['group-a']['lastUpdated'] = 'preserved timestamp'

                    manager._finishOperation(
                        contexts[0], successful=contexts[0], structural=True
                    )
                    manager._releaseFinishedReply(replies[0])

                    QtTest.QTest.mouseClick(
                        page.stopUpdatesButton, QtCore.Qt.MouseButton.LeftButton
                    )
                    processQtEvents()

                    self.assertFalse(page.stopUpdatesButton.isEnabled())
                    self.assertEqual(
                        subscriptions['group-a']['lastSyncStatus'], 'success'
                    )
                    self.assertEqual(
                        subscriptions['group-a']['lastUpdated'], 'preserved timestamp'
                    )
                    self.assertEqual(
                        subscriptions['group-b']['lastSyncStatus'], 'cancelled'
                    )

                    self.assertFalse(replies[1].isOpen())
                    self.assertEqual(manager._batches, {})
                    self.assertEqual(len(completed), 1)
                    self.assertEqual(completed[0].successful, (contexts[0],))
                    self.assertEqual(completed[0].failed, ())
                    self.assertFalse(manager._isCurrentRequest(contexts[1]))

                    self.assertEqual(
                        timerIds,
                        {
                            key: timer.timerId()
                            for key, timer in manager._autoUpdateTimers.items()
                        },
                    )

                    with mock.patch.object(
                        manager, '_startImportPreparation'
                    ) as prepare:
                        manager.successCallback(replies[1], **contexts[1])

                        prepare.assert_not_called()

                    manager.updateSubscriptions(('group-b',))

                    self.assertTrue(page.stopUpdatesButton.isEnabled())
                    self.assertTrue(manager._isCurrentRequest(contexts[2]))

                    manager.stopUpdates()

                    self.assertEqual(len(completed), 1)
            finally:
                manager.shutdown()

                for reply in replies:
                    manager._releaseFinishedReply(reply)
                    reply.deleteLater()

                page.deleteLater()
                manager.deleteLater()
                processQtEvents()

    def testStopCompletionCanImmediatelyStartANewUpdate(self):
        """A synchronous abort observer must not lose the next generation's status."""
        subscriptions = {'group-a': self._subscription(lastSyncStatus='syncing')}
        manager = self._manager()

        context = {'unique': 'group-a', 'batchId': 1, 'requestVersion': 1}
        manager._nextBatchId = 1
        manager._requestVersions['group-a'] = 1
        manager._batches[1] = _SubscriptionBatchState(
            {('group-a', 1)}, False, [{'unique': 'completed-group'}], []
        )

        reply = mock.Mock()
        manager._activeReplies[reply] = reply
        manager._replySubscriptions[reply] = 'group-a'

        def abort():
            manager._releaseFinishedReply(reply)
            manager._finishOperation(context)

        reply.abort.side_effect = abort
        manager.updateCompleted.connect(
            lambda _batch: manager.updateSubsByUnique('group-a')
        )

        def upsert(groups):
            for group in groups:
                subscriptions[group.id] = group.toMapping()

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(
                Storage,
                'SubscriptionGroup',
                side_effect=lambda unique: SubscriptionGroup.fromMapping(
                    unique, subscriptions[unique]
                ),
            ),
            mock.patch.object(Storage, 'upsertSubscriptionGroups', side_effect=upsert),
            mock.patch.object(Storage, 'persistSubscriptionGroups'),
            mock.patch.object(manager, 'updateSubsByWebGET') as request,
        ):
            try:
                manager.stopUpdates()

                request.assert_called_once()

                self.assertTrue(manager._isCurrentRequest(request.call_args.kwargs))
                self.assertEqual(subscriptions['group-a']['lastSyncStatus'], 'syncing')
                self.assertEqual(len(manager._batches), 1)
            finally:
                manager.shutdown()
                manager.deleteLater()

                processQtEvents()

    def testAbortDoesNotCancelReplacementPreparation(self):
        manager = self._manager({})
        reply = _Reply()
        reply.setParent(manager)
        manager._activeReplies[reply] = reply
        manager._replySubscriptions[reply] = 'group-a'
        replacement = SimpleNamespace(context={'unique': 'group-a'}, cancel=mock.Mock())

        def abort():
            manager._releaseFinishedReply(reply)
            manager._preparationJobs[1] = replacement

        reply.abort = abort
        manager.cancelUpdates('group-a')
        replacement.cancel.assert_not_called()
        manager._preparationJobs.clear()
        manager.shutdown()
        manager.deleteLater()
        processQtEvents()

    def testAbortReplacementKeepsItsOwnRequestGeneration(self):
        subscriptions = {'group-a': self._subscription(autoupdate='Never')}
        manager = self._manager(subscriptions)
        reply = _Reply()
        reply.setParent(manager)
        manager._activeReplies[reply] = reply
        manager._replySubscriptions[reply] = 'group-a'

        def abort():
            manager._releaseFinishedReply(reply)
            manager.updateSubsByUnique('group-a')

        reply.abort = abort
        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(
                Storage,
                'SubscriptionGroup',
                side_effect=lambda unique: SubscriptionGroup.fromMapping(
                    unique, subscriptions[unique]
                ),
            ),
            mock.patch.object(Storage, 'upsertSubscriptionGroups'),
            mock.patch.object(Storage, 'persistSubscriptionGroups'),
            mock.patch.object(manager, 'updateSubsByWebGET') as request,
        ):
            manager.updateSubsByUnique('group-a')
            request.assert_called_once()
            self.assertEqual(request.call_args.kwargs['batchId'], 2)
            self.assertEqual(set(manager._batches), {2})

        manager.shutdown()
        manager.deleteLater()
        processQtEvents()

    def testStopUpdatesRetainsRunningPreparationUntilItsLateResultArrives(self):
        """Logical stop releases the batch, while the pool retains executing work."""
        manager = self._manager()
        started = threading.Event()
        release = threading.Event()

        context = {'unique': 'group-a', 'batchId': 1, 'requestVersion': 1}
        subscriptions = {'group-a': self._subscription()}

        manager._requestVersions['group-a'] = 1
        context['webURL'] = subscriptions['group-a']['webURL']
        manager._batches[1] = _SubscriptionBatchState({('group-a', 1)}, True, [], [])
        manager._handleImportedResult = mock.Mock()

        completed = []
        manager.updateCompleted.connect(completed.append)

        def work(_isCancelled):
            started.set()
            release.wait(3)

            return object()

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'persistSubscriptionGroups'),
        ):
            try:
                manager._startPreparationJob('import', context, work)
                self.assertTrue(started.wait(2))

                manager.stopUpdates()

                self.assertEqual(manager._batches, {})
                self.assertEqual(len(manager._preparationJobs), 1)
                self.assertEqual(manager._preparationPool.activeThreadCount(), 1)

                release.set()
                self.assertTrue(waitFor(lambda: not manager._preparationJobs))

                manager._handleImportedResult.assert_not_called()
                self.assertEqual(completed, [])
            finally:
                release.set()
                manager.shutdown()
                manager.deleteLater()

                processQtEvents()

    def testLargePreparationRunsOffGuiThreadAndKeepsEventLoopResponsive(self):
        """Keep unrelated Qt delivery responsive while payload work is gated."""
        subscriptions = {'group-a': self._subscription(autoupdate='Never')}
        started = threading.Event()
        release = threading.Event()
        workerThreads = []

        class Registry:
            """Declare this deterministic test importer worker-safe."""

            @staticmethod
            def subscriptionDecoderWorkerSafe(_decoderId):
                return True

        class Importer:
            """Gate representative expensive parsing in the pool."""

            registry = Registry()

            @staticmethod
            def importPayload(_data, _source, **_kwargs):
                workerThreads.append(threading.get_ident())
                started.set()
                release.wait(2)

                return None

        def group(unique):
            return SubscriptionGroup.fromMapping(unique, subscriptions[unique])

        def upsert(groups):
            for value in groups:
                subscriptions[value.id] = value.toMapping()

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'SubscriptionGroup', side_effect=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroups', side_effect=upsert),
            mock.patch.object(
                Storage,
                'upsertSubscriptionGroup',
                side_effect=lambda value: upsert((value,)),
            ),
            mock.patch.object(Storage, 'persistSubscriptionGroups') as persist,
        ):
            manager = SubscriptionManager()
            manager.importer = Importer()
            requests = []
            manager.updateSubsByWebGET = lambda **context: requests.append(context)
            completed = []
            manager.updateCompleted.connect(completed.append)

            manager.updateSubsByUnique('group-a')
            manager.successCallback(_Reply(b'large payload'), **requests[0])

            self.assertTrue(started.wait(2))

            handled = []

            QtCore.QTimer.singleShot(0, lambda: handled.append(True))
            processQtEvents(1)

            self.assertEqual(handled, [True])
            self.assertNotEqual(workerThreads, [threading.get_ident()])

            release.set()
            deadline = time.monotonic() + 3

            while not completed and time.monotonic() < deadline:
                QtTest.QTest.qWait(10)

            self.assertEqual(len(completed), 1)
            self.assertEqual(
                completed[0].failed[0]['error'],
                'UnsupportedSubscriptionFormat',
            )
            self.assertEqual(persist.call_count, 2)

            manager.shutdown()
            manager.deleteLater()

        processQtEvents()

    def testParsingResultIsDiscardedAfterReupdateDeleteOrSourceEdit(self):
        """Reject worker results after every source-generation invalidation path."""
        base = self._subscription(autoupdate='Never')

        for mutation in ('reupdate', 'delete', 'edit'):
            with self.subTest(mutation=mutation):
                subscriptions = {'group-a': dict(base)}
                manager = self._manager(subscriptions)
                manager._requestVersions['group-a'] = 1
                context = {
                    'unique': 'group-a',
                    'webURL': base['webURL'],
                    'requestVersion': 1,
                    'requestSignature': manager._requestSignature(base),
                }
                manager._handleImportedResult = mock.Mock()

                if mutation == 'reupdate':
                    manager._nextRequestVersion('group-a')
                elif mutation == 'delete':
                    subscriptions.clear()
                else:
                    subscriptions['group-a']['filter'] = 'changed'

                with mock.patch.object(Storage, 'UserSubs', return_value=subscriptions):
                    self.assertFalse(manager._isCurrentRequest(context))

                manager._handleImportedResult.assert_not_called()
                manager.shutdown()
                manager.deleteLater()

        processQtEvents()

    def testPartialBatchFailuresRemainIsolatedAndPublishOnce(self):
        """Commit successful members and report failed members as one batch."""
        subscriptions = {
            unique: self._subscription(
                remark=unique,
                webURL=f'https://invalid.test/{unique}',
                autoupdate='Never',
            )
            for unique in ('group-a', 'group-b', 'group-c', 'group-d')
        }
        manager = self._manager(subscriptions)
        contexts = []

        for unique, subscription in subscriptions.items():
            manager._requestVersions[unique] = 1
            contexts.append(
                {
                    'unique': unique,
                    'remark': subscription['remark'],
                    'webURL': subscription['webURL'],
                    'requestVersion': 1,
                    'requestSignature': manager._requestSignature(subscription),
                    'batchId': 7,
                }
            )

        manager._batches[7] = _SubscriptionBatchState(
            {(context['unique'], 1) for context in contexts},
            True,
            [],
            [],
        )

        completed = []
        structural = []
        manager.updateCompleted.connect(completed.append)
        manager.subscriptionsChanged.connect(lambda: structural.append(True))

        def group(unique):
            return SubscriptionGroup.fromMapping(unique, subscriptions[unique])

        def upsert(value):
            subscriptions[value.id] = value.toMapping()

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(Storage, 'SubscriptionGroup', side_effect=group),
            mock.patch.object(Storage, 'upsertSubscriptionGroup', side_effect=upsert),
            mock.patch.object(Storage, 'persistSubscriptionGroups') as persist,
        ):
            manager._finishOperation(
                contexts[0], successful=contexts[0], structural=True
            )
            manager._failOperation(contexts[1], 'decode failed')
            manager._failOperation(contexts[2], 'network failed')
            manager._finishOperation(
                contexts[3], successful=contexts[3], structural=True
            )

        self.assertEqual(len(completed), 1)
        self.assertEqual(
            [value['unique'] for value in completed[0].successful],
            ['group-a', 'group-d'],
        )
        self.assertEqual(
            [value['unique'] for value in completed[0].failed],
            ['group-b', 'group-c'],
        )
        self.assertEqual(structural, [True])
        persist.assert_called_once_with()

        manager.shutdown()
        manager.deleteLater()
        processQtEvents()

    def testNativeDestructionCancelsPreparationBeforeDeletingRelay(self):
        """A held invalid wrapper must not retain work or its copied payload."""
        for duringAbort in (False, True):
            with self.subTest(duringAbort=duringAbort):
                manager = self._manager({})
                manager._preparationPool.setMaxThreadCount(1)

                started = threading.Event()
                queuedStarted = threading.Event()
                observations = []
                relay = manager._preparationRelay

                def work(isCancelled):
                    started.set()
                    deadline = time.monotonic() + 1

                    while not isCancelled() and time.monotonic() < deadline:
                        threading.Event().wait(0.005)

                    observations.append((isCancelled(), isValid(relay)))

                manager._startPreparationJob('probe', {}, work)
                self.assertTrue(started.wait(2))

                manager._startPreparationJob('probe', {}, lambda _: queuedStarted.set())
                manager._preparationPayloads[1] = b'copied payload'

                if duringAbort:
                    reply = _Reply()
                    reply.setParent(manager)
                    reply.abort = lambda: deleteQObject(manager)
                    manager._activeReplies[reply] = reply

                    manager.shutdown()
                else:
                    deleteQObject(manager)

                self.assertFalse(isValid(manager))
                self.assertEqual(observations, [(True, True)])
                self.assertFalse(queuedStarted.is_set())
                self.assertFalse(manager._preparationJobs)
                self.assertFalse(manager._preparationPayloads)
                self.assertFalse(manager._batches)

                processQtEvents()

    def testRepeatedPreparationTeardownReclaimsWorkersAndDeliversOnOwnerThread(self):
        errors = []
        with mock.patch.object(
            sys, 'excepthook', side_effect=lambda *args: errors.append(args)
        ):
            result = runSubscriptionWorkerTeardownProbe(30)

        self.assertFalse(errors)
        self.assertEqual(result['subscriptionObjectsDestroyed'], 90)
        self.assertEqual(result['subscriptionWrappersRetained'], 0)
        self.assertEqual(result['subscriptionOwnerThreadDeliveries'], 30)

    def testBatchPublicationStopsWhenStructuralObserverDestroysManager(self):
        manager = self._manager({})
        manager._batches[1] = _SubscriptionBatchState(
            {('group-a', None)}, False, [], []
        )
        manager.subscriptionsChanged.connect(lambda: deleteQObject(manager))

        with mock.patch.object(Storage, 'persistSubscriptionGroups'):
            manager._finishOperation(
                {'batchId': 1, 'unique': 'group-a'},
                successful={'unique': 'group-a'},
                structural=True,
            )

        self.assertFalse(isValid(manager))
        self.assertFalse(manager._batches)

    def testCommittedPublicationStopsWhenObserverDestroysManager(self):
        for signal in ('subscriptionCommitted', 'subscriptionStateChanged'):
            with self.subTest(signal=signal):
                manager = self._manager({})
                getattr(manager, signal).connect(lambda *_: deleteQObject(manager))
                with (
                    mock.patch.object(manager, '_synchronizeProfiles') as commit,
                    mock.patch.object(manager, '_recordGroupSuccess'),
                ):
                    manager.handleSynchronizationResults(
                        successArgs=[{'unique': 'group-a', 'profiles': ()}]
                    )
                commit.assert_called_once()
                self.assertFalse(isValid(manager))

    def testQueuedCommitStopsAfterObserverDestroysOrShutsDownManager(self):
        for signal in (
            'subscriptionCommitted',
            'subscriptionStateChanged',
            'subscriptionsChanged',
        ):
            for destroy in (False, True):
                with self.subTest(signal=signal, destroy=destroy):
                    manager = self._manager({})
                    context = {'batchId': 1, 'unique': 'group-a'}
                    manager._batches[1] = _SubscriptionBatchState(
                        {('group-a', None)}, False, [], []
                    )
                    errors = []
                    completed = []

                    def retire(*_args):
                        if destroy:
                            deleteQObject(manager)
                        else:
                            manager.shutdown()

                    getattr(manager, signal).connect(retire)
                    manager.updateCompleted.connect(completed.append)
                    with (
                        mock.patch.object(manager, '_synchronizePreparedProfiles'),
                        mock.patch.object(manager, '_recordGroupSuccess'),
                        mock.patch.object(Storage, 'persistSubscriptionGroups'),
                        mock.patch.object(
                            sys,
                            'excepthook',
                            side_effect=lambda *args: errors.append(args),
                        ),
                    ):
                        manager._preparationRelay.completed.emit(
                            SubscriptionPreparationOutcome(
                                1, 'reconcile', context, object()
                            )
                        )
                        processQtEvents()

                    self.assertFalse(errors)
                    self.assertFalse(completed)
                    self.assertFalse(manager._batches)
                    if isValid(manager):
                        deleteQObject(manager)

    def testSyncingObserverCanCancelBeforeAnyNetworkAdmission(self):
        subscriptions = {'group-a': self._subscription(autoupdate='Never')}
        manager = self._manager(subscriptions)
        manager.subscriptionStateChanged.connect(lambda *_: manager.cancelUpdates())

        with (
            mock.patch.object(Storage, 'UserSubs', return_value=subscriptions),
            mock.patch.object(
                Storage,
                'SubscriptionGroup',
                side_effect=lambda unique: SubscriptionGroup.fromMapping(
                    unique, subscriptions[unique]
                ),
            ),
            mock.patch.object(Storage, 'upsertSubscriptionGroups'),
            mock.patch.object(Storage, 'persistSubscriptionGroups'),
            mock.patch.object(manager, 'updateSubsByWebGET') as request,
        ):
            manager.updateSubscriptions(('group-a',))
            request.assert_not_called()
            self.assertFalse(manager._batches)

        manager.shutdown()
        manager.deleteLater()
        processQtEvents()

    def testShutdownCancelsAndWaitsForRunningPreparation(self):
        """Prevent an active worker from delivering any post-shutdown commit."""
        subscriptions = {'group-a': self._subscription(autoupdate='Never')}
        manager = self._manager(subscriptions)
        manager._requestVersions['group-a'] = 1
        context = {
            'unique': 'group-a',
            'webURL': subscriptions['group-a']['webURL'],
            'requestVersion': 1,
            'requestSignature': manager._requestSignature(subscriptions['group-a']),
        }
        started = threading.Event()
        stopped = threading.Event()

        def work(isCancelled):
            started.set()

            while not isCancelled():
                stopped.wait(0.005)

            stopped.set()

        with mock.patch.object(Storage, 'UserSubs', return_value=subscriptions):
            manager._startPreparationJob('probe', context, work)
            self.assertTrue(started.wait(2))
            manager.shutdown()

        self.assertTrue(stopped.is_set())
        self.assertEqual(manager._preparationJobs, {})
        self.assertEqual(manager._preparationPool.activeThreadCount(), 0)

        manager.deleteLater()
        processQtEvents()

    def testSlowShutdownWarnsAndRetainsWorkersUntilTheyFinish(self):
        """Keep synchronous ownership after the warning threshold expires."""
        manager = self._manager({})
        manager.ShutdownWarningMilliseconds = 1
        manager._preparationPool.setMaxThreadCount(1)

        started = threading.Event()
        release = threading.Event()
        queuedStarted = threading.Event()
        manager._handleImportedResult = mock.Mock()

        def work(_isCancelled):
            started.set()
            # Bound the fixture even if the expected diagnostic never arrives.
            release.wait(5)
            return object()

        def afterWarning(*_args):
            self.assertTrue(manager._preparationJobs)
            self.assertTrue(isValid(manager._preparationRelay))
            self.assertTrue(
                all(job.cancelled.is_set() for job in manager._preparationJobs.values())
            )

            release.set()

        try:
            manager._startPreparationJob('import', {}, work)
            self.assertTrue(started.wait(2))

            manager._startPreparationJob('import', {}, lambda _: queuedStarted.set())

            with mock.patch(
                'Furious.Service.SubscriptionManager.logger.warning',
                side_effect=afterWarning,
            ) as warning:
                manager.shutdown()
                manager.shutdown()

                warning.assert_called_once()

            processQtEvents()

            self.assertFalse(queuedStarted.is_set())
            self.assertEqual(manager._preparationPool.activeThreadCount(), 0)
            self.assertEqual(manager._preparationJobs, {})
            self.assertEqual(manager._preparationPayloads, {})
            manager._handleImportedResult.assert_not_called()
        finally:
            release.set()
            manager.shutdown()
            manager.deleteLater()

            processQtEvents()

    def testShutdownRejectsNewWorkAndVersionlessCompletions(self):
        subscriptions = {'group-a': self._subscription()}
        manager = self._manager(subscriptions)
        manager.shutdown()
        with mock.patch.object(Storage, 'persistSubscriptionGroups') as persist:
            self.assertFalse(manager._isCurrentRequest({}))
            self.assertIsNone(manager._startPreparationJob('import', {}, mock.Mock()))
            with mock.patch.object(manager, 'webGET') as request:
                manager.updateSubsByWebGET(webURL='https://invalid.test')
                manager.updateSubscriptions(('group-a',))
                request.assert_not_called()
            with mock.patch.object(Storage, 'UserSubs', return_value=subscriptions):
                manager.refreshAutoUpdates()
            self.assertFalse(manager._autoUpdateTimers['group-a'].isActive())
            persist.assert_not_called()
        manager.deleteLater()
        processQtEvents()

    def testPageNavigationIsPresentationOnlyForAutoUpdateScheduler(self):
        """Keep page show/hide cycles outside scheduler policy ownership."""
        subscriptions = {'group-a': self._subscription()}

        with mock.patch.object(Storage, 'UserSubs', return_value=subscriptions):
            manager = SubscriptionManager()
            timer = manager._autoUpdateTimers['group-a']
            timerId = timer.timerId()

            manager.refreshAutoUpdates = mock.Mock(
                side_effect=AssertionError(
                    'page presentation must not reconcile background schedules'
                )
            )

            serverTable = SimpleNamespace(subsManager=manager)
            page = SubscriptionPage(serverTable)
            placeholder = QtWidgets.QWidget()

            stack = QtWidgets.QStackedWidget()
            stack.addWidget(page)
            stack.addWidget(placeholder)
            stack.show()

            for _index in range(20):
                stack.setCurrentWidget(placeholder)
                processQtEvents(1)
                stack.setCurrentWidget(page)
                processQtEvents(1)

            self.assertIs(page.table.subsManager, manager)
            self.assertIs(manager._autoUpdateTimers['group-a'], timer)
            self.assertEqual(timer.timerId(), timerId)
            self.assertEqual(len(manager._autoUpdateTimers), 1)
            manager.refreshAutoUpdates.assert_not_called()

            stack.deleteLater()
            manager.deleteLater()

        processQtEvents()

    def testManagerDestructionDestroysItsServiceOwnedTimers(self):
        """Let QObject parent ownership release all scheduler resources."""
        subscriptions = {'group-a': self._subscription()}
        manager = self._manager(subscriptions)
        timer = manager._autoUpdateTimers['group-a']
        destroyed = []
        manager.destroyed.connect(lambda *_args: destroyed.append('manager'))
        timer.destroyed.connect(lambda *_args: destroyed.append('timer'))

        manager.deleteLater()
        processQtEvents()

        self.assertFalse(isValid(manager))
        self.assertFalse(isValid(timer))
        self.assertCountEqual(destroyed, ('manager', 'timer'))

    def testMissingSubscriptionRemovalPrunesTimerDuringFullReconciliation(self):
        """Retain removal behavior while unchanged groups remain untouched."""
        subscriptions = {'group-a': self._subscription()}
        manager = self._manager(subscriptions)
        timer = manager._autoUpdateTimers['group-a']

        subscriptions.clear()

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value=subscriptions,
        ):
            manager.refreshAutoUpdates()

        self.assertEqual(manager._autoUpdateTimers, {})
        self.assertFalse(timer.isActive())

        manager.deleteLater()

    def testDeletedSubscriptionVersionIsPrunedAfterItsReplyFinishes(self):
        """Release stale-request bookkeeping after the last exact owner ends."""
        manager = self._manager()
        reply = _AbortableReply()
        manager._requestVersions['deleted-group'] = 4
        manager._activeReplies[reply] = reply
        manager._replySubscriptions[reply] = 'deleted-group'

        with mock.patch(
            'Furious.Service.SubscriptionManager.Storage.UserSubs',
            return_value={},
        ):
            manager._pruneRequestVersion('deleted-group')
            self.assertIn('deleted-group', manager._requestVersions)

            manager._activeReplies.pop(reply)
            manager._replySubscriptions.pop(reply)
            manager._pruneRequestVersion('deleted-group')

        self.assertNotIn('deleted-group', manager._requestVersions)

        manager.deleteLater()