From 7479469ceb533f56e86b6657a7276c8f7b552aa4 Mon Sep 17 00:00:00 2001 From: Loren Eteval Date: Mon, 17 Aug 2026 10:07:17 +0800 Subject: [PATCH] Expand application test coverage Signed-off-by: Loren Eteval --- tests/README.md | 69 ++++--- tests/support.py | 113 +++++++--- tests/test_controllers.py | 96 ++++++++- tests/test_isolation_and_navigation.py | 216 +++++++++++++++++++ tests/test_metrics_behavior.py | 275 +++++++++++++++++++++++++ tests/test_plugin_architecture.py | 150 ++++++++++++++ tests/test_process_stress.py | 106 ++++++++++ tests/test_qt_lifetime.py | 12 ++ tests/test_qt_stress.py | 107 +++++++++- tests/test_shadowsocks_uri.py | 138 +++++++++++++ tests/test_subscription_sync.py | 142 +++++++++++++ tests/test_ui_behavior.py | 172 ++++++++++++++++ 12 files changed, 1546 insertions(+), 50 deletions(-) create mode 100644 tests/test_isolation_and_navigation.py create mode 100644 tests/test_metrics_behavior.py create mode 100644 tests/test_process_stress.py create mode 100644 tests/test_shadowsocks_uri.py create mode 100644 tests/test_subscription_sync.py diff --git a/tests/README.md b/tests/README.md index 60c8ff7c..2c11f5e3 100644 --- a/tests/README.md +++ b/tests/README.md @@ -13,11 +13,15 @@ system proxy, TUN, routing, update network clients, or real proxy cores. | Configuration, profiles, migration, repositories | `test_models_and_services.py` | | Plugin registration, capability dispatch, factories, rollback | `test_plugin_architecture.py` | | Controller state and error transitions with injected runtimes | `test_controllers.py` | -| SOCKS URI codec and import boundaries | `test_socks_uri.py` | +| SOCKS/SIP002 Shadowsocks codecs and generated round trips | `test_socks_uri.py`, `test_shadowsocks_uri.py` | +| Subscription-group reconciliation and ownership isolation | `test_subscription_sync.py` | | External process launch, output, shutdown, threads, TUN metadata | `test_external_core.py` | -| Editor mappings, lazy log rendering, routing/message-box behavior | `test_ui_behavior.py` | +| Rolling metrics, stable buckets, lazy rendering, and hover | `test_metrics_behavior.py` | +| Settings sandbox and navigation overlay behavior | `test_isolation_and_navigation.py` | +| Editor mappings, lazy log rendering, routing/message-box/connection UI | `test_ui_behavior.py` | | Direct Qt ownership and destruction across independent UI families | `test_qt_lifetime.py` | -| Batched Qt object, Python allocation, and RSS trends | `test_qt_stress.py` | +| Batched real/probe Qt object, handle, Python allocation, and RSS trends | `test_qt_stress.py` | +| Repeated harmless subprocess, pipe, thread, handle, and RSS trends | `test_process_stress.py` | The lifecycle tests classify `AppQTransientDialog`, protocol/plugin editors, routing dialogs, subscription editors, message boxes, QR windows, and TUN @@ -29,33 +33,48 @@ than by starting the production application runtime. ## Commands -From the repository root on Windows PowerShell: +From the repository root, select the offscreen Qt platform for your shell. + +Windows PowerShell: ```powershell $env:QT_QPA_PLATFORM = 'offscreen' - -# Everything -.\.venv-python313\Scripts\python.exe -m unittest discover -s tests -v - -# Fast logic, persistence, plugin, controller, process, codec, and UI behavior -.\.venv-python313\Scripts\python.exe -m unittest ` - tests.test_models_and_services ` - tests.test_plugin_architecture ` - tests.test_controllers ` - tests.test_external_core ` - tests.test_socks_uri ` - tests.test_ui_behavior -v - -# Direct Qt destruction/lifetime checks -.\.venv-python313\Scripts\python.exe -m unittest tests.test_qt_lifetime -v - -# Hundreds-of-cycles allocation/RSS trend check -.\.venv-python313\Scripts\python.exe -m unittest tests.test_qt_stress -v ``` -Equivalent commands work on Linux/macOS after replacing the virtual-environment -executable path with the platform's Python path. No external network access or -installed Xray/Hysteria executable is required. +Windows Command Prompt: + +```cmd +set QT_QPA_PLATFORM=offscreen +``` + +Linux, macOS, and other Unix-compatible shells: + +```sh +export QT_QPA_PLATFORM=offscreen +``` + +Then run the desired test tier. These commands use the active Python +environment, so activate the project's virtual environment first when needed. + +```text +# Everything +python -m unittest discover -s tests -v + +# Fast logic, persistence, plugin, controller, codec, and UI behavior +python -m unittest tests.test_models_and_services tests.test_plugin_architecture tests.test_controllers tests.test_subscription_sync tests.test_socks_uri tests.test_shadowsocks_uri tests.test_metrics_behavior tests.test_isolation_and_navigation tests.test_ui_behavior -v + +# Direct Qt/process integration and destruction/lifetime checks +python -m unittest tests.test_external_core tests.test_qt_lifetime -v + +# Explicit slow stress tier +python -m unittest tests.test_qt_stress tests.test_process_stress -v + +# Order-independence spot check (reverse the module order, then run all tests) +python -m unittest tests.test_ui_behavior tests.test_isolation_and_navigation tests.test_metrics_behavior tests.test_subscription_sync tests.test_controllers tests.test_plugin_architecture tests.test_models_and_services -v +python -m unittest discover -s tests -v +``` + +No external network access or installed Xray/Hysteria executable is required. ## Packaged-build smoke procedure diff --git a/tests/support.py b/tests/support.py index ef1ce1f5..fa70030a 100644 --- a/tests/support.py +++ b/tests/support.py @@ -26,6 +26,7 @@ import os import gc import time import uuid +import atexit import ctypes import tempfile import weakref @@ -96,6 +97,47 @@ class TestApplication(QApplication): _application = None +_settingsDirectory = None + + +def _initializeSettingsSandbox(): + """Install one process-lifetime QSettings root owned by the test suite.""" + global _settingsDirectory + + if _settingsDirectory is not None: + return Path(_settingsDirectory.name) + + _settingsDirectory = tempfile.TemporaryDirectory( + prefix='furious-tests-settings-root-' + ) + atexit.register(_settingsDirectory.cleanup) + + root = Path(_settingsDirectory.name).resolve() + + QtCore.QSettings.setDefaultFormat(QtCore.QSettings.Format.IniFormat) + QtCore.QSettings.setPath( + QtCore.QSettings.Format.IniFormat, + QtCore.QSettings.Scope.UserScope, + str(root), + ) + + return root + + +def settingsSandboxPath() -> Path: + """Return the suite-owned root containing every test QSettings file.""" + return _initializeSettingsSandbox() + + +def assertIsolatedSettings(settings: QtCore.QSettings): + """Raise when *settings* resolves outside the suite-owned sandbox.""" + settingsPath = Path(settings.fileName()).resolve() + sandbox = settingsSandboxPath() + + if settingsPath != sandbox and sandbox not in settingsPath.parents: + raise AssertionError( + f'test QSettings escaped sandbox: {settingsPath} (root: {sandbox})' + ) def application() -> QApplication: @@ -105,6 +147,7 @@ def application() -> QApplication: current = QApplication.instance() if current is None: + _initializeSettingsSandbox() _application = TestApplication() current = _application @@ -226,38 +269,62 @@ def currentRSS() -> int | None: return None +def currentNativeHandleCount() -> int | None: + """Return process handles/descriptors when the platform exposes them safely.""" + if os.name == 'nt': + count = ctypes.c_ulong() + + kernel32 = ctypes.WinDLL('kernel32', use_last_error=True) + kernel32.GetCurrentProcess.argtypes = () + kernel32.GetCurrentProcess.restype = ctypes.c_void_p + kernel32.GetProcessHandleCount.argtypes = ( + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_ulong), + ) + kernel32.GetProcessHandleCount.restype = ctypes.c_int + + if kernel32.GetProcessHandleCount( + kernel32.GetCurrentProcess(), + ctypes.byref(count), + ): + return int(count.value) + + return None + + descriptors = Path('/proc/self/fd') + + if descriptors.exists(): + return sum(1 for _path in descriptors.iterdir()) + + return None + + @contextmanager def isolatedSettings(): """Route every QSettings read/write to one temporary test namespace.""" app = application() + oldOrganization, oldApplication = ( + app.organizationName(), + app.applicationName(), + ) - with tempfile.TemporaryDirectory(prefix='furious-tests-settings-') as directory: - oldOrganization, oldApplication = ( - app.organizationName(), - app.applicationName(), - ) + namespace = uuid.uuid4().hex - namespace = uuid.uuid4().hex + app.setOrganizationName(f'Furious Tests {namespace}') + app.setApplicationName(f'Furious Tests {namespace}') - QtCore.QSettings.setDefaultFormat(QtCore.QSettings.Format.IniFormat) - QtCore.QSettings.setPath( - QtCore.QSettings.Format.IniFormat, - QtCore.QSettings.Scope.UserScope, - directory, - ) + settings = QtCore.QSettings() - app.setOrganizationName(f'Furious Tests {namespace}') - app.setApplicationName(f'Furious Tests {namespace}') + assertIsolatedSettings(settings) - settings = QtCore.QSettings() + settings.clear() + settings.sync() + + try: + yield settings + finally: settings.clear() settings.sync() - try: - yield settings - finally: - settings.clear() - settings.sync() - - app.setOrganizationName(oldOrganization) - app.setApplicationName(oldApplication) + app.setOrganizationName(oldOrganization) + app.setApplicationName(oldApplication) diff --git a/tests/test_controllers.py b/tests/test_controllers.py index 9391add6..0d32783c 100644 --- a/tests/test_controllers.py +++ b/tests/test_controllers.py @@ -49,10 +49,19 @@ class ControllerConfiguration(ConfigFactory): class FixtureCoreManager: """Record lifecycle calls without launching a subprocess or changing routes.""" - def __init__(self, *, startResult=True, startError=''): + def __init__( + self, + *, + startResult=True, + startError='', + startException=None, + stopException=None, + ): """Initialize deterministic start behavior.""" self.startResult = startResult self.lastStartError = startError + self.startException = startException + self.stopException = stopException self.processesPool = [] self.startCalls = [] self.stopCalls = 0 @@ -61,6 +70,9 @@ class FixtureCoreManager: """Record one requested start and return the configured result.""" self.startCalls.append((configuration, kwargs)) + if self.startException is not None: + raise self.startException + return self.startResult def stopAll(self): @@ -68,6 +80,9 @@ class FixtureCoreManager: self.stopCalls += 1 self.processesPool.clear() + if self.stopException is not None: + raise self.stopException + class FixtureUpdatesManager: """Record update hooks without contacting any update service.""" @@ -243,6 +258,85 @@ class ConnectionControllerTest(unittest.TestCase): controller.deleteLater() + def testDuplicateStartIsRejectedWithoutASecondRuntime(self): + """Keep one lifecycle owner while already connected.""" + with isolatedSettings(): + core = FixtureCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.set'), + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + mock.patch.object(controller, '_runPostConnectTasksOnce'), + ): + self.assertTrue(controller.startConnection(self.profile)) + self.assertFalse(controller.startConnection(self.profile)) + self.assertEqual(len(core.startCalls), 1) + self.assertTrue(controller.startDisconnection()) + + controller.deleteLater() + + def testStartAndProxyExceptionsReturnToStableDisconnectedState(self): + """Clean every partially acquired resource after injected failures.""" + for core, proxySideEffect in ( + (FixtureCoreManager(startException=RuntimeError('start')), None), + (FixtureCoreManager(), RuntimeError('proxy')), + ): + with self.subTest( + startException=core.startException, + proxyException=proxySideEffect, + ), isolatedSettings(): + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set', + side_effect=proxySideEffect, + ), + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.off' + ), + mock.patch.object(controller, '_runPostConnectTasksOnce'), + ): + self.assertFalse(controller.startConnection(self.profile)) + + self.assertEqual(controller.state, ConnectionState.Disconnected) + self.assertIsNone(controller.activeConfiguration) + self.assertEqual(core.stopCalls, 1) + self.assertFalse(controller._actionTimer.isActive()) + + controller.deleteLater() + + def testStopExceptionCannotStrandConnectionControls(self): + """Complete disconnect even when the runtime reports cleanup failure.""" + with isolatedSettings(): + core = FixtureCoreManager(stopException=RuntimeError('stop')) + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.set'), + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + mock.patch.object(controller, '_runPostConnectTasksOnce'), + ): + self.assertTrue(controller.startConnection(self.profile)) + self.assertTrue(controller.startDisconnection()) + + self.assertEqual(controller.state, ConnectionState.Disconnected) + self.assertIsNone(controller.activeConfiguration) + self.assertTrue(controller.interactionEnabled) + self.assertFalse(controller._actionTimer.isActive()) + + controller.deleteLater() + if __name__ == '__main__': unittest.main() diff --git a/tests/test_isolation_and_navigation.py b/tests/test_isolation_and_navigation.py new file mode 100644 index 00000000..e2d7bb0a --- /dev/null +++ b/tests/test_isolation_and_navigation.py @@ -0,0 +1,216 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# 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 . + +"""Verify test-process isolation and Fluent navigation behavior.""" + +from __future__ import annotations + +from Furious.Widget.NavigationView import NavigationView + +from PySide6 import QtCore +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QPushButton, QVBoxLayout, QWidget + +from pathlib import Path + +from tests.support import ( + application, + assertIsolatedSettings, + collectAtBoundary, + isolatedSettings, + processQtEvents, + settingsSandboxPath, +) + +import unittest + + +class TestHarnessIsolationTest(unittest.TestCase): + """Prove QSettings never resolves to a production Furious namespace.""" + + @classmethod + def setUpClass(cls): + """Create the suite-owned QApplication and settings root.""" + application() + + def testNestedSettingsNamespacesRestoreProcessMetadata(self): + """Keep nested tests independent and restore their caller's namespace.""" + app = application() + originalIdentity = (app.organizationName(), app.applicationName()) + + with isolatedSettings() as outer: + outerIdentity = (app.organizationName(), app.applicationName()) + outer.setValue('fixture', 'outer') + + assertIsolatedSettings(outer) + + self.assertIn(settingsSandboxPath(), Path(outer.fileName()).parents) + + with isolatedSettings() as inner: + innerIdentity = (app.organizationName(), app.applicationName()) + + assertIsolatedSettings(inner) + + self.assertNotEqual(innerIdentity, outerIdentity) + self.assertIsNone(inner.value('fixture')) + + inner.setValue('fixture', 'inner') + + self.assertEqual( + (app.organizationName(), app.applicationName()), + outerIdentity, + ) + self.assertEqual(outer.value('fixture'), 'outer') + + self.assertEqual( + (app.organizationName(), app.applicationName()), + originalIdentity, + ) + + def testDefaultSettingsObjectRemainsInsideSuiteSandbox(self): + """Keep unscoped test helpers away from the user's real settings file.""" + settings = QtCore.QSettings() + + assertIsolatedSettings(settings) + + self.assertIn(settingsSandboxPath(), Path(settings.fileName()).parents) + + +class NavigationBehaviorTest(unittest.TestCase): + """Protect overlay geometry, outside-click dismissal, and page persistence.""" + + @classmethod + def setUpClass(cls): + """Create the process-wide headless QApplication.""" + application() + + def setUp(self): + """Build one isolated navigation view with top and bottom pages.""" + self.navigation = NavigationView() + self.navigation.resize(900, 600) + + self.homePage = QWidget() + + homeLayout = QVBoxLayout(self.homePage) + + self.outsideButton = QPushButton('Outside target', self.homePage) + + homeLayout.addWidget(self.outsideButton) + + self.settingsPage = QWidget() + + self.navigation.addPage('home', self.homePage, 'Home', 'house-door.svg') + self.navigation.addPage( + 'settings', + self.settingsPage, + 'Settings', + 'gear-wide-connected.svg', + placement='bottom', + ) + self.navigation.show() + + processQtEvents() + + def tearDown(self): + """Destroy the complete page tree between tests.""" + self.navigation.close() + self.navigation.deleteLater() + + collectAtBoundary() + + def testExpandedPanelOverlaysWithoutMovingContent(self): + """Expand only the panel geometry while the page stack remains fixed.""" + collapsedGeometry = QtCore.QRect(self.navigation.pageStack.geometry()) + + self.navigation.setExpanded(True, animated=False) + + processQtEvents() + + self.assertTrue(self.navigation.isExpanded()) + self.assertEqual( + self.navigation.navigationPanel.width(), + self.navigation.ExpandedWidth, + ) + self.assertEqual(self.navigation.pageStack.geometry(), collapsedGeometry) + self.assertEqual( + self.navigation.navigationRail.width(), + self.navigation.CollapsedWidth, + ) + + def testOutsideClickCollapsesAndStillReachesTarget(self): + """Dismiss the temporary overlay without consuming the original click.""" + clicks = [] + + self.outsideButton.clicked.connect(lambda: clicks.append(True)) + self.navigation.setExpanded(True, animated=False) + + QTest.mouseClick( + self.outsideButton, + QtCore.Qt.MouseButton.LeftButton, + ) + + processQtEvents() + + self.assertEqual(clicks, [True]) + self.assertFalse(self.navigation.isExpanded()) + self.assertFalse(self.navigation._outsideClickFilterInstalled) + + def testInsideNavigationClickKeepsOverlayExpandedAndSwitchesPage(self): + """Allow normal navigation interaction without flyout dismissal.""" + self.navigation.setExpanded(True, animated=False) + + settingsButton = self.navigation._pages['settings'].button + + QTest.mouseClick( + settingsButton, + QtCore.Qt.MouseButton.LeftButton, + ) + + processQtEvents() + + self.assertTrue(self.navigation.isExpanded()) + self.assertEqual(self.navigation.currentPageId(), 'settings') + self.assertIs(self.navigation.pageStack.currentWidget(), self.settingsPage) + + def testRepeatedExpansionAndPageSwitchingReuseOwnedObjects(self): + """Avoid duplicate pages, animations, buttons, or event-filter state.""" + animation = self.navigation._widthAnimation + pages = tuple(page.widget for page in self.navigation._pages.values()) + buttons = tuple(page.button for page in self.navigation._pages.values()) + + for index in range(75): + self.navigation.setExpanded(True, animated=False) + self.navigation.setCurrentPage('settings' if index % 2 else 'home') + self.navigation.setExpanded(False, animated=False) + + processQtEvents() + + self.assertIs(self.navigation._widthAnimation, animation) + self.assertEqual( + tuple(page.widget for page in self.navigation._pages.values()), + pages, + ) + self.assertEqual( + tuple(page.button for page in self.navigation._pages.values()), + buttons, + ) + self.assertEqual(self.navigation.pageStack.count(), 2) + self.assertFalse(self.navigation._outsideClickFilterInstalled) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_metrics_behavior.py b/tests/test_metrics_behavior.py new file mode 100644 index 00000000..bb163f43 --- /dev/null +++ b/tests/test_metrics_behavior.py @@ -0,0 +1,275 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# 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 . + +"""Protect rolling metrics history, aggregation, lazy rendering, and hover.""" + +from __future__ import annotations + +from Furious.Service.MetricsDataManager import ( + DOWNLOAD_SPEED_METRIC, + DOWNLOAD_USAGE_METRIC, + UPLOAD_SPEED_METRIC, + MetricPoint, + MetricsDataManager, +) +from Furious.Service.TrafficStatsManager import formatTrafficSpeed +from Furious.Widget.MetricsGraph import MetricsGraphWidget +from Furious.Window.MetricsPage import MetricsPage + +from PySide6 import QtCore + +from tests.support import application, collectAtBoundary, processQtEvents, waitFor + +import math +import time +import unittest + + +class MetricsTimeSeriesBehaviorTest(unittest.TestCase): + """Verify raw history and display buckets remain distinct and deterministic.""" + + def testRawSamplesRemainImmutableAsVisibleWindowMoves(self): + """Advance now without rewriting timestamps or historical metric values.""" + manager = MetricsDataManager(maximumHistorySeconds=1000) + manager.recordSample({DOWNLOAD_SPEED_METRIC: 10}, sampledAt=100) + manager.recordSample({DOWNLOAD_SPEED_METRIC: 30}, sampledAt=104) + manager.recordSample({DOWNLOAD_SPEED_METRIC: 50}, sampledAt=111) + + rawBefore = manager.rawSamples() + valuesBefore = tuple( + sample.values[DOWNLOAD_SPEED_METRIC] for sample in rawBefore + ) + firstWindow = manager.series( + DOWNLOAD_SPEED_METRIC, + 20, + granularitySeconds=10, + now=115, + ) + secondWindow = manager.series( + DOWNLOAD_SPEED_METRIC, + 20, + granularitySeconds=10, + now=118, + ) + + self.assertEqual(manager.rawSamples(), rawBefore) + self.assertEqual( + tuple( + sample.values[DOWNLOAD_SPEED_METRIC] for sample in manager.rawSamples() + ), + valuesBefore, + ) + self.assertEqual(firstWindow, secondWindow) + self.assertEqual( + tuple((point.sampledAt, point.value) for point in firstWindow), + ((104.0, 20.0), (111.0, 50.0)), + ) + self.assertGreater( + MetricsGraphWidget._normalizedTimePosition(104, 20, 115), + MetricsGraphWidget._normalizedTimePosition(104, 20, 118), + ) + + def testWindowEntryExitAndTimestampAlignedBuckets(self): + """Drop old points naturally while stable absolute buckets retain values.""" + manager = MetricsDataManager(maximumHistorySeconds=1000) + + for sampledAt, value in ((1, 10), (9, 30), (10, 50), (19, 70), (20, 90)): + manager.recordSample( + { + DOWNLOAD_SPEED_METRIC: value, + DOWNLOAD_USAGE_METRIC: value * 100, + }, + sampledAt=sampledAt, + ) + + speedAt20 = manager.series( + DOWNLOAD_SPEED_METRIC, + 20, + granularitySeconds=10, + now=20, + ) + usageAt20 = manager.series( + DOWNLOAD_USAGE_METRIC, + 20, + granularitySeconds=10, + now=20, + ) + speedAt21 = manager.series( + DOWNLOAD_SPEED_METRIC, + 20, + granularitySeconds=10, + now=21, + ) + + self.assertEqual( + tuple((point.sampledAt, point.value) for point in speedAt20), + ((9.0, 20.0), (19.0, 60.0), (20.0, 90.0)), + ) + self.assertEqual( + tuple((point.sampledAt, point.value) for point in usageAt20), + ((9.0, 3000.0), (19.0, 7000.0), (20.0, 9000.0)), + ) + self.assertEqual( + tuple((point.sampledAt, point.value) for point in speedAt21), + ((9.0, 20.0), (19.0, 60.0), (20.0, 90.0)), + ) + + def testAutoGranularityAndNonMonotonicInputArePredictable(self): + """Select documented buckets and normalize backward timestamps safely.""" + manager = MetricsDataManager(maximumHistorySeconds=24 * 60 * 60) + + self.assertEqual(manager.effectiveGranularity(5 * 60), 5) + self.assertEqual(manager.effectiveGranularity(15 * 60), 10) + self.assertEqual(manager.effectiveGranularity(60 * 60), 30) + self.assertEqual(manager.effectiveGranularity(24 * 60 * 60), 15 * 60) + self.assertEqual(manager.effectiveGranularity(60, 120), 60) + + manager.recordSample({UPLOAD_SPEED_METRIC: 10}, sampledAt=50) + manager.recordSample({UPLOAD_SPEED_METRIC: 20}, sampledAt=40) + + self.assertEqual( + tuple(sample.sampledAt for sample in manager.rawSamples()), + (50.0, 50.0), + ) + self.assertEqual( + tuple( + point.value + for point in manager.series( + UPLOAD_SPEED_METRIC, + 60, + granularitySeconds=10, + now=50, + ) + ), + (15.0,), + ) + + def testUsageHistoryCanClearWithoutMutatingSpeedHistory(self): + """Reset cumulative graphs while retaining immutable speed samples.""" + manager = MetricsDataManager() + manager.recordSample( + { + DOWNLOAD_SPEED_METRIC: 25, + DOWNLOAD_USAGE_METRIC: 100, + }, + sampledAt=10, + ) + manager.clearTrafficUsageHistory() + + samples = manager.rawSamples() + + self.assertEqual(len(samples), 1) + self.assertEqual(samples[0].values[DOWNLOAD_SPEED_METRIC], 25.0) + self.assertNotIn(DOWNLOAD_USAGE_METRIC, samples[0].values) + + +class MetricsPageAndGraphTest(unittest.TestCase): + """Exercise hidden-page rendering and graph lookup with real widgets.""" + + @classmethod + def setUpClass(cls): + """Create the process-wide headless QApplication.""" + application() + + def tearDown(self): + """Drain deferred widget destruction after every UI case.""" + collectAtBoundary() + + def testHiddenPageStoresHistoryWithoutRenderingThenCatchesUp(self): + """Keep collection eager and graph submission strictly visibility-bound.""" + manager = MetricsDataManager() + page = MetricsPage(manager) + sampledAt = math.floor((time.monotonic() - 20) / 10) * 10 + 1 + manager.recordSample({DOWNLOAD_SPEED_METRIC: 32}, sampledAt=sampledAt) + + processQtEvents() + + self.assertEqual(page._renderRevision, 0) + self.assertEqual(page.downloadSpeedGraph._points, tuple()) + self.assertFalse(page._renderTimer.isActive()) + self.assertFalse(page._timelineTimer.isActive()) + + page.show() + + self.assertTrue(waitFor(lambda: page._renderRevision == 1)) + self.assertEqual( + tuple(point.value for point in page.downloadSpeedGraph._points), + (32.0,), + ) + self.assertTrue(page._timelineTimer.isActive()) + + page.hide() + + processQtEvents() + + hiddenRevision = page._renderRevision + + manager.recordSample({DOWNLOAD_SPEED_METRIC: 64}, sampledAt=sampledAt + 1) + page._renderLatest() + + processQtEvents() + + self.assertEqual(page._renderRevision, hiddenRevision) + self.assertFalse(page._renderTimer.isActive()) + self.assertFalse(page._timelineTimer.isActive()) + + page.show() + + self.assertTrue(waitFor(lambda: page._renderRevision > hiddenRevision)) + self.assertEqual( + tuple(point.value for point in page.downloadSpeedGraph._points), + (48.0,), + ) + + page.close() + page.deleteLater() + manager.deleteLater() + + def testHoverLookupMatchesDisplayedPointAndTimestampRange(self): + """Report the exact prepared point rather than recomputing raw history.""" + graph = MetricsGraphWidget( + MetricsGraphWidget.Download, + formatTrafficSpeed, + ) + graph.resize(640, 360) + graph.setMetricLabel('Download Speed') + point = MetricPoint(95, 2048, 91, 95, 3) + graph.setSeries((point,), 20, 100, currentWallTime=1000) + graph.show() + + processQtEvents() + + position = graph._pointPosition( + point, + graph._chartRect(), + graph._maximumValue(), + ).toPoint() + + self.assertIs(graph._nearestPoint(position), point) + + tooltip = graph._tooltipText(point) + + self.assertIn('Download Speed: 2 KiB/s', tooltip) + self.assertIn(' – ', tooltip) + self.assertNotIn('×3', tooltip) + + graph.close() + graph.deleteLater() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_plugin_architecture.py b/tests/test_plugin_architecture.py index 748c1bf6..8894be39 100644 --- a/tests/test_plugin_architecture.py +++ b/tests/test_plugin_architecture.py @@ -41,7 +41,12 @@ from Furious.Service.SubscriptionImporter import ( SubscriptionSource, ) +from PySide6 import QtCore, QtWidgets + +from tests.support import application, collectAtBoundary, waitFor + import unittest +import weakref class FixtureConfiguration(ConfigFactory): @@ -363,5 +368,150 @@ class PluginRollbackTest(unittest.TestCase): registry.shutdown() +class PluginFailureIsolationTest(unittest.TestCase): + """Keep optional plugin failures bounded to the owning capability.""" + + @classmethod + def setUpClass(cls): + """Create Qt before exercising a plugin-provided editor widget.""" + application() + + def tearDown(self): + """Drain deferred deletes after plugin UI tests.""" + collectAtBoundary() + + def testDecoderExceptionFallsThroughToNextCandidate(self): + """Allow one broken decoder without blocking later valid decoders.""" + + class RaisingDecoder(SubscriptionDecoder): + decoderId = 'raising-decoder' + displayName = 'Raising decoder' + priority = 200 + + def decode(self, data: bytes): + raise RuntimeError('decoder fixture') + + class RaisingPlugin(FuriousPlugin): + metadata = PluginMetadata('tests.raising-decoder', 'Raising decoder') + capabilities = (RaisingDecoder(),) + + registry = PluginRegistry() + registry.register(RaisingPlugin()) + registry.register(FixturePlugin()) + + try: + result = registry.decodeSubscription(b'fixture') + + self.assertIsNotNone(result) + self.assertEqual(result.decoderId, 'fixture-decoder') + finally: + registry.shutdown() + + def testKernelFactoryExceptionReturnsControlledStartFailure(self): + """Translate construction failure into a false runtime result.""" + + class RaisingFactory(KernelFactory): + factoryId = 'raising-factory' + configurationTypes = (FixtureConfiguration,) + kernelTypes = (FixtureKernel,) + + def create(self, request: KernelRequest): + raise RuntimeError('factory fixture') + + class RaisingPlugin(FuriousPlugin): + metadata = PluginMetadata('tests.raising-factory', 'Raising factory') + capabilities = (RaisingFactory(),) + + registry = PluginRegistry() + registry.register(RaisingPlugin()) + + try: + kernel, success = registry.startKernel( + FixtureConfiguration({'type': 'fixture'}), + 'direct', + ) + + self.assertIsNone(kernel) + self.assertFalse(success) + finally: + registry.shutdown() + + def testShutdownExceptionDoesNotBlockOtherPlugins(self): + """Run every shutdown hook once even when one hook raises.""" + stopped = [] + + class OrderedPlugin(FuriousPlugin): + capabilities = tuple() + + def __init__(self, identifier: str, *, raises=False): + self.metadata = PluginMetadata(identifier, identifier) + self.raises = raises + + def shutdown(self): + stopped.append(self.metadata.id) + + if self.raises: + raise RuntimeError('shutdown fixture') + + registry = PluginRegistry() + registry.register(OrderedPlugin('tests.shutdown.first')) + registry.register(OrderedPlugin('tests.shutdown.second', raises=True)) + registry.register(OrderedPlugin('tests.shutdown.third')) + + registry.shutdown() + registry.shutdown() + + self.assertEqual( + stopped, + [ + 'tests.shutdown.third', + 'tests.shutdown.second', + 'tests.shutdown.first', + ], + ) + + def testRegistryStoresEditorFactoryButNotTransientEditorInstance(self): + """Destroy a plugin editor while its provider remains registered.""" + + class WidgetEditorProvider(ProtocolEditorProvider): + editorId = 'fixture.widget-editor' + protocolIds = ('FIXTURE',) + + def createEditor(self, protocolId: str, parent=None, **kwargs): + editor = QtWidgets.QDialog(parent) + editor.setAttribute(QtCore.Qt.WidgetAttribute.WA_DeleteOnClose, True) + + return editor + + class WidgetPlugin(FuriousPlugin): + metadata = PluginMetadata('tests.widget-editor', 'Widget editor') + capabilities = ( + FixtureProtocolHandler(), + WidgetEditorProvider(), + ) + + registry = PluginRegistry() + registry.register(WidgetPlugin()) + + try: + editor = registry.createEditorForConfig( + FixtureConfiguration({'type': 'fixture', 'value': 'node'}) + ) + editorReference = weakref.ref(editor) + editor.show() + editor.close() + editor = None + + self.assertTrue(waitFor(lambda: editorReference() is None)) + self.assertIsNotNone( + registry.capability( + CapabilityKind.ProtocolEditor, + 'fixture.widget-editor', + ) + ) + finally: + registry.shutdown() + + if __name__ == '__main__': unittest.main() diff --git a/tests/test_process_stress.py b/tests/test_process_stress.py new file mode 100644 index 00000000..711dcaaf --- /dev/null +++ b/tests/test_process_stress.py @@ -0,0 +1,106 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# 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 . + +"""Stress the harmless Python-backed external-core lifecycle.""" + +from __future__ import annotations + +from Furious.Backends.ExternalCore import ConfigExternalCore, ExternalCoreProcess + +from tests.support import currentNativeHandleCount, currentRSS + +from pathlib import Path + +import sys +import tempfile +import threading +import unittest + + +class ExternalProcessStressTest(unittest.TestCase): + """Reject subprocess, pipe, and reader-thread growth across many restarts.""" + + def testTwentyFourStartStopCyclesReleaseNativeResources(self): + """Reap the exact child and return reader/watcher ownership every cycle.""" + baselineThreads = {thread.ident for thread in threading.enumerate()} + baselineHandles = currentNativeHandleCount() + samples = [] + + with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory: + config = ConfigExternalCore( + { + 'type': 'external-core', + 'executable': str(Path(sys.executable).resolve()), + 'workingDirectory': directory, + 'arguments': ['-u', '-c', 'import time; time.sleep(60)'], + 'environment': {}, + 'httpProxy': '127.0.0.1:10809', + 'socksProxy': '127.0.0.1:10808', + 'shutdownTimeout': 1, + } + ) + runtime = ExternalCoreProcess() + + try: + for index in range(24): + self.assertTrue(runtime.start(config)) + + process = runtime.process + + self.assertIsNotNone(process) + self.assertTrue(runtime.isAlive()) + + runtime.stop() + + self.assertFalse(runtime.isAlive()) + self.assertIsNone(runtime.process) + self.assertFalse(runtime._readerThreads) + self.assertIsNone(runtime._watcherThread) + self.assertIsNotNone(process.poll()) + + if index % 6 == 5: + samples.append( + { + 'cycle': index + 1, + 'handles': currentNativeHandleCount(), + 'rss': currentRSS(), + 'threads': len(threading.enumerate()), + } + ) + finally: + runtime.dispose() + + remainingThreads = { + thread.ident + for thread in threading.enumerate() + if thread.ident not in baselineThreads + } + + self.assertEqual(remainingThreads, set()) + + handles = tuple( + sample['handles'] for sample in samples if sample['handles'] is not None + ) + + if baselineHandles is not None and len(handles) == len(samples): + self.assertLessEqual(max((*handles, baselineHandles)) - baselineHandles, 4) + + print('External process stress:', {'cycles': 24, 'samples': samples}) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_qt_lifetime.py b/tests/test_qt_lifetime.py index 040ada3f..b185ad01 100644 --- a/tests/test_qt_lifetime.py +++ b/tests/test_qt_lifetime.py @@ -20,6 +20,8 @@ from __future__ import annotations from Furious.Backends.ExternalCore.Editor import ExternalCoreEditor +from Furious.Backends.Hysteria1.Editor import Hysteria1Editor +from Furious.Backends.Hysteria2.Editor import Hysteria2Editor from Furious.Backends.Hysteria2.TunSettingsDialog import Hysteria2TunSettingsDialog from Furious.Backends.Xray.RoutingWindow import ( RoutingPreviewDialog, @@ -27,7 +29,11 @@ from Furious.Backends.Xray.RoutingWindow import ( RoutingRulesDialog, ) from Furious.Backends.Xray.SocksEditor import SocksEditor +from Furious.Backends.Xray.ShadowsocksEditor import ShadowsocksEditor +from Furious.Backends.Xray.TrojanEditor import TrojanEditor from Furious.Backends.Xray.TunSettingsDialog import XrayTunSettingsDialog +from Furious.Backends.Xray.VlessEditor import VlessEditor +from Furious.Backends.Xray.VmessEditor import VmessEditor from Furious.Frozenlib import Mixins from Furious.Qt import ( AppQAction, @@ -284,6 +290,12 @@ class QtLifetimeTest(unittest.TestCase): factories = ( ('external-core', ExternalCoreEditor, 35), ('socks-protocol', SocksEditor, 35), + ('vless-protocol', VlessEditor, 12), + ('vmess-protocol', VmessEditor, 12), + ('trojan-protocol', TrojanEditor, 12), + ('shadowsocks-protocol', ShadowsocksEditor, 12), + ('hysteria1-protocol', Hysteria1Editor, 12), + ('hysteria2-protocol', Hysteria2Editor, 12), ('xray-tun-settings', XrayTunSettingsDialog, 25), ('hysteria2-tun-settings', Hysteria2TunSettingsDialog, 25), ( diff --git a/tests/test_qt_stress.py b/tests/test_qt_stress.py index 1365889c..92edf5f5 100644 --- a/tests/test_qt_stress.py +++ b/tests/test_qt_stress.py @@ -21,13 +21,18 @@ from __future__ import annotations from Furious.Frozenlib import Mixins from Furious.Qt import AppQAction, AppQMenu, AppQTransientDialog +from Furious.Backends.Hysteria2.Editor import Hysteria2Editor +from Furious.Backends.Xray.RoutingWindow import RoutingPreviewDialog +from Furious.Backends.Xray.VlessEditor import VlessEditor -from PySide6 import QtCore +from PySide6 import QtCore, QtGui from tests.support import ( application, collectAtBoundary, currentRSS, + currentNativeHandleCount, + isolatedSettings, qObjectCount, ) @@ -209,6 +214,106 @@ class QtMemoryStressTest(unittest.TestCase): finally: tracemalloc.stop() + def testRealDialogFamiliesPlateauAcrossBatches(self): + """Measure representative production editors instead of only a probe.""" + factories = ( + ('routing-preview', lambda: RoutingPreviewDialog({'rules': []})), + ('vless-editor', VlessEditor), + ('hysteria2-editor', Hysteria2Editor), + ) + warmupIterations, batchIterations, batchCount = 3, 8, 3 + + with isolatedSettings(): + for family, factory in factories: + with self.subTest(family=family): + collectAtBoundary() + + baseline = { + 'timer': qObjectCount(QtCore.QTimer), + 'action': qObjectCount(QtGui.QAction), + 'menu': qObjectCount(AppQMenu), + 'translationPool': len(Mixins.QTranslatable.ObjectsPool), + 'themePool': len(Mixins.ThemeAware.ObjectsPool), + 'connectionPool': len(Mixins.ConnectionAware.ObjectsPool), + } + + references, destroyed, samples = [], [], [] + + for batch, iterations in enumerate( + (warmupIterations,) + (batchIterations,) * batchCount + ): + for _index in range(iterations): + dialog = factory() + dialog.destroyed.connect( + lambda *_args, _destroyed=destroyed: _destroyed.append( + True + ) + ) + + references.append(weakref.ref(dialog)) + + dialog.show() + dialog.close() + + del dialog + + collectAtBoundary() + + samples.append( + { + 'batch': batch, + 'timer': qObjectCount(QtCore.QTimer), + 'action': qObjectCount(QtGui.QAction), + 'menu': qObjectCount(AppQMenu), + 'handles': currentNativeHandleCount(), + 'rss': currentRSS(), + } + ) + + total = warmupIterations + batchIterations * batchCount + + self.assertEqual(len(destroyed), total) + self.assertTrue( + all(reference() is None for reference in references) + ) + + for sample in samples: + self.assertEqual(sample['timer'], baseline['timer']) + self.assertEqual(sample['action'], baseline['action']) + self.assertEqual(sample['menu'], baseline['menu']) + + self.assertEqual( + len(Mixins.QTranslatable.ObjectsPool), + baseline['translationPool'], + ) + self.assertEqual( + len(Mixins.ThemeAware.ObjectsPool), + baseline['themePool'], + ) + self.assertEqual( + len(Mixins.ConnectionAware.ObjectsPool), + baseline['connectionPool'], + ) + + handles = tuple( + sample['handles'] + for sample in samples + if sample['handles'] is not None + ) + + if len(handles) == len(samples): + self.assertLessEqual(max(handles) - min(handles), 4) + + print( + 'Qt real-dialog stress:', + { + 'family': family, + 'iterations': total, + 'destroyed': len(destroyed), + 'samples': samples, + }, + ) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_shadowsocks_uri.py b/tests/test_shadowsocks_uri.py new file mode 100644 index 00000000..4e81ee6a --- /dev/null +++ b/tests/test_shadowsocks_uri.py @@ -0,0 +1,138 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# 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 . + +"""Exercise SIP002 canonical and generated Shadowsocks URI boundaries.""" + +from __future__ import annotations + +from Furious.Backends.ShadowsocksURI import ( + ShadowsocksURIData, + ShadowsocksURIError, + formatPluginArgument, + parseShadowsocksURI, + serializeShadowsocksURI, +) + +import random +import string +import unittest + + +class ShadowsocksURITest(unittest.TestCase): + """Protect round trips, encoding rules, IPv6, plugins, and rejection.""" + + def testCanonicalRoundTripsReservedUnicodeAndIPv6Fields(self): + """Preserve meaningful fields while canonicalizing SIP002 syntax.""" + plugin = formatPluginArgument( + 'simple-obfs', + ( + ('obfs', 'http'), + ('obfs-host', '例子.example'), + ('escaped', r'a:b;c=d\e'), + ), + ) + cases = ( + ShadowsocksURIData( + 'aes-128-gcm', + ' +/=: @% 密码 ', + 'example.com', + 443, + '', + 'Tag with spaces 测试', + ), + ShadowsocksURIData( + 'chacha20-ietf-poly1305', + 'secret', + '192.0.2.1', + 8388, + plugin, + 'Plugin profile', + ), + ShadowsocksURIData( + '2022-blake3-aes-128-gcm', + 'plain+/=:password', + '2001:db8::42', + 8443, + '', + 'IPv6', + ), + ) + + for value in cases: + with self.subTest(value=value): + uri = serializeShadowsocksURI(value) + + self.assertNotIn(' ', uri) + self.assertEqual(parseShadowsocksURI(uri), value) + + if ':' in value.host: + self.assertIn(f'@[{value.host}]:{value.port}', uri) + + def testUnknownQueryParametersAreIgnored(self): + """Accept future SIP002 query keys without changing known meaning.""" + base = serializeShadowsocksURI( + ShadowsocksURIData('aes-256-gcm', 'secret', 'example.com', 443) + ) + parsed = parseShadowsocksURI(f'{base}/?future=value&another=1') + + self.assertEqual(parsed.method, 'aes-256-gcm') + self.assertEqual(parsed.password, 'secret') + self.assertEqual(parsed.plugin, '') + + def testDeterministicGeneratedRoundTrips(self): + """Explore many safe strings without an optional property-test package.""" + randomizer = random.Random(20260817) + alphabet = string.ascii_letters + string.digits + '+/=: @%_-' + + for index in range(75): + password = ''.join( + randomizer.choice(alphabet) for _character in range(1 + index % 19) + ) + value = ShadowsocksURIData( + ('2022-blake3-aes-256-gcm' if index % 5 == 0 else 'aes-256-gcm'), + password, + f'node-{index}.example', + 1024 + index, + '', + f'Generated {index}', + ) + + self.assertEqual( + parseShadowsocksURI(serializeShadowsocksURI(value)), + value, + ) + + def testMalformedUrisFailWithoutPartialResult(self): + """Reject corrupt authority, encoding, cipher, credentials, and ports.""" + malformed = ( + 'http://example.com', + 'ss://not-base64@example.com:443', + 'ss://YWVzLTEyOC1nY206@example.com:443', + 'ss://YWVzLTEyOC1nY206c2VjcmV0@example.com', + 'ss://YWVzLTEyOC1nY206c2VjcmV0@example.com:0', + 'ss://YWVzLTEyOC1nY206c2VjcmV0@example.com:70000', + 'ss://unsupported%3Asecret@example.com:443', + 'ss://2022-blake3-aes-128-gcm:bad%ZZ@example.com:443', + ) + + for uri in malformed: + with self.subTest(uri=uri), self.assertRaises(ShadowsocksURIError): + parseShadowsocksURI(uri) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_subscription_sync.py b/tests/test_subscription_sync.py new file mode 100644 index 00000000..a6f4918c --- /dev/null +++ b/tests/test_subscription_sync.py @@ -0,0 +1,142 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# 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 . + +"""Protect isolated subscription-group synchronization semantics.""" + +from __future__ import annotations + +from Furious.Models import ConfigFactory, ServerProfile +from Furious.Service.SubscriptionSync import SubscriptionSynchronizer + +import unittest + + +def profile( + name: str, + address: str, + *, + source: str = '', + managed: bool = False, + key: str = '', + favorite: bool = False, +): + """Create one core-neutral profile with explicit ownership metadata.""" + return ServerProfile.fromConfiguration( + ConfigFactory({'type': 'fixture', 'address': address, 'port': 443}), + { + 'displayName': name, + 'subscriptionSource': source, + 'subscriptionManaged': managed, + 'subscriptionProfileKey': key, + 'favorite': favorite, + }, + ) + + +class SubscriptionSynchronizerTest(unittest.TestCase): + """Verify one subscription can never mutate unrelated or manual profiles.""" + + def testGroupUpdateIsAtomicAndPreservesLocalIdentityMetadata(self): + """Update, add, and remove only the selected managed group.""" + manual = profile('Manual', 'manual.example') + retained = profile( + 'Local label', + 'old.example', + source='group-a', + managed=True, + key='upstream:one', + favorite=True, + ) + removed = profile( + 'Removed', + 'removed.example', + source='group-a', + managed=True, + key='upstream:removed', + ) + other = profile( + 'Other group', + 'other.example', + source='group-b', + managed=True, + key='upstream:other', + ) + originalId = retained.metadata.profileId + profiles = [manual, retained, removed, other] + incoming = [ + profile('Remote label', 'new.example', key='upstream:one'), + profile('Added', 'added.example', key='upstream:two'), + ] + + result = SubscriptionSynchronizer().reconcile( + profiles, + incoming, + 'group-a', + ) + + self.assertEqual((result.added, result.updated, result.removed), (1, 1, 1)) + self.assertIs(profiles[0], manual) + self.assertIs(profiles[1], retained) + self.assertIs(profiles[-1], other) + self.assertEqual(retained.metadata.profileId, originalId) + self.assertEqual(retained.itemRemark, 'Remote label') + self.assertTrue(retained.metadata.favorite) + self.assertEqual(retained.connection['address'], 'new.example') + self.assertTrue(removed.deleted) + self.assertEqual( + tuple(item.index for item in profiles), + tuple(range(len(profiles))), + ) + self.assertEqual(manual.itemSubscription, '') + self.assertFalse(manual.itemSubscriptionManaged) + self.assertEqual(other.connection['address'], 'other.example') + + def testLegacyKeysAndDuplicateConnectionsAreDeterministic(self): + """Assign stable occurrence keys without relying on row position alone.""" + first = profile('First', 'same.example', source='group', managed=True) + second = profile('Second', 'same.example', source='group', managed=True) + profiles = [first, second] + + SubscriptionSynchronizer._ensureKeys(profiles, 'group') + + firstKey = first.metadata.subscriptionProfileKey + secondKey = second.metadata.subscriptionProfileKey + + self.assertTrue(firstKey.startswith('config:')) + self.assertEqual(secondKey, f'{firstKey}#2') + + SubscriptionSynchronizer._ensureKeys(profiles, 'group') + + self.assertEqual(first.metadata.subscriptionProfileKey, firstKey) + self.assertEqual(second.metadata.subscriptionProfileKey, secondKey) + + def testEmptyGroupFailsBeforeMutatingInput(self): + """Reject ambiguous ownership without partially rewriting profiles.""" + existing = profile('Manual', 'manual.example') + incoming = profile('Incoming', 'incoming.example') + profiles = [existing] + + with self.assertRaisesRegex(ValueError, 'group ID'): + SubscriptionSynchronizer().reconcile(profiles, [incoming], '') + + self.assertEqual(profiles, [existing]) + self.assertEqual(incoming.itemSubscription, '') + self.assertFalse(incoming.itemSubscriptionManaged) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_ui_behavior.py b/tests/test_ui_behavior.py index 24f5e865..9a7dee26 100644 --- a/tests/test_ui_behavior.py +++ b/tests/test_ui_behavior.py @@ -25,6 +25,11 @@ from Furious.Backends.ExternalCore.Configuration import ( ) from Furious.Backends.ExternalCore.Editor import ExternalCoreEditor from Furious.Backends.Xray.RoutingWindow import RoutingRulesDialog +from Furious.Actions.Connection import ConnectAction +from Furious.Controllers.ConnectionController import ( + ConnectionController, + ConnectionState, +) from Furious.Models import ProfileMetadata, ServerProfile from Furious.Qt import AppQMessageBox from Furious.Service import ( @@ -34,6 +39,10 @@ from Furious.Service import ( ) from Furious.Window.LogPage import LogPage from Furious.Window.SubscriptionPage import _SubscriptionEditorDialog +from Furious.Widget.ConnectionButton import ConnectionButton + +from PySide6 import QtCore +from PySide6.QtTest import QTest from tests.support import ( application, @@ -45,6 +54,8 @@ from tests.support import ( import copy import unittest +from unittest import mock + class EditorMappingTest(unittest.TestCase): """Verify editor fields preserve structured configuration semantics.""" @@ -260,6 +271,167 @@ class DialogBehaviorTest(unittest.TestCase): self.assertEqual(finished, [int(AppQMessageBox.StandardButton.Yes)]) self.assertIs(messageBox.clickedButton(), yesButton) + def testMessageBoxButtonsHaveAdaptiveFluentLayoutAndRoles(self): + """Keep one, two, and three actions slim, separated, and content-driven.""" + configurations = ( + (AppQMessageBox.StandardButton.Ok, 1), + ( + AppQMessageBox.StandardButton.Ok | AppQMessageBox.StandardButton.Cancel, + 2, + ), + ( + AppQMessageBox.StandardButton.Save + | AppQMessageBox.StandardButton.Discard + | AppQMessageBox.StandardButton.Cancel, + 3, + ), + ) + widths = [] + + for buttons, count in configurations: + with self.subTest(buttonCount=count): + messageBox = AppQMessageBox( + icon=AppQMessageBox.Icon.Question, + text='Ready', + buttons=buttons, + ) + messageBox.show() + + processQtEvents() + + actionButtons = messageBox.buttons() + + self.assertEqual(len(actionButtons), count) + self.assertTrue( + all( + button.width() > button.height() * 2 for button in actionButtons + ) + ) + + geometries = sorted( + (button.geometry() for button in actionButtons), + key=lambda geometry: geometry.x(), + ) + + for left, right in zip(geometries, geometries[1:]): + self.assertGreaterEqual( + right.left() - left.right() - 1, + messageBox.ButtonSpacing, + ) + + if count > 1: + self.assertLessEqual( + max(button.width() for button in actionButtons) + - min(button.width() for button in actionButtons), + 1, + ) + + if buttons & AppQMessageBox.StandardButton.Discard: + discard = messageBox.button(AppQMessageBox.StandardButton.Discard) + self.assertEqual( + discard.property('messageBoxRole'), + 'destructive', + ) + + widths.append(messageBox.width()) + + messageBox.close() + + self.assertLess(widths[0], widths[2]) + + def testMessageBoxEscapeUsesConfiguredCancelResult(self): + """Preserve Escape semantics without leaving masks or open-box owners.""" + messageBox = AppQMessageBox( + text='Continue?', + buttons=( + AppQMessageBox.StandardButton.Yes | AppQMessageBox.StandardButton.Cancel + ), + ) + + finished = [] + + messageBox.finished.connect(finished.append) + messageBox.setEscapeButton(AppQMessageBox.StandardButton.Cancel) + messageBox.show() + + QTest.keyClick(messageBox, QtCore.Qt.Key.Key_Escape) + + processQtEvents() + + self.assertEqual( + finished, + [int(AppQMessageBox.StandardButton.Cancel)], + ) + self.assertEqual(AppQMessageBox._openMessageBoxes, {}) + + +class SharedConnectionPresentationTest(unittest.TestCase): + """Keep Home and tray adapters synchronized to one controller state.""" + + @classmethod + def setUpClass(cls): + application() + + def tearDown(self): + collectAtBoundary() + + def testHomeSelectionPolicyAndTrayPresentationShareController(self): + """Apply selection only to Home while lifecycle text remains identical.""" + controller = ConnectionController() + activation = mock.Mock(return_value=True) + + with ( + mock.patch( + 'Furious.Widget.ConnectionButton.AppConnectionController', + return_value=controller, + ), + mock.patch( + 'Furious.Actions.Connection.AppConnectionController', + return_value=controller, + ), + mock.patch.object(controller, 'toggle', return_value=True) as toggle, + ): + home = ConnectionButton(activation) + tray = ConnectAction() + + home.setSelectionCount(0) + + self.assertTrue(home.isEnabled()) + + home.click() + toggle.assert_not_called() + + home.setSelectionCount(2) + + self.assertFalse(home.isEnabled()) + + home.setSelectionCount(1) + + self.assertTrue(home.isEnabled()) + + home.click() + activation.assert_called_once() + toggle.assert_called_once() + + for state, enabled in ( + (ConnectionState.Connecting, False), + (ConnectionState.Connected, True), + (ConnectionState.Disconnecting, False), + (ConnectionState.Disconnected, True), + ): + controller._setState(state) + + processQtEvents() + + self.assertEqual(home.text(), tray.text()) + self.assertEqual(home.isEnabled(), enabled) + self.assertEqual(tray.isEnabled(), enabled) + + tray.deleteLater() + home.deleteLater() + + controller.deleteLater() + if __name__ == '__main__': unittest.main()