Max/LorenEteval_Furious
1
0
Fork 0
mirror of https://github.com/LorenEteval/Furious.git synced 2026-09-22 23:08:08 +03:00
Code Issues Packages Projects Releases Wiki Activity

Harden repository and subscription transactions

Browse Source 
Signed-off-by: Loren Eteval <loren.eteval@proton.me>
This commit is contained in:
Loren Eteval
2026-08-25 06:49:07 +08:00
parent a7bdb398aa
commit 3b78447f72
12 changed files with 504 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Furious/Repository/AGENTS.md
+4
View File
@@ -12,6 +12,10 @@
when practical.
- Load failures retain recoverable data where possible and log actionable context. Never silently replace malformed
explicit configuration with unrelated defaults when doing so loses user intent.
- If persisted data cannot be decoded, automatic shutdown cleanup must not overwrite the unreadable value with an empty
fallback. A deliberate non-empty repository mutation or explicit sync may replace it as a recovery action.
- Prepare fallible reconciliation work—identity calculation, metadata normalization, and the final collection—before
mutating live repository objects. The commit phase should contain only deterministic assignments.
- Singleton repository caches are application-lifetime finite objects and must be reset/sandboxed in tests.
## Verification
Furious/Repository/Routings.py
+22 -6
View File
@@ -23,8 +23,12 @@ from Furious.Frozenlib import *
from Furious.Interface import *
from Furious.Models.Encoding import *
import logging
__all__ = ['UserRoutings']
logger = logging.getLogger(__name__)
registerAppSettings('CustomRouting')
@@ -35,19 +39,25 @@ class UserRoutings(Mixins.CleanupOnExit, StorageBackend):
"""Initialize the UserRoutings."""
super().__init__(*args, **kwargs)
self._restoreFailed = False
def restore():
"""Restore the user routings."""
raw = AppSettings.get('CustomRouting')
if raw is None:
return {}
try:
data = UJSONEncoder.decode(
PyBase64Encoder.decode(AppSettings.get('CustomRouting'))
)
data = UJSONEncoder.decode(PyBase64Encoder.decode(raw))
if isinstance(data, dict):
return data
except Exception:
# Any non-exit exceptions
pass
raise TypeError('routing repository root must be an object')
except Exception:
self._restoreFailed = True
logger.exception('failed to restore persisted routings')
return {}
@@ -59,6 +69,7 @@ class UserRoutings(Mixins.CleanupOnExit, StorageBackend):
'CustomRouting',
PyBase64Encoder.encode(UJSONEncoder.encode(self._data).encode()),
)
self._restoreFailed = False
def data(self) -> dict[str, dict]:
"""Return the live mutable collection managed by this repository."""
@@ -66,4 +77,9 @@ class UserRoutings(Mixins.CleanupOnExit, StorageBackend):
def cleanup(self):
"""Release resources owned by the user routings."""
if self._restoreFailed and not self._data:
logger.warning('preserving unreadable persisted routings during cleanup')
return
self.sync()
Furious/Repository/Servers.py
+27 -4
View File
@@ -29,8 +29,12 @@ from collections.abc import Mapping
from dataclasses import dataclass, field, fields
from typing import Any
import logging
__all__ = ['UserServer', 'UserServers']
logger = logging.getLogger(__name__)
registerAppSettings('Configuration')
@@ -115,14 +119,25 @@ class UserServers(Mixins.CleanupOnExit, StorageBackend):
"""Initialize the UserServers."""
super().__init__(*args, **kwargs)
self._restoreFailed = False
def restore():
"""Restore the user servers."""
raw = AppSettings.get('Configuration')
if raw is None:
return {'model': []}
try:
return UJSONEncoder.decode(
PyBase64Encoder.decode(AppSettings.get('Configuration'))
)
data = UJSONEncoder.decode(PyBase64Encoder.decode(raw))
if isinstance(data, dict) and isinstance(data.get('model', []), list):
return data
raise TypeError('server repository root must contain a model list')
except Exception:
# Any non-exit exceptions
self._restoreFailed = True
logger.exception('failed to restore persisted server configurations')
return {'model': []}
@@ -164,6 +179,7 @@ class UserServers(Mixins.CleanupOnExit, StorageBackend):
).encode()
),
)
self._restoreFailed = False
def data(self) -> list[ServerProfile]:
"""Return the live mutable collection managed by this repository."""
@@ -171,4 +187,11 @@ class UserServers(Mixins.CleanupOnExit, StorageBackend):
def cleanup(self):
"""Release resources owned by the user servers."""
if self._restoreFailed and not self._list:
logger.warning(
'preserving unreadable persisted server configurations during cleanup'
)
return
self.sync()
Furious/Repository/Subscriptions.py
+28 -5
View File
@@ -27,8 +27,12 @@ from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any
import logging
__all__ = ['SubscriptionGroup', 'UserSubs']
logger = logging.getLogger(__name__)
registerAppSettings('CustomSubscription')
@@ -143,20 +147,31 @@ class UserSubs(Mixins.CleanupOnExit, StorageBackend):
"""Initialize the UserSubs."""
super().__init__(*args, **kwargs)
self._restoreFailed = False
def restore():
"""Restore the user subs."""
raw = AppSettings.get('CustomSubscription')
if raw is None:
return {}
try:
return UJSONEncoder.decode(
PyBase64Encoder.decode(AppSettings.get('CustomSubscription'))
)
data = UJSONEncoder.decode(PyBase64Encoder.decode(raw))
if isinstance(data, dict):
return data
raise TypeError('subscription repository root must be an object')
except Exception:
# Any non-exit exceptions
self._restoreFailed = True
logger.exception('failed to restore persisted subscriptions')
return {}
restored = restore()
self._data = restored if isinstance(restored, dict) else {}
self._data = restored
# Normalize legacy URL-only entries into the current group schema. The
# dictionary key remains the stable group ID used by existing profiles.
@@ -176,6 +191,7 @@ class UserSubs(Mixins.CleanupOnExit, StorageBackend):
UJSONEncoder.encode(self._data).encode(),
),
)
self._restoreFailed = False
def data(self) -> dict[str, dict]:
"""Return the live mutable collection managed by this repository."""
@@ -222,4 +238,11 @@ class UserSubs(Mixins.CleanupOnExit, StorageBackend):
def cleanup(self):
"""Release resources owned by the user subs."""
if self._restoreFailed and not self._data:
logger.warning(
'preserving unreadable persisted subscriptions during cleanup'
)
return
self.sync()
Furious/Repository/TunSettings.py
+24 -6
View File
@@ -23,8 +23,12 @@ from Furious.Frozenlib import *
from Furious.Interface import *
from Furious.Models.Encoding import *
import logging
__all__ = ['UserTUNSettings']
logger = logging.getLogger(__name__)
registerAppSettings('CustomTUNSettings')
@@ -35,19 +39,25 @@ class UserTUNSettings(Mixins.CleanupOnExit, StorageBackend):
"""Initialize the UserTUNSettings."""
super().__init__(*args, **kwargs)
self._restoreFailed = False
def restore():
"""Restore the user TUN settings."""
raw = AppSettings.get('CustomTUNSettings')
if raw is None:
return {}
try:
data = UJSONEncoder.decode(
PyBase64Encoder.decode(AppSettings.get('CustomTUNSettings'))
)
data = UJSONEncoder.decode(PyBase64Encoder.decode(raw))
if isinstance(data, dict):
return data
except Exception:
# Any non-exit exceptions
pass
raise TypeError('TUN settings repository root must be an object')
except Exception:
self._restoreFailed = True
logger.exception('failed to restore persisted TUN settings')
return {}
@@ -61,6 +71,7 @@ class UserTUNSettings(Mixins.CleanupOnExit, StorageBackend):
UJSONEncoder.encode(self._data).encode(),
),
)
self._restoreFailed = False
def data(self) -> dict[str, str]:
"""Return the live mutable collection managed by this repository."""
@@ -68,4 +79,11 @@ class UserTUNSettings(Mixins.CleanupOnExit, StorageBackend):
def cleanup(self):
"""Release resources owned by the user TUN settings."""
if self._restoreFailed and not self._data:
logger.warning(
'preserving unreadable persisted TUN settings during cleanup'
)
return
self.sync()
Furious/Service/AGENTS.md
+4
View File
@@ -45,6 +45,10 @@
- `SubscriptionManager` owns download, decoding, filtering, reconciliation, persistence effects, stale-request
rejection, and stable-ID auto-update timers. Subscription views invoke commands and render semantic results; they do
not own this workflow. Remote data is untrusted.
- Subscription reply callbacks stage decoded results only. Persist group status and reconcile profiles after the final
request-version check; one group's failure must not abort other current groups in the same completion batch.
- Treat reconnect/disconnect after subscription reconciliation as a post-commit effect. Failure there must be logged
without reporting the already-committed repository update as rolled back.
## Code review rules
Furious/Service/SubscriptionManager.py
+98 -44
View File
@@ -300,16 +300,83 @@ class SubscriptionManager(HttpGetManager):
-1,
)
AppSettings.set('ActivatedItemIndex', str(newActivatedIndex))
try:
AppSettings.set('ActivatedItemIndex', str(newActivatedIndex))
except Exception:
# Any non-exit exceptions
if wasConnected and activeProfileId:
if newActivatedIndex < 0 and activeWasManagedByGroup:
controller.startDisconnection()
elif activeProfileId in result.changedProfileIds:
controller.startReconnection()
# Profile reconciliation has committed. The legacy row-index
# setting is derived compatibility state, not part of that commit.
logger.exception(
'failed to persist the active profile index after '
f'synchronizing subscription {unique!r}'
)
try:
if wasConnected and activeProfileId:
if newActivatedIndex < 0 and activeWasManagedByGroup:
controller.startDisconnection()
elif activeProfileId in result.changedProfileIds:
controller.startReconnection()
except Exception:
# Any non-exit exceptions
# Reconciliation has committed at this point. A controller-side
# follow-up failure is not a failed or rolled-back synchronization.
logger.exception(
f'failed to apply connection effects after synchronizing '
f'subscription {unique!r}'
)
return result
@staticmethod
def _recordGroupFailure(param):
"""Best-effort persist one current request's terminal failure state."""
try:
group = Storage.SubscriptionGroup(param.get('unique', ''))
if group is None:
return
group.lastSyncStatus = 'error'
group.lastSyncError = str(param.get('error', ''))
Storage.upsertSubscriptionGroup(group)
except Exception:
# Any non-exit exceptions
logger.exception(
'failed to record synchronization failure for subscription '
f'{param.get("unique", "")!r}'
)
@staticmethod
def _recordGroupSuccess(param, result):
"""Best-effort persist one successfully committed synchronization state."""
try:
group = Storage.SubscriptionGroup(param.get('unique', ''))
if group is None:
return
group.lastUpdated = (
datetime.datetime.now().astimezone().isoformat(timespec='seconds')
)
group.lastDecoderId = param.get('decoderId', '')
group.lastSyncStatus = 'success'
group.lastSyncError = ''
group.profileCount = len(result.profileIds)
Storage.upsertSubscriptionGroup(group)
except Exception:
# Any non-exit exceptions
logger.exception(
'failed to record synchronization success for subscription '
f'{param.get("unique", "")!r}'
)
def handleSynchronizationResults(self, **kwargs):
"""Commit successful group-scoped synchronization results."""
successArgs = kwargs.pop('successArgs', list())
@@ -328,24 +395,39 @@ class SubscriptionManager(HttpGetManager):
continue
result = self._synchronizeProfiles(param['unique'], param['profiles'])
try:
result = self._synchronizeProfiles(
param['unique'],
param['profiles'],
)
except Exception as ex:
# Any non-exit exceptions
error = str(ex) or type(ex).__name__
logger.exception(
f'failed to synchronize subscription '
f'{param.get("unique", "")!r}'
)
failed = {**param, 'error': error}
committedFailure.append(failed)
self._recordGroupFailure(failed)
continue
param['syncResult'] = result
committedSuccess.append(param)
group = Storage.SubscriptionGroup(param['unique'])
if group is not None:
group.lastDecoderId = param.get('decoderId', '')
group.lastSyncStatus = 'success'
group.lastSyncError = ''
group.profileCount = len(result.profileIds)
Storage.upsertSubscriptionGroup(group)
self._recordGroupSuccess(param, result)
for param in failureArgs:
if self._isCurrentRequest(param):
committedFailure.append(param)
self._recordGroupFailure(param)
else:
logger.info(
f'ignore stale subscription failure for '
@@ -415,13 +497,6 @@ class SubscriptionManager(HttpGetManager):
if result is None or not result.profiles:
failureArgs.append({'error': 'UnsupportedSubscriptionFormat', **kwargs})
group = Storage.SubscriptionGroup(kwargs.get('unique', ''))
if group is not None:
group.lastSyncStatus = 'error'
group.lastSyncError = 'UnsupportedSubscriptionFormat'
Storage.upsertSubscriptionGroup(group)
return
@@ -435,19 +510,6 @@ class SubscriptionManager(HttpGetManager):
{**kwargs, 'profiles': result.profiles, 'decoderId': result.decoderId}
)
unique = kwargs.get('unique', '')
if unique in Storage.UserSubs():
group = Storage.SubscriptionGroup(unique)
if group is not None:
group.lastUpdated = (
datetime.datetime.now().astimezone().isoformat(timespec='seconds')
)
group.lastDecoderId = result.decoderId
Storage.upsertSubscriptionGroup(group)
def failureCallback(self, networkReply, **kwargs):
"""Record one failed subscription response."""
if not self._isCurrentRequest(kwargs):
@@ -463,14 +525,6 @@ class SubscriptionManager(HttpGetManager):
failureArgs.append({'error': error, **kwargs})
group = Storage.SubscriptionGroup(kwargs.get('unique', ''))
if group is not None:
group.lastSyncStatus = 'error'
group.lastSyncError = error
Storage.upsertSubscriptionGroup(group)
def updateSubsByWebGET(self, **kwargs):
"""Start one configured subscription request."""
url = kwargs.get('webURL', '')
Furious/Service/SubscriptionSync.py
+54 -20
View File
@@ -21,6 +21,8 @@ from __future__ import annotations
from Furious.Models import ServerProfile, profileConnectionFingerprint
import copy
from dataclasses import dataclass
__all__ = ['SubscriptionSyncResult', 'SubscriptionSynchronizer']
@@ -52,9 +54,10 @@ class SubscriptionSynchronizer:
)
@staticmethod
def _ensureKeys(profiles: list[ServerProfile], groupId: str):
"""Migrate legacy group profiles to deterministic occurrence keys."""
def _keyAssignments(profiles: list[ServerProfile], groupId: str):
"""Plan deterministic keys for legacy profiles without mutating them."""
occurrences = {}
assignments = []
for profile in profiles:
metadata = profile.metadata
@@ -71,10 +74,20 @@ class SubscriptionSynchronizer:
baseIdentity = f'config:{profileConnectionFingerprint(profile)}'
occurrence = occurrences.get(baseIdentity, 0)
occurrences[baseIdentity] = occurrence + 1
metadata.subscriptionProfileKey = (
key = (
baseIdentity if occurrence == 0 else f'{baseIdentity}#{occurrence + 1}'
)
assignments.append((profile, key))
return assignments
@classmethod
def _ensureKeys(cls, profiles: list[ServerProfile], groupId: str):
"""Migrate legacy group profiles to deterministic occurrence keys."""
for profile, key in cls._keyAssignments(profiles, groupId):
profile.metadata.subscriptionProfileKey = key
def reconcile(
self,
profiles: list[ServerProfile],
@@ -87,7 +100,8 @@ class SubscriptionSynchronizer:
incoming = list(incomingProfiles)
self._ensureKeys(profiles, groupId)
legacyKeyAssignments = self._keyAssignments(profiles, groupId)
legacyKeys = {id(profile): key for profile, key in legacyKeyAssignments}
managedIndexes = [
index
@@ -96,16 +110,21 @@ class SubscriptionSynchronizer:
]
insertionIndex = min(managedIndexes) if managedIndexes else len(profiles)
existingByKey = {
profiles[index].metadata.subscriptionProfileKey: profiles[index]
legacyKeys.get(
id(profiles[index]),
profiles[index].metadata.subscriptionProfileKey,
): profiles[index]
for index in managedIndexes
}
synchronized = []
incomingMetadata = []
existingUpdates = []
updated = 0
added = 0
changedProfileIds = []
for profile in incoming:
metadata = profile.metadata
metadata = copy.deepcopy(profile.metadata)
metadata.subscriptionSource = groupId
metadata.subscriptionManaged = True
@@ -119,31 +138,27 @@ class SubscriptionSynchronizer:
if existing is None:
added += 1
synchronized.append(profile)
incomingMetadata.append((profile, metadata))
continue
oldFingerprint = profileConnectionFingerprint(existing)
newFingerprint = profileConnectionFingerprint(profile)
for fieldName in self.LocalMetadataFields:
setattr(metadata, fieldName, getattr(existing.metadata, fieldName))
metadata.profileId = existing.metadata.profileId
existing.connection = profile.connection
existing.metadata = metadata
existing.deleted = False
synchronized.append(existing)
existingUpdates.append((existing, profile.connection, metadata))
updated += 1
if oldFingerprint != profileConnectionFingerprint(existing):
if oldFingerprint != newFingerprint:
changedProfileIds.append(metadata.profileId)
removedProfiles = tuple(existingByKey.values())
for profile in removedProfiles:
profile.deleted = True
unmanagedOrOther = [
profile
for profile in profiles
@@ -151,17 +166,13 @@ class SubscriptionSynchronizer:
profile.itemSubscription == groupId and profile.itemSubscriptionManaged
)
]
profiles[:] = (
finalProfiles = (
unmanagedOrOther[:insertionIndex]
+ synchronized
+ unmanagedOrOther[insertionIndex:]
)
for index, profile in enumerate(profiles):
profile.index = index
profile.deleted = False
return SubscriptionSyncResult(
result = SubscriptionSyncResult(
groupId=groupId,
added=added,
updated=updated,
@@ -172,3 +183,26 @@ class SubscriptionSynchronizer:
),
changedProfileIds=tuple(changedProfileIds),
)
# Everything above is preparation and may fail. The assignments below
# are the commit point and preserve existing profile object identities.
for profile, key in legacyKeyAssignments:
profile.metadata.subscriptionProfileKey = key
for profile, metadata in incomingMetadata:
profile.metadata = metadata
for profile, connection, metadata in existingUpdates:
profile.connection = connection
profile.metadata = metadata
for profile in removedProfiles:
profile.deleted = True
profiles[:] = finalProfiles
for index, profile in enumerate(finalProfiles):
profile.index = index
profile.deleted = False
return result
tests/AGENTS.md
+2
View File
@@ -15,6 +15,8 @@
- Test public behavior and architectural contracts, not implementation trivia. Cover success, validation failure,
timeout, cancel, partial/stale result, cleanup, and backward-compatible persisted input.
- For multi-step mutations, inject a failure immediately before the commit point and assert that live objects and
persisted bytes remain unchanged; separately test post-commit side-effect failures without pretending rollback.
- Use canonical `CoreRuntime` vocabulary in runtime factories, registry fixtures, and traffic-statistics providers.
- Separate persisted configuration from runtime-copy assertions. Normal connection, generated native TUN, preserved user
TUN, and proxy-only stripping must not share a helper that erases their differences.
tests/test_repository_contracts.py
+40
View File
@@ -23,7 +23,9 @@ from Furious.Frozenlib import AppSettings
from Furious.Models import CoreConfiguration, ServerProfile
from Furious.Models.Encoding import PyBase64Encoder, UJSONEncoder
from Furious.Repository.Routings import UserRoutings
from Furious.Repository.Servers import UserServers
from Furious.Repository.Storage import Storage
from Furious.Repository.Subscriptions import UserSubs
from Furious.Repository.TunSettings import UserTUNSettings
from tests.support import application, isolatedSettings
@@ -182,6 +184,44 @@ class RepositoryContractTest(unittest.TestCase):
self.assertEqual(Storage.UserActivatedItemIndex(), -1)
self.assertEqual(settings.value('ActivatedItemIndex'), 'not-an-index')
def testFailedRestoreIsNotOverwrittenByAutomaticCleanup(self):
"""Preserve recoverable persisted bytes when decoding fails at startup."""
corrupt = b'eA=='
with isolatedSettings():
for setting, repositoryType, emptyValue in (
('Configuration', UserServers, []),
('CustomSubscription', UserSubs, {}),
('CustomRouting', UserRoutings, {}),
('CustomTUNSettings', UserTUNSettings, {}),
):
with self.subTest(setting=setting):
AppSettings.set(setting, corrupt)
repository = repositoryType()
self.assertEqual(repository.data(), emptyValue)
repository.cleanup()
self.assertEqual(AppSettings.get(setting), corrupt)
def testExplicitMutationCanReplaceAFailedRestoreFallback(self):
"""Allow a deliberate repository change to recover unreadable storage."""
corrupt = b'eA=='
with isolatedSettings():
AppSettings.set('CustomTUNSettings', corrupt)
repository = UserTUNSettings()
repository.data()['interfaceName'] = 'replacement-tun'
repository.cleanup()
self.assertNotEqual(AppSettings.get('CustomTUNSettings'), corrupt)
self.assertEqual(
UserTUNSettings().data(),
{'interfaceName': 'replacement-tun'},
)
if __name__ == '__main__':
unittest.main()
tests/test_subscription_manager.py
+169
View File
@@ -208,6 +208,172 @@ class SubscriptionManagerTest(TestCase):
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 = []
manager.updateCompleted.connect(completed.append)
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'])
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()
@@ -225,6 +391,7 @@ class SubscriptionManagerTest(TestCase):
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()
@@ -263,6 +430,7 @@ class SubscriptionManagerTest(TestCase):
self.assertEqual(manager._autoUpdateTimers, {})
self.assertFalse(timer.isActive())
manager.deleteLater()
def testDeletedSubscriptionVersionIsPrunedAfterItsReplyFinishes(self):
@@ -285,4 +453,5 @@ class SubscriptionManagerTest(TestCase):
manager._pruneRequestVersion('deleted-group')
self.assertNotIn('deleted-group', manager._requestVersions)
manager.deleteLater()
tests/test_subscription_sync.py
+32
View File
@@ -23,6 +23,7 @@ from Furious.Models import CoreConfiguration, ServerProfile
from Furious.Service.SubscriptionSync import SubscriptionSynchronizer
import unittest
from unittest import mock
def profile(
@@ -137,6 +138,37 @@ class SubscriptionSynchronizerTest(unittest.TestCase):
self.assertEqual(incoming.itemSubscription, '')
self.assertFalse(incoming.itemSubscriptionManaged)
def testPreparationFailureLeavesExistingAndIncomingProfilesUntouched(self):
"""Do not expose partial key or metadata changes before the commit point."""
retained = profile(
'Retained',
'old.example',
source='group',
managed=True,
)
incoming = profile('Incoming', 'new.example')
profiles = [retained]
originalMetadata = retained.metadata.toMapping()
originalConnection = retained.connection.deepcopy()
with mock.patch(
'Furious.Service.SubscriptionSync.profileConnectionFingerprint',
side_effect=('legacy-key', RuntimeError('injected preparation failure')),
):
with self.assertRaisesRegex(RuntimeError, 'injected preparation failure'):
SubscriptionSynchronizer().reconcile(
profiles,
[incoming],
'group',
)
self.assertEqual(profiles, [retained])
self.assertEqual(retained.metadata.toMapping(), originalMetadata)
self.assertEqual(retained.connection, originalConnection)
self.assertFalse(retained.deleted)
self.assertEqual(incoming.itemSubscription, '')
self.assertFalse(incoming.itemSubscriptionManaged)
if __name__ == '__main__':
unittest.main()