LorenEteval_Furious/tests/test_subscription_manager.py

# 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 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.Window.SubscriptionPage import SubscriptionPage
from Furious.Widget.SubscriptionTableView import SubscriptionTableView

from tests.support import application, processQtEvents

from PySide6 import QtCore, QtTest, QtWidgets

from shiboken6 import isValid

from types import SimpleNamespace
from unittest import TestCase, mock


class _Payload:
    """Expose the minimal QNetworkReply byte-array contract."""

    def __init__(self, value):
        self._value = value

    def data(self):
        return self._value


class _Reply:
    """Provide deterministic response data and failure diagnostics."""

    def __init__(self, value=b'', error='request failed', headers=None):
        self._value = value
        self._error = error
        self._headers = {
            bytes(name).lower(): value for name, value in (headers or {}).items()
        }

    def readAll(self):
        return _Payload(self._value)

    def errorString(self):
        return self._error

    def rawHeader(self, name):
        return self._headers.get(bytes(name).lower(), b'')


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'),
                ('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('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 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 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 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()